Conway, Nicholas J
2012-05-23 21:30:45 UTC
Hi,
I have a class that needs to kick off a method the repeatedly gets call every so many seconds, lets say 2 seconds.
I chose to use task.deferLater to do this, but seems like LoopingCall gives similar results..
I've boiled it down to the following example
from twisted.internet import reactor, task
from twisted.trial import unittest
class MyLoop(object):
def __init__(self):
self.updateParameters()
def updateParameters(self):
d = task.deferLater(reactor, 2, self.dosomething)
def dosomething(self):
print "cool"
class LoopTestCase(unittest.TestCase):
def setUp(self):
self.cL = MyLoop()
def test_dummy(self):
d = defer.Deferred()
d.addCallback(lambda x: x)
d.callback(None)
return d # return a deferred just in case but happens either way
MyLoop works when running normally with a reactor.run(), but when I try run this test, I get the following error:
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x1018fbb90 [1.99891901016s] called=0 cancelled=0 Deferred.callback(None)>
So I don't fully understand what's going on to cause this. Any pointers would be appreciated
Thanks,
-Nick
I have a class that needs to kick off a method the repeatedly gets call every so many seconds, lets say 2 seconds.
I chose to use task.deferLater to do this, but seems like LoopingCall gives similar results..
I've boiled it down to the following example
from twisted.internet import reactor, task
from twisted.trial import unittest
class MyLoop(object):
def __init__(self):
self.updateParameters()
def updateParameters(self):
d = task.deferLater(reactor, 2, self.dosomething)
def dosomething(self):
print "cool"
class LoopTestCase(unittest.TestCase):
def setUp(self):
self.cL = MyLoop()
def test_dummy(self):
d = defer.Deferred()
d.addCallback(lambda x: x)
d.callback(None)
return d # return a deferred just in case but happens either way
MyLoop works when running normally with a reactor.run(), but when I try run this test, I get the following error:
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x1018fbb90 [1.99891901016s] called=0 cancelled=0 Deferred.callback(None)>
So I don't fully understand what's going on to cause this. Any pointers would be appreciated
Thanks,
-Nick