src/allmydata/test/test_observer.py

   1
   2 from twisted.trial import unittest
   3 from twisted.internet import defer, reactor
   4 from allmydata.util import observer
   5
   6 def nextTurn(res=None):
   7     d = defer.Deferred()
   8     reactor.callLater(1, d.callback, res)
   9     return d
  10
  11 class Observer(unittest.TestCase):
  12     def test_oneshot(self):
  13         ol = observer.OneShotObserverList()
  14         rep = repr(ol)
  15         self.failUnlessEqual(rep, "<OneShotObserverList [[]]>")
  16         d1 = ol.when_fired()
  17         d2 = ol.when_fired()
  18         def _addmore(res):
  19             self.failUnlessEqual(res, "result")
  20             d3 = ol.when_fired()
  21             d3.addCallback(self.failUnlessEqual, "result")
  22             return d3
  23         d1.addCallback(_addmore)
  24         ol.fire("result")
  25         rep = repr(ol)
  26         self.failUnlessEqual(rep, "<OneShotObserverList -> result>")
  27         d4 = ol.when_fired()
  28         dl = defer.DeferredList([d1,d2,d4])
  29         return dl
  30
  31     def test_oneshot_fireagain(self):
  32         ol = observer.OneShotObserverList()
  33         d = ol.when_fired()
  34         def _addmore(res):
  35             self.failUnlessEqual(res, "result")
  36             ol.fire_if_not_fired("result3") # should be ignored
  37             d2 = ol.when_fired()
  38             d2.addCallback(self.failUnlessEqual, "result")
  39             return d2
  40         d.addCallback(_addmore)
  41         ol.fire_if_not_fired("result")
  42         ol.fire_if_not_fired("result2")
  43         return d
  44
  45     def test_lazy_oneshot(self):
  46         ol = observer.LazyOneShotObserverList()
  47         d1 = ol.when_fired()
  48         d2 = ol.when_fired()
  49         def _addmore(res):
  50             self.failUnlessEqual(res, "result")
  51             d3 = ol.when_fired()
  52             d3.addCallback(self.failUnlessEqual, "result")
  53             return d3
  54         d1.addCallback(_addmore)
  55         def _get_result():
  56             return "result"
  57         ol.fire(_get_result)
  58         d4 = ol.when_fired()
  59         dl = defer.DeferredList([d1,d2,d4])
  60         return dl
  61
  62     def test_observerlist(self):
  63         ol = observer.ObserverList()
  64         l1 = []
  65         l2 = []
  66         l3 = []
  67         ol.subscribe(l1.append)
  68         ol.notify(1)
  69         ol.subscribe(l2.append)
  70         ol.notify(2)
  71         ol.unsubscribe(l1.append)
  72         ol.notify(3)
  73         def _check(res):
  74             self.failUnlessEqual(l1, [1,2])
  75             self.failUnlessEqual(l2, [2,3])
  76         d = nextTurn()
  77         d.addCallback(_check)
  78         def _step2(res):
  79             def _add(a, b, c=None):
  80                 l3.append((a,b,c))
  81             ol.unsubscribe(l2.append)
  82             ol.subscribe(_add)
  83             ol.notify(4, 5, c=6)
  84             return nextTurn()
  85         def _check2(res):
  86             self.failUnlessEqual(l3, [(4,5,6)])
  87         d.addCallback(_step2)
  88         d.addCallback(_check2)
  89         return d