Phillip sounds good. Please see comments inline.
Phillip J. Eby wrote:
At 04:23 PM 6/6/2005 -0700, Brian Kirsch wrote:
Ok sounds good. Especially if you wanna do the work of implementing
the mock reactor :)
Well, I took a look this morning, and implementing simulated time may
actually be easier than I thought, because ReactorBase now has a
'timeout()' method that does the hard part - computing how much time
there is until the next delayed call. In earlier versions of Twisted,
that logic was buried inside something else. There are also nice new
functions like 'installReactor' to play with.
So, I'm thinking I'll create a 'testreactor' module (maybe living in
osaf.framework.twisted?) with 'install()' and 'uninstall()'
functions. When installed, it'll mixin test methods into the standard
reactor, and the reactor will run on simulated time. I'm thinking
that these will probably be very simple, like a 'reset()' to
completely clear the reactor's state between tests.
I've also investigated the run()/stop() issue, and as far as I can
tell this should be a non-issue for tests that run using the Twisted
default reactor (which is what Chandler uses). The reason that you
aren't supposed to use run()/stop() or certain other functions
multiple times (AFAICT), is because certain GUI-platform-specific
reactor implementations won't tolerate it. However, tests that don't
run with a GUI don't need to use a GUI-specific reactor, and we're not
using one in Chandler currently anyway.
Assuming that StringTransport is sufficient for all our
simulated-socket needs, I should only need to add a few methods. Here
are the ones I have in mind:
StringTransport is one way of doing it. You can also use the loopback
mechanism in Twisted to hook up real Twisted client to Twisted server
communication. The test_pop3client.py unittest I modified for Twisted
core does this to test timeouts. I Wrote a testPop3Server that runs
under Twisted to test against the pop3client.py code and it worked like
a charm:
class POP3HelperMixin:
serverCTX = None
clientCTX = None
def setUp(self):
d = defer.Deferred()
self.server =
pop3TestServer.POP3TestServer(contextFactory=self.serverCTX)
self.client = SimpleClient(d, contextFactory=self.clientCTX)
self.client.timeout = 30
self.connected = d
def tearDown(self):
del self.server
del self.client
del self.connected
def _cbStopClient(self, ignore):
self.client.transport.loseConnection()
def _ebGeneral(self, failure):
self.client.transport.loseConnection()
self.server.transport.loseConnection()
failure.printTraceback(open('failure.log', 'w'))
failure.printTraceback()
raise failure.value
def loopback(self):
loopback.loopbackTCP(self.server, self.client, noisy=False)
class POP3TimeoutTestCase(POP3HelperMixin, unittest.TestCase):
def testTimeout(self):
def login():
#this will startTLS automatically
d = self.client.login('test', 'twisted')
d.addErrback(timedOut)
return d
def timedOut(failure):
self._cbStopClient(None)
failure.trap(error.TimeoutError)
def quit():
return self.client.quit()
self.client.timeout = 3
#No need to leverage SSL for timeout test
pop3TestServer.SSL_SUPPORT = False
#Tell the server to not return a response to client.
#This will trigger a timeout.
pop3TestServer.TIMEOUT_RESPONSE = True
methods = [login, quit]
map(self.connected.addCallback, map(strip, methods))
self.connected.addCallback(login)
self.connected.addCallback(quit)
self.connected.addCallback(self._cbStopClient)
self.connected.addErrback(self._ebGeneral)
self.loopback()
reactor.start() -- completely reset the reactor's state and
reinitialize it so that the current test runs with a clean slate.
Also sets the reactor's state to "running", so that the test will
appear to be called from within the reactor run loop, but in fact will
have control over the reactor.
reactor.getTime() -- return the current simulated "now" moment
(initially set to the time when the test mixin is installed)
reactor.setTime(seconds) -- change the simulated "now" moment
reactor.waitFor(deferred, timeout) -- iterate until the deferred
succeeds or fails, or until the timeout occurs. Raise a TimeoutError
if the timeout happens first, or raises the deferred's exception if
the deferred fails. Otherwise, returns the deferred's value. (Note:
the timeout is in *simulated* time, not real time, so you can safely
set large values if you have a test with complex timing.)
So, I'm thinking there would also be a 'testreactor.ReactorTestCase'
whose setUp() would testreactor.install(), and reactor.start(), and
whose tearDown() would reactor.stop() (if it's still running), and
perhaps testreactor.uninstall() as well.
Then, test methods would just set up whatever conditions they want,
and then reactor.waitFor() results. Tests could safely call waitFor()
multiple times, even within the same test.
Heikki, does that sound about like what you need for your tests?
--
Brian Kirsch - Email Framework Engineer
Open Source Applications Foundation
543 Howard St. 5th Floor
San Francisco, CA 94105
(415) 946-3056
http://www.osafoundation.org
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
Open Source Applications Foundation "Dev" mailing list
http://lists.osafoundation.org/mailman/listinfo/dev
