buildbot/process/step_twisted2.py

   1
   2 from buildbot.status import tests
   3 from buildbot.process.step import SUCCESS, FAILURE, BuildStep
   4 from buildbot.process.step_twisted import RunUnitTests
   5
   6 from zope.interface import implements
   7 from twisted.python import log, failure
   8 from twisted.spread import jelly
   9 from twisted.pb.tokens import BananaError
  10 from twisted.web.html import PRE
  11 from twisted.web.error import NoResource
  12
  13 class Null: pass
  14 ResultTypes = Null()
  15 ResultTypeNames = ["SKIP",
  16                    "EXPECTED_FAILURE", "FAILURE", "ERROR",
  17                    "UNEXPECTED_SUCCESS", "SUCCESS"]
  18 try:
  19     from twisted.trial import reporter # introduced in Twisted-1.0.5
  20     # extract the individual result types
  21     for name in ResultTypeNames:
  22         setattr(ResultTypes, name, getattr(reporter, name))
  23 except ImportError:
  24     from twisted.trial import unittest # Twisted-1.0.4 has them here
  25     for name in ResultTypeNames:
  26         setattr(ResultTypes, name, getattr(unittest, name))
  27
  28 log._keepErrors = 0
  29 from twisted.trial import remote # for trial/jelly parsing
  30
  31 import StringIO
  32
  33 class OneJellyTest(tests.OneTest):
  34     def html(self, request):
  35         tpl = "<HTML><BODY>\n\n%s\n\n</body></html>\n"
  36         pptpl = "<HTML><BODY>\n\n<pre>%s</pre>\n\n</body></html>\n"
  37         t = request.postpath[0] # one of 'short', 'long' #, or 'html'
  38         if isinstance(self.results, failure.Failure):
  39             # it would be nice to remove unittest functions from the
  40             # traceback like unittest.format_exception() does.
  41             if t == 'short':
  42                 s = StringIO.StringIO()
  43                 self.results.printTraceback(s)
  44                 return pptpl % PRE(s.getvalue())
  45             elif t == 'long':
  46                 s = StringIO.StringIO()
  47                 self.results.printDetailedTraceback(s)
  48                 return pptpl % PRE(s.getvalue())
  49             #elif t == 'html':
  50             #    return tpl % formatFailure(self.results)
  51             # ACK! source lines aren't stored in the Failure, rather,
  52             # formatFailure pulls them (by filename) from the local
  53             # disk. Feh. Even printTraceback() won't work. Double feh.
  54             return NoResource("No such mode '%s'" % t)
  55         if self.results == None:
  56             return tpl % "No results to show: test probably passed."
  57         # maybe results are plain text?
  58         return pptpl % PRE(self.results)
  59
  60 class TwistedJellyTestResults(tests.TestResults):
  61     oneTestClass = OneJellyTest
  62     def describeOneTest(self, testname):
  63         return "%s: %s\n" % (testname, self.tests[testname][0])
  64
  65 class RunUnitTestsJelly(RunUnitTests):
  66     """I run the unit tests with the --jelly option, which generates
  67     machine-parseable results as the tests are run.
  68     """
  69     trialMode = "--jelly"
  70     implements(remote.IRemoteReporter)
  71
  72     ourtypes = { ResultTypes.SKIP: tests.SKIP,
  73                  ResultTypes.EXPECTED_FAILURE: tests.EXPECTED_FAILURE,
  74                  ResultTypes.FAILURE: tests.FAILURE,
  75                  ResultTypes.ERROR: tests.ERROR,
  76                  ResultTypes.UNEXPECTED_SUCCESS: tests.UNEXPECTED_SUCCESS,
  77                  ResultTypes.SUCCESS: tests.SUCCESS,
  78                  }
  79
  80     def __getstate__(self):
  81         #d = RunUnitTests.__getstate__(self)
  82         d = self.__dict__.copy()
  83         # Banana subclasses are Ephemeral
  84         if d.has_key("decoder"):
  85             del d['decoder']
  86         return d
  87     def start(self):
  88         self.decoder = remote.DecodeReport(self)
  89         # don't accept anything unpleasant from the (untrusted) build slave
  90         # The jellied stream may have Failures, but everything inside should
  91         # be a string
  92         security = jelly.SecurityOptions()
  93         security.allowBasicTypes()
  94         security.allowInstancesOf(failure.Failure)
  95         self.decoder.taster = security
  96         self.results = TwistedJellyTestResults()
  97         RunUnitTests.start(self)
  98
  99     def logProgress(self, progress):
 100         # XXX: track number of tests
 101         BuildStep.logProgress(self, progress)
 102
 103     def addStdout(self, data):
 104         if not self.decoder:
 105             return
 106         try:
 107             self.decoder.dataReceived(data)
 108         except BananaError:
 109             self.decoder = None
 110             log.msg("trial --jelly output unparseable, traceback follows")
 111             log.deferr()
 112
 113     def remote_start(self, expectedTests, times=None):
 114         print "remote_start", expectedTests
 115     def remote_reportImportError(self, name, aFailure, times=None):
 116         pass
 117     def remote_reportStart(self, testClass, method, times=None):
 118         print "reportStart", testClass, method
 119
 120     def remote_reportResults(self, testClass, method, resultType, results,
 121                              times=None):
 122         print "reportResults", testClass, method, resultType
 123         which = testClass + "." + method
 124         self.results.addTest(which,
 125                              self.ourtypes.get(resultType, tests.UNKNOWN),
 126                              results)
 127
 128     def finished(self, rc):
 129         # give self.results to our Build object
 130         self.build.testsFinished(self.results)
 131         total = self.results.countTests()
 132         count = self.results.countFailures()
 133         result = SUCCESS
 134         if total == None:
 135             result = (FAILURE, ['tests%s' % self.rtext(' (%s)')])
 136         if count:
 137             result = (FAILURE, ["%d tes%s%s" % (count,
 138                                                 (count == 1 and 't' or 'ts'),
 139                                                 self.rtext(' (%s)'))])
 140         return self.stepComplete(result)
 141     def finishStatus(self, result):
 142         total = self.results.countTests()
 143         count = self.results.countFailures()
 144         color = "green"
 145         text = []
 146         if count == 0:
 147             text.extend(["%d %s" % \
 148                          (total,
 149                           total == 1 and "test" or "tests"),
 150                          "passed"])
 151         else:
 152             text.append("tests")
 153             text.append("%d %s" % \
 154                         (count,
 155                          count == 1 and "failure" or "failures"))
 156             color = "red"
 157         self.updateCurrentActivity(color=color, text=text)
 158         self.addFileToCurrentActivity("tests", self.results)
 159         #self.finishStatusSummary()
 160         self.finishCurrentActivity()
 161