"""Remote reporting for Trial.
For reporting test results in a seperate process.
"""
from __future__ import nested_scopes
import reporter
from twisted.internet import protocol
from twisted.python import components, failure, reflect
from twisted.spread import banana, jelly
import os, types
class OneWayBanana(banana.Banana):
knownDialects = ["none"]
isClient = 0
def connectionMade(self):
banana.Banana.connectionMade(self)
self._selectDialect("none")
class JellyReporter(reporter.Reporter):
"""I report results as a Banana-encoded Jelly stream.
This reporting format is machine-readable. It might make more sense
to proxy to a pb.Referenceable Reporter, but then it would need a
two-way connection and a reactor running to manage the protocol. I'm
not sure if we want to do that.
Decode this stream with L{DecodeReport}.
"""
doSendTimes = True
def __init__(self, stream=None):
self.stream = stream
self.banana = OneWayBanana(isClient=0)
if stream is not None:
self.makeConnection(stream)
reporter.Reporter.__init__(self)
def reportImportError(self, name, exc):
f = failure.Failure(exc)
self.jellyMethodCall("reportImportError", name, f)
reporter.Reporter.reportImportError(self, name, exc)
def jellyMethodCall(self, methodName, *args):
if self.doSendTimes:
sexp = jelly.jelly((methodName, args, os.times()))
else:
sexp = jelly.jelly((methodName, args))
self.banana.sendEncoded(sexp)
def makeConnection(self, transport):
self.banana.makeConnection(transport)
def dataReceived(self, data):
if not data:
pass
elif (not self._gotNegotiation) and (data == '\x04\x82none'):
self._gotNegotiation = data
else:
raise ValueError("I should not be getting this data", data)
_gotNegotiation = None
def connectionLost(self, reason):
pass
def start(self, *args):
self.jellyMethodCall("start", *args)
reporter.Reporter.start(self, *args)
def stop(self, *args):
self.jellyMethodCall("stop", *args)
reporter.Reporter.stop(self, *args)
def cleanResults(self, testClass, method):
if type(testClass) == types.ClassType:
testClass = reflect.qual(testClass)
if type(method) == types.MethodType:
method = method.__name__
return (testClass, method)
def reportStart(self, testClass, method):
testClassName, methodName = self.cleanResults(testClass, method)
self.jellyMethodCall("reportStart", testClassName, methodName)
reporter.Reporter.reportStart(self, testClass, method)
def reportResults(self, testClass, method, resultType, results=None):
jresults = results
if type(jresults) == types.TupleType:
typ, val, tb = jresults
jresults = failure.Failure(val, typ, tb)
if reflect.isinst(jresults, failure.Failure):
jresults.type = str(jresults.type)
jresults.value = str(jresults.value)
testClassName, methodName = self.cleanResults(testClass, method)
self.jellyMethodCall("reportResults",
testClassName, methodName,
resultType,
jresults)
reporter.Reporter.reportResults(self, testClass, method, resultType,
results)
class NullTransport:
"""Transport to /dev/null."""
def write(self, *bytes):
return
class IRemoteReporter(components.Interface):
"""I am reporting results from a test suite running someplace else.
The interface is mostly identical to reporter.Reporter, the main difference
being that where it uses exc_info tuples, I use L{failure.Failure}s.
"""
def remote_start(self, expectedTests, times=None):
pass
def remote_reportImportError(self, name, aFailure, times=None):
pass
def remote_reportStart(self, testClass, method, times=None):
pass
def remote_reportResults(self, testClass, method, resultType, results,
times=None):
pass
def remote_stop(self, times=None):
pass
class DecodeReport(banana.Banana):
def __init__(self, reporter):
self.reporter = reporter
self.taster = jelly.DummySecurityOptions()
banana.Banana.__init__(self)
self.transport = NullTransport()
self.connectionMade()
def expressionReceived(self, lst):
lst = jelly.unjelly(lst, self.taster)
methodName, args = lst[:2]
if len(lst) > 2:
times = lst[2]
else:
times = None
if len(lst) > 3:
raise ValueError("I did something wrong", len(lst))
method = getattr(self.reporter, "remote_" + methodName, None)
if method is not None:
method(*(args + (times,)))
class TrialProcessProtocol(DecodeReport, protocol.ProcessProtocol):
def outReceived(self, data):
return self.dataReceived(data)
def errReceived(self, data):
self.log(data)
class DemoRemoteReporter:
__implements__ = (IRemoteReporter,)
def remote_start(self, expectedTests, times=None):
self.startTimes = times
self.printTimeStamped(times, "start")
def remote_reportImportError(self, name, aFailure, times=None):
pass
def remote_reportStart(self, testClass, method, times=None):
self.printTimeStamped(times, "startTest", method)
def remote_reportResults(self, testClass, method, resultType, results,
times=None):
pass
def remote_stop(self, times=None):
self.endTimes = times
self.printTimeStamped(times, "Done!")
usert, syst, wallt = (self.endTimes[0] - self.startTimes[0],
self.endTimes[1] - self.startTimes[1],
self.endTimes[-1] - self.startTimes[-1])
print "CPU time: %.2f Wall clock time: %.2f" % (usert+syst,
wallt)
def printTimeStamped(self, times, *args):
print ("%.2f %.2f %.2f %.2f %.2f " % times), " ".join(map(str, args))
from twisted.application import service
class _Demo(service.Service):
"""Demonstrate trial.remote by spawning trial in a subprocess
and displaying results. Wrapped in an ApplicationService so I can
use twistd --debug on it.
"""
def startService(self):
from twisted.internet import reactor
from twisted.python import log
from twisted.trial import remote
reporter = remote.DemoRemoteReporter()
proto = remote.TrialProcessProtocol(reporter)
def processEnded(reason):
log.err(reason)
reactor.callLater(0, reactor.stop)
proto.processEnded = processEnded
targs = ('--jelly', '-m', 'twisted.test.test_trial')
log.msg("Running `bin/trial %s`" % (" ".join(targs),))
reactor.spawnProcess(proto, "bin/trial", ('trial',) + targs)
def demo():
"""Make a .tap which will demonstrate trial.remote."""
from twisted.trial import remote
myApp = service.Application("tdemo")
remote._Demo().setServiceParent(myApp)
myApp.save()
print "demo saved to tdemo.tap"