from twisted.trial import unittest
from twisted.internet import task
from twisted.internet import reactor
class TestException(Exception):
pass
class LoopTestCase(unittest.TestCase):
def testBasicFunction(self):
L = []
def foo(a, b, c=None, d=None):
L.append((a, b, c, d))
lc = task.LoopingCall(foo, "a", "b", d="d")
d = lc.start(0.1)
reactor.callLater(1, lc.stop)
result = unittest.deferredResult(d)
self.assertIdentical(lc, result)
self.failUnless(9 <= len(L) <= 11)
for (a, b, c, d) in L:
self.assertEquals(a, "a")
self.assertEquals(b, "b")
self.assertEquals(c, None)
self.assertEquals(d, "d")
def testFailure(self):
def foo(x):
raise TestException(x)
lc = task.LoopingCall(foo, "bar")
d = lc.start(0.1)
err = unittest.deferredError(d)
err.trap(TestException)