Michael Foord avatar Michael Foord committed 26ac222

stopTest event now has much more information available
TestResult has new addReport method using StopTestEvent instances
Test reporting uses the new reports instead of old methods
_WritelnDecorator has a write method so that writes go through the message API
TextTestResult no longer needs the copious add* methods

Comments (0)

Files changed (5)

 Should there be an "excluded-plugins" section in the config files, so projects
 can prevent incompatible plugins from being loaded?
 
+The junit-xml plugin needs access to stdout / stderr for the test. This is
+only currently available if the test is run with buffering on.
+
+Unrelated to plugins, but expectedFailure decorator should probably
+optionally allow a reason like skips do.
+
+If TestCase.run is called without a result object then it calls
+result.startTest. Should it also fire the startTestRun and stopTestRun events?
+
 Should multiline values in config files be merged instead of overriding each
 other. (This is effectively what happens for the plugins list.)
 

unittest2/case.py

     def __repr__(self):
         return "<%s testMethod=%s>" % \
                (strclass(self.__class__), self._testMethodName)
-    
-    def _addSkip(self, result, reason):
-        addSkip = getattr(result, 'addSkip', None)
-        if addSkip is not None:
-            addSkip(self, reason)
-        else:
-            warnings.warn("Use of a TestResult out an addSkip method is deprecated", 
-                          DeprecationWarning, 2)
-            result.addSuccess(self)
+
 
     def withTestFailEvent(self, func, result, when):
         try:
                 return
             raise info[0], info[1], info[2]
 
+    def _createReport(self, exc_info, stage, outcome, result):
+        startTime = self._startTime
+        stopTime = time.time()
+        timeTaken = stopTime - startTime
+
+        traceback = None
+        
+        if outcome in ('failed', 'error'):
+            traceback = self.formatTraceback(exc_info)
+        
+        event = StopTestEvent(self, result, stopTime, timeTaken, outcome,
+                              exc_info, stage, traceback)
+
+        # first event allows customisation of the report
+        hooks.createReport(event)
+        hooks.stopTest(event)
+
+        if hasattr(result, 'addReport'):
+            result.addReport(event)
+        else:
+            # legacy result objects
+            if event.error:
+                result.addError(self, exc_info)
+            elif event.failed:
+                result.addFailure(self, exc_info)
+            elif event.passed:
+                result.addSuccess(self)
+            elif event.skipped:
+                addSkip = getattr(result, 'addSkip', None)
+                if addSkip is not None:
+                    addSkip(self, event.skipReason)
+                else:
+                    # legacy-legacy result objects
+                    warnings.warn("Use of a TestResult without an addSkip method is deprecated", 
+                                  DeprecationWarning, 2)
+                    result.addSuccess(self)
+            elif event.expectedFailure:
+                addExpectedFailure = getattr(result, 'addExpectedFailure', None)
+                if addExpectedFailure is not None:
+                    addExpectedFailure(self, exc_info)
+                else:
+                    warnings.warn("TestResult has no addExpectedFailure method, reporting as passes",
+                                  DeprecationWarning)
+                result.addSuccess(self)
+            elif event.unexpectedSuccess:
+                addUnexpectedSuccess = getattr(result, 'addUnexpectedSuccess', None)
+                if addUnexpectedSuccess is not None:
+                    addUnexpectedSuccess(self)
+                else:
+                    warnings.warn("TestResult has no addUnexpectedSuccess method, reporting as failures",
+                                  DeprecationWarning)
+                    result.addFailure(self, exc_info)
+        
+
     def run(self, result=None):
         orig_result = result
         if result is None:
         
         testMethod = getattr(self, self._testMethodName)
         
-        def doStop(outcome, exc_info, stage):
-            stopTime = time.time()
-            timeTaken = stopTime - startTime
-            event = StopTestEvent(self, result, stopTime, timeTaken, 
-                                  outcome, exc_info, stage)
-            hooks.stopTest(event)
-        
         if (getattr(self.__class__, "__unittest_skip__", False) or 
             getattr(testMethod, "__unittest_skip__", False)):
             # If the class or method was skipped.
             try:
                 skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                             or getattr(testMethod, '__unittest_skip_why__', ''))
-                self._addSkip(result, skip_why)
             finally:
+                exc_info = (SkipTest, SkipTest(skip_why), None)
+                self._createReport(exc_info, 'setUp', 'skipped', result)
                 result.stopTest(self)
-                exc_info = (SkipTest, SkipTest(skip_why), None)
-                doStop('skipped', exc_info, 'class')
             return
+
+        sys.exc_clear()
         try:
-            success = False
+            success = True
 
             def fireAfterSetUp():
                 exc_info = sys.exc_info()
 
             try:
                 self.withTestFailEvent(self.setUp, result, 'setUp')
-            except SkipTest, e:
-                self._addSkip(result, str(e))
+            except Exception, e:
+                success = False
                 exc_info = fireAfterSetUp()
-                doStop('skipped', exc_info, 'setUp')
-            except Exception:
-                exc_info = fireAfterSetUp()
-                doStop('error', exc_info, 'setUp')
-                result.addError(self, exc_info)
-            else:
+                if isinstance(e, SkipTest):
+                    outcome = 'skipped'
+                else:
+                    outcome = 'error'
+                self._createReport(exc_info, 'setUp', outcome, result)
+
+            if success:
                 fireAfterSetUp()
+                exc_info = None
     
                 try:
                     self.withTestFailEvent(testMethod, result, 'call')
-                except self.failureException:
+                except Exception, e:
+                    success = False
                     exc_info = sys.exc_info()
-                    doStop('failed', exc_info, 'call')
-                    result.addFailure(self, exc_info)
-                except _ExpectedFailure, e:
-                    exc_info = e.exc_info
-                    doStop('expectedFailure', exc_info, 'call')
-                    addExpectedFailure = getattr(result, 'addExpectedFailure', None)
-                    if addExpectedFailure is not None:
-                        addExpectedFailure(self, exc_info)
-                    else: 
-                        warnings.warn("Use of a TestResult without an addExpectedFailure method is deprecated", 
-                                      DeprecationWarning)
-                        result.addSuccess(self)
-                except _UnexpectedSuccess:
-                    exc_info = sys.exc_info()
-                    doStop('unexpectedSuccess', exc_info, 'call')
-                    addUnexpectedSuccess = getattr(result, 'addUnexpectedSuccess', None)
-                    if addUnexpectedSuccess is not None:
-                        addUnexpectedSuccess(self)
+                    if isinstance(e, self.failureException):
+                        outcome = 'failed'
+                    elif isinstance(e, _ExpectedFailure):
+                        outcome = 'expectedFailure'
+                    elif isinstance(e, _UnexpectedSuccess):
+                        outcome = 'unexpectedSuccess'
+                    elif isinstance(e, SkipTest):
+                        outcome = 'skipped'
                     else:
-                        warnings.warn("Use of a TestResult without an addUnexpectedSuccess method is deprecated", 
-                                      DeprecationWarning)
-                        result.addFailure(self, sys.exc_info())
-                except SkipTest, e:
-                    doStop('skipped', sys.exc_info(), 'call')
-                    self._addSkip(result, str(e))
-                except Exception:
-                    exc_info = sys.exc_info()
-                    doStop('error', exc_info, 'call')
-                    result.addError(self, exc_info)
-                else:
-                    success = True
+                        outcome = 'error'
+                    self._createReport(exc_info, 'call', outcome, result)
 
                 event = BeforeTearDownEvent(self, result, success, time.time())
                 hooks.beforeTearDown(event)
                 try:
                     self.withTestFailEvent(self.tearDown, result, 'tearDown')
                 except Exception:
+                    success = False
                     exc_info = sys.exc_info()
-                    doStop('error', exc_info, 'tearDown')
-                    result.addError(self, exc_info)
-                    success = False
+                    self._createReport(exc_info, 'tearDown', 'error', result)
 
             cleanUpSuccess = self.doCleanups()
             success = success and cleanUpSuccess
             if success:
-                doStop('passed', None, None)
-                result.addSuccess(self)
+                self._createReport(None, 'call', 'passed', result)
+
         finally:
             result.stopTest(self)
             if orig_result is None:
                 cleanUp = lambda: function(*args, **kwargs)
                 self.withTestFailEvent(cleanUp, result, 'cleanUp')
             except Exception:
+                ok = False
                 exc_info = sys.exc_info()
-                stopTime = time.time()
-                timeTaken = stopTime - self._startTime
-                event = StopTestEvent(self, result, stopTime, timeTaken, 
-                                      'error', exc_info, 'cleanup')
-                hooks.stopTest(event)
-                ok = False
-                result.addError(self, exc_info)
+                self._createReport(exc_info, 'cleanUp', 'error', result)
         return ok
 
     def __call__(self, *args, **kwds):

unittest2/events.py

         self.success = success
         self.time = time
 
+_DEFAULT_RESULTS = {
+    'passed': ('ok', '.'),
+    'error': ('ERROR', 'E'),
+    'failed': ('FAIL', 'F'),
+    'skipped': ("skipped %r", 's'),
+    'expectedFailure': ("expected failure", 'x'),
+    'unexpectedSuccess': ('unexpected success', 'u'),
+}
 class StopTestEvent(_Event):
-    def __init__(self, test, result, stopTime, timeTaken, 
-                    outcome, exc_info=None, stage=None):
+    def __init__(self, test, result, stopTime, timeTaken, outcome, exc_info,
+                 stage, traceback):
         _Event.__init__(self)
         self.test = test
         self.result = result
         self.stopTime = stopTime
         self.timeTaken = timeTaken
         self.exc_info = exc_info
+
+        self.longResult = None
+        self.shortResult = None
+        self.traceback = traceback
+        try:
+            self.description = result.getDescription(test)
+        except AttributeError:
+            self.description = str(test)
+
+        self.metadata = {}
         
-        # class, setUp, call, tearDown, cleanUp
-        # or None for a pass
-        self.stage = stage
-        self.outcome = outcome
-        
+        self.outcome = None
         self.passed = False
         self.failed = False
         self.error = False
         self.skipReason = None
         self.unexpectedSuccess = False
         self.expectedFailure = False
+
+        # class, setUp, call, tearDown, cleanUp
+        # or None for a pass
+        self.stage = stage
+        self.outcome = outcome
+
+        longResult, shortResult = _DEFAULT_RESULTS[outcome]
+        self.shortResult = shortResult
+        self.longResult = longResult
+
         if outcome == 'passed':
             self.passed = True
         elif outcome == 'failed':
             self.error = True
         elif outcome == 'skipped':
             self.skipped = True
-            self.skipReason = str(exc_info[1])
+            self.skipReason = str(self.exc_info[1])
+            self.longResult = longResult % self.skipReason
         elif outcome == 'unexpectedSuccess':
             self.unexpectedSuccess = True
         elif outcome == 'expectedFailure':
             self.expectedFailure = True
 
+
 class PluginsLoadedEvent(_Event):
     loadedPlugins = loadedPlugins
 
     afterSetUp = _EventHook()
     onTestFail = _EventHook()
     beforeTearDown = _EventHook()
+    createReport = _EventHook()
     stopTest = _EventHook()
     stopTestRun = _EventHook()
     message = _EventHook()

unittest2/result.py

         self._original_stdout = sys.stdout
         self._original_stderr = sys.stderr
         self._mirrorOutput = False
+        self.reports = []
     
     def startTest(self, test):
         "Called when the given test is about to be run"
 
         See startTest for a method called before each test.
         """
+    
+    @failfast
+    def addReport(self, report):
+        self.reports.append(report)
+
+        test = report.test
+        err = report.exc_info
+        reason = report.skipReason
+
+        if report.passed:
+            # call directly on this class instead?
+            self.addSuccess(test)
+        elif report.failed:
+            self.addFailure(test, err)
+        elif report.error:
+            self.addError(test, err)
+        elif report.skipped:
+            self.addSkip(test, reason)
+        elif report.unexpectedSuccess:
+            self.addUnexpectedSuccess(test)
+        elif report.expectedFailure:
+            self.addExpectedFailure(test, err)
 
     def stopTest(self, test):
         """Called when the given test has been run"""

unittest2/runner.py

             raise AttributeError(attr)
         return getattr(self.stream,attr)
     
+    def write(self, arg):
+        if self.runner:
+            self.runner.message(arg, (0, 1, 2))
+        else:
+            self.stream.write(arg)
+
     def writeln(self, arg=None):
         if self.runner:
             arg = arg or ''
             return '\n'.join((str(test), doc_first_line))
         else:
             return str(test)
+    
+    def addReport(self, report):
+        super(TextTestResult, self).addReport(report)
+        if self.showAll:
+            self.stream.writeln(report.longResult)
+        elif self.dots:
+            self.stream.write(report.shortResult)
+            self.stream.flush()
 
     def startTest(self, test):
         super(TextTestResult, self).startTest(test)
             self.stream.write(" ... ")
             self.stream.flush()
 
-    def addSuccess(self, test):
-        super(TextTestResult, self).addSuccess(test)
-        if self.showAll:
-            self.stream.writeln("ok")
-        elif self.dots:
-            self.stream.write('.')
-            self.stream.flush()
-
-    def addError(self, test, err):
-        super(TextTestResult, self).addError(test, err)
-        if self.showAll:
-            self.stream.writeln("ERROR")
-        elif self.dots:
-            self.stream.write('E')
-            self.stream.flush()
-
-    def addFailure(self, test, err):
-        super(TextTestResult, self).addFailure(test, err)
-        if self.showAll:
-            self.stream.writeln("FAIL")
-        elif self.dots:
-            self.stream.write('F')
-            self.stream.flush()
-
-    def addSkip(self, test, reason):
-        super(TextTestResult, self).addSkip(test, reason)
-        if self.showAll:
-            self.stream.writeln("skipped %r" % (reason,))
-        elif self.dots:
-            self.stream.write("s")
-            self.stream.flush()
-
-    def addExpectedFailure(self, test, err):
-        super(TextTestResult, self).addExpectedFailure(test, err)
-        if self.showAll:
-            self.stream.writeln("expected failure")
-        elif self.dots:
-            self.stream.write("x")
-            self.stream.flush()
-
-    def addUnexpectedSuccess(self, test):
-        super(TextTestResult, self).addUnexpectedSuccess(test)
-        if self.showAll:
-            self.stream.writeln("unexpected success")
-        elif self.dots:
-            self.stream.write("u")
-            self.stream.flush()
-
     def printErrors(self):
         if self.dots or self.showAll:
             self.stream.writeln()
+
+        reportsCategories = {}
+        for report in self.reports:
+            reportList = reportsCategories.setdefault(report.outcome, [])
+            reportList.append((report.description, report.traceback))
+        
+        errors = reportsCategories.pop('passed', [])
+        failures = reportsCategories.pop('failed', [])
+        reportsCategories.pop('skipped', None)
+        reportsCategories.pop('ok', None)
+        reportsCategories.pop('expectedFailures', None)
+        reportsCategories.pop('unexpectedSuccess', None)
+
         self.printErrorList('ERROR', self.errors)
         self.printErrorList('FAIL', self.failures)
+        
+        for flavour, results in reportsCategories.items():
+            self.printErrorList(flavour.upper(), results)
 
     def printErrorList(self, flavour, errors):
-        for test, err in errors:
+        for desc, err in errors:
+            if not isinstance(desc, basestring):
+                desc = self.getDescription(desc)
             self.stream.writeln(self.separator1)
-            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
+            self.stream.writeln("%s: %s" % (flavour, desc))
             self.stream.writeln(self.separator2)
             self.stream.writeln("%s" % err)
 
         The default verbosity is (1, 2). If this method is called without an
         explicit verbosity it will be output for verbosities of both 1 and 2.
         """
+        stream = self.stream
+        if hasattr(stream, 'stream'):
+            # for _WritelnDecorator decorated streams
+            stream = stream.stream
+
         if not isinstance(verbosity, basestring):
             try:
                 iter(verbosity)
                 # as non-matched channels will be passed through here
                 verb = VERBOSITIES.get(verb, verb)
             if verb == self.verbosity:
-                self.stream.write(msg)
-                self.stream.flush()
+                stream.write(msg)
+                stream.flush()
                 break
 
     def _makeResult(self):
             if not event.handled:
                 test(result)
         finally:
+            stopTime = time.time()
+            timeTaken = stopTime - startTime
+
+            event = StopTestRunEvent(self, result, stopTime, timeTaken)
+            hooks.stopTestRun(event)
+
             stopTestRun = getattr(result, 'stopTestRun', None)
             if stopTestRun is not None:
                 stopTestRun()
             else:
                 result.printErrors()
-        
-            stopTime = time.time()
-            timeTaken = stopTime - startTime
-            
-            event = StopTestRunEvent(self, result, stopTime, timeTaken)
-            hooks.stopTestRun(event)
 
         if hasattr(result, 'separator2'):
             self.message(result.separator2, (0, 1, 2))
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.