On Mon, 2009-11-02 at 01:12 -0500, Glyph Lefkowitz wrote:
On Nov 1, 2009, at 11:17 PM, Crispin Wellington wrote:
I tried using LoopingCall, but it does not work. It only calls the scheduler once. I think this has to do with the fact that the stackless scheduler needs to be interwoven with the twisted reactor pump.
What do you mean "does not work"?
OK. Having a closer look, its not that looping call doesn't work, its that there is some unknown number of "reactor pumps" between starting the test, and finishing it. What I need is a way for the reactor to be pumping away while a particular test function of a testcase continues working. As an example, here is a non-working test case. Notice the comment "#what to do here to pump the reactor?". ------------------ from twisted.trial import unittest, reporter, runner import os, time import stackless from twisted.internet import reactor, task def example_func(t=10.0): """wait for t seconds then return""" exit_time = time.time()+t while time.time()<exit_time: stackless.schedule() t = task.LoopingCall(stackless.schedule) t.start(0.1) class StacklessTest(unittest.TestCase): def setUpClass(self): pass def test_stackless(self): """Test that we can successfuly create a user proxy cert""" # get the time now... start = time.time() task = stackless.tasklet(example_func) task.setup(10.0) task.run() while task.alive: pass #what to do here to pump the reactor? # end time end = time.time() self.assert_( endtime - starttime >= 10.0 ) --------------------- Running this under trial, it just hangs, inside the while task.alive: loop. So I guess my problem is my approach. How do I test long running "tasklets" that use twisted calls (unlike this contrived example that only sleeps) within the twisted trial framework? How would the Twisted experts write test code for a case like this? Any help is much appreciated! Kind Regards Crispin Wellington