2 from twisted.trial import unittest
3 from twisted.internet import defer, reactor
4 from allmydata.util import observer
6 def nextTurn(res=None):
8 reactor.callLater(1, d.callback, res)
11 class Observer(unittest.TestCase):
12 def test_oneshot(self):
13 ol = observer.OneShotObserverList()
15 self.failUnlessEqual(rep, "<OneShotObserverList [[]]>")
19 self.failUnlessEqual(res, "result")
21 d3.addCallback(self.failUnlessEqual, "result")
23 d1.addCallback(_addmore)
26 self.failUnlessEqual(rep, "<OneShotObserverList -> result>")
28 dl = defer.DeferredList([d1,d2,d4])
31 def test_oneshot_fireagain(self):
32 ol = observer.OneShotObserverList()
35 self.failUnlessEqual(res, "result")
36 ol.fire_if_not_fired("result3") # should be ignored
38 d2.addCallback(self.failUnlessEqual, "result")
40 d.addCallback(_addmore)
41 ol.fire_if_not_fired("result")
42 ol.fire_if_not_fired("result2")
45 def test_lazy_oneshot(self):
46 ol = observer.LazyOneShotObserverList()
50 self.failUnlessEqual(res, "result")
52 d3.addCallback(self.failUnlessEqual, "result")
54 d1.addCallback(_addmore)
59 dl = defer.DeferredList([d1,d2,d4])
62 def test_observerlist(self):
63 ol = observer.ObserverList()
67 ol.subscribe(l1.append)
69 ol.subscribe(l2.append)
71 ol.unsubscribe(l1.append)
74 self.failUnlessEqual(l1, [1,2])
75 self.failUnlessEqual(l2, [2,3])
79 def _add(a, b, c=None):
81 ol.unsubscribe(l2.append)
86 self.failUnlessEqual(l3, [(4,5,6)])
88 d.addCallback(_check2)