from twisted.trial import unittest
from twisted.internet import defer
from twisted.python import log
from buildbot.util import eventual
class Eventually(unittest.TestCase):
def setUp(self):
# reset the queue to its base state
eventual._theSimpleQueue = eventual._SimpleCallQueue()
self.old_log_err = log.err
self.results = []
def tearDown(self):
log.err = self.old_log_err
return eventual.flushEventualQueue()
# utility callback
def cb(self, *args, **kwargs):
r = args
if kwargs: r = r + (kwargs,)
self.results.append(r)
# flush the queue and assert results
def assertResults(self, exp):
d = eventual.flushEventualQueue()
def cb(_):
self.assertEqual(self.results, exp)
d.addCallback(cb)
return d
## tests
def test_eventually_calls(self):
eventual.eventually(self.cb)
return self.assertResults([()])
def test_eventually_args(self):
eventual.eventually(self.cb, 1, 2, a='a')
return self.assertResults([(1, 2, dict(a='a'))])
def test_eventually_err(self):
# monkey-patch log.err; this is restored by tearDown
log.err = lambda : self.results.append("err")
def cb_fails():
raise RuntimeError("should not cause test failure")
eventual.eventually(cb_fails)
return self.assertResults(['err'])
def test_eventually_butNotNow(self):
eventual.eventually(self.cb, 1)
self.failIf(self.results != [])
return self.assertResults([(1,)])
def test_eventually_order(self):
eventual.eventually(self.cb, 1)
eventual.eventually(self.cb, 2)
eventual.eventually(self.cb, 3)
return self.assertResults([(1,), (2,), (3,)])
def test_flush_waitForChainedEventuallies(self):
def chain(n):
self.results.append(n)
if n <= 0: return
eventual.eventually(chain, n-1)
chain(3)
# (the flush this tests is implicit in assertResults)
return self.assertResults([3, 2, 1, 0])
def test_flush_waitForTreeEventuallies(self):
# a more complex set of eventualities
def tree(n):
self.results.append(n)
if n <= 0: return
eventual.eventually(tree, n-1)
eventual.eventually(tree, n-1)
tree(2)
# (the flush this tests is implicit in assertResults)
return self.assertResults([2, 1, 1, 0, 0, 0, 0])
def test_flush_duringTurn(self):
testd = defer.Deferred()
def cb():
d = eventual.flushEventualQueue()
d.addCallback(testd.callback)
eventual.eventually(cb)
return testd
def test_fireEventually_call(self):
d = eventual.fireEventually(13)
d.addCallback(self.cb)
return self.assertResults([(13,)])
|