| """Test result object""" |
| |
| import os |
| import sys |
| import traceback |
| |
| from StringIO import StringIO |
| |
| from . import util |
| from functools import wraps |
| |
| __unittest = True |
| |
| def failfast(method): |
| @wraps(method) |
| def inner(self, *args, **kw): |
| if getattr(self, 'failfast', False): |
| self.stop() |
| return method(self, *args, **kw) |
| return inner |
| |
| STDOUT_LINE = '\nStdout:\n%s' |
| STDERR_LINE = '\nStderr:\n%s' |
| |
| |
| class TestResult(object): |
| """Holder for test result information. |
| |
| Test results are automatically managed by the TestCase and TestSuite |
| classes, and do not need to be explicitly manipulated by writers of tests. |
| |
| Each instance holds the total number of tests run, and collections of |
| failures and errors that occurred among those test runs. The collections |
| contain tuples of (testcase, exceptioninfo), where exceptioninfo is the |
| formatted traceback of the error that occurred. |
| """ |
| _previousTestClass = None |
| _testRunEntered = False |
| _moduleSetUpFailed = False |
| def __init__(self, stream=None, descriptions=None, verbosity=None): |
| self.failfast = False |
| self.failures = [] |
| self.errors = [] |
| self.testsRun = 0 |
| self.skipped = [] |
| self.expectedFailures = [] |
| self.unexpectedSuccesses = [] |
| self.shouldStop = False |
| self.buffer = False |
| self._stdout_buffer = None |
| self._stderr_buffer = None |
| self._original_stdout = sys.stdout |
| self._original_stderr = sys.stderr |
| self._mirrorOutput = False |
| |
| def printErrors(self): |
| "Called by TestRunner after test run" |
| |
| def startTest(self, test): |
| "Called when the given test is about to be run" |
| self.testsRun += 1 |
| self._mirrorOutput = False |
| self._setupStdout() |
| |
| def _setupStdout(self): |
| if self.buffer: |
| if self._stderr_buffer is None: |
| self._stderr_buffer = StringIO() |
| self._stdout_buffer = StringIO() |
| sys.stdout = self._stdout_buffer |
| sys.stderr = self._stderr_buffer |
| |
| def startTestRun(self): |
| """Called once before any tests are executed. |
| |
| See startTest for a method called before each test. |
| """ |
| |
| def stopTest(self, test): |
| """Called when the given test has been run""" |
| self._restoreStdout() |
| self._mirrorOutput = False |
| |
| def _restoreStdout(self): |
| if self.buffer: |
| if self._mirrorOutput: |
| output = sys.stdout.getvalue() |
| error = sys.stderr.getvalue() |
| if output: |
| if not output.endswith('\n'): |
| output += '\n' |
| self._original_stdout.write(STDOUT_LINE % output) |
| if error: |
| if not error.endswith('\n'): |
| error += '\n' |
| self._original_stderr.write(STDERR_LINE % error) |
| |
| sys.stdout = self._original_stdout |
| sys.stderr = self._original_stderr |
| self._stdout_buffer.seek(0) |
| self._stdout_buffer.truncate() |
| self._stderr_buffer.seek(0) |
| self._stderr_buffer.truncate() |
| |
| def stopTestRun(self): |
| """Called once after all tests are executed. |
| |
| See stopTest for a method called after each test. |
| """ |
| |
| @failfast |
| def addError(self, test, err): |
| """Called when an error has occurred. 'err' is a tuple of values as |
| returned by sys.exc_info(). |
| """ |
| self.errors.append((test, self._exc_info_to_string(err, test))) |
| self._mirrorOutput = True |
| |
| @failfast |
| def addFailure(self, test, err): |
| """Called when an error has occurred. 'err' is a tuple of values as |
| returned by sys.exc_info().""" |
| self.failures.append((test, self._exc_info_to_string(err, test))) |
| self._mirrorOutput = True |
| |
| def addSuccess(self, test): |
| "Called when a test has completed successfully" |
| pass |
| |
| def addSkip(self, test, reason): |
| """Called when a test is skipped.""" |
| self.skipped.append((test, reason)) |
| |
| def addExpectedFailure(self, test, err): |
| """Called when an expected failure/error occured.""" |
| self.expectedFailures.append( |
| (test, self._exc_info_to_string(err, test))) |
| |
| @failfast |
| def addUnexpectedSuccess(self, test): |
| """Called when a test was expected to fail, but succeed.""" |
| self.unexpectedSuccesses.append(test) |
| |
| def wasSuccessful(self): |
| "Tells whether or not this result was a success" |
| return len(self.failures) == len(self.errors) == 0 |
| |
| def stop(self): |
| "Indicates that the tests should be aborted" |
| self.shouldStop = True |
| |
| def _exc_info_to_string(self, err, test): |
| """Converts a sys.exc_info()-style tuple of values into a string.""" |
| exctype, value, tb = err |
| # Skip test runner traceback levels |
| while tb and self._is_relevant_tb_level(tb): |
| tb = tb.tb_next |
| |
| if exctype is test.failureException: |
| # Skip assert*() traceback levels |
| length = self._count_relevant_tb_levels(tb) |
| msgLines = traceback.format_exception(exctype, value, tb, length) |
| else: |
| msgLines = traceback.format_exception(exctype, value, tb) |
| |
| if self.buffer: |
| output = sys.stdout.getvalue() |
| error = sys.stderr.getvalue() |
| if output: |
| if not output.endswith('\n'): |
| output += '\n' |
| msgLines.append(STDOUT_LINE % output) |
| if error: |
| if not error.endswith('\n'): |
| error += '\n' |
| msgLines.append(STDERR_LINE % error) |
| return ''.join(msgLines) |
| |
| |
| def _is_relevant_tb_level(self, tb): |
| return '__unittest' in tb.tb_frame.f_globals |
| |
| def _count_relevant_tb_levels(self, tb): |
| length = 0 |
| while tb and not self._is_relevant_tb_level(tb): |
| length += 1 |
| tb = tb.tb_next |
| return length |
| |
| def __repr__(self): |
| return ("<%s run=%i errors=%i failures=%i>" % |
| (util.strclass(self.__class__), self.testsRun, len(self.errors), |
| len(self.failures))) |
