Package buildbot :: Package process :: Module step_twisted2
[hide private]
[frames] | no frames]

Source Code for Module buildbot.process.step_twisted2

  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 text = [] 145 if count == 0: 146 text.extend(["%d %s" % \ 147 (total, 148 total == 1 and "test" or "tests"), 149 "passed"]) 150 else: 151 text.append("tests") 152 text.append("%d %s" % \ 153 (count, 154 count == 1 and "failure" or "failures")) 155 self.updateCurrentActivity(text=text) 156 self.addFileToCurrentActivity("tests", self.results) 157 #self.finishStatusSummary() 158 self.finishCurrentActivity()
159