Add test for Linearizer.is_queued(..)

This commit is contained in:
Erik Johnston 2020-05-27 19:41:06 +01:00
parent a6a40a1519
commit a72d5f39db

View File

@ -45,6 +45,38 @@ class LinearizerTestCase(unittest.TestCase):
with (yield d2):
pass
@defer.inlineCallbacks
def test_linearizer_is_queued(self):
linearizer = Linearizer()
key = object()
d1 = linearizer.queue(key)
cm1 = yield d1
# Since d1 gets called immediately, "is_queued" should return false.
self.assertFalse(linearizer.is_queued(key))
d2 = linearizer.queue(key)
self.assertFalse(d2.called)
# Now d2 is queued up behind successful completion of cm1
self.assertTrue(linearizer.is_queued(key))
with cm1:
self.assertFalse(d2.called)
# cm1 still not done, so d2 still queued.
self.assertTrue(linearizer.is_queued(key))
# And now d2 is called and nothing is in the queue again
self.assertFalse(linearizer.is_queued(key))
with (yield d2):
self.assertFalse(linearizer.is_queued(key))
self.assertFalse(linearizer.is_queued(key))
def test_lots_of_queued_things(self):
# we have one slow thing, and lots of fast things queued up behind it.
# it should *not* explode the stack.