You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mitogen/tests/select_test.py

477 lines
16 KiB
Python

import mitogen.core
import mitogen.select
import testlib
class BoolTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_latch(self):
latch = mitogen.core.Latch() # oneshot
select = self.klass()
self.assertFalse(select)
select.add(latch)
self.assertTrue(select)
latch.put(123)
self.assertTrue(select)
self.assertEqual(123, select.get())
self.assertFalse(select)
def test_receiver(self):
recv = mitogen.core.Receiver(self.router) # oneshot
select = self.klass()
self.assertFalse(select)
select.add(recv)
self.assertTrue(select)
recv._on_receive(mitogen.core.Message.pickled('123'))
self.assertTrue(select)
self.assertEqual('123', select.get().unpickle())
self.assertFalse(select)
class AddTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_latch(self):
latch = mitogen.core.Latch()
select = self.klass()
select.add(latch)
self.assertEqual(1, len(select._receivers))
self.assertEqual(latch, select._receivers[0])
self.assertEqual(select._put, latch.notify)
def test_receiver(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass()
select.add(recv)
self.assertEqual(1, len(select._receivers))
self.assertEqual(recv, select._receivers[0])
self.assertEqual(select._put, recv.notify)
def test_channel(self):
context = self.router.local()
chan = mitogen.core.Channel(self.router, context, 1234)
select = self.klass()
select.add(chan)
self.assertEqual(1, len(select._receivers))
self.assertEqual(chan, select._receivers[0])
self.assertEqual(select._put, chan.notify)
def test_subselect_empty(self):
select = self.klass()
subselect = self.klass()
select.add(subselect)
self.assertEqual(1, len(select._receivers))
self.assertEqual(subselect, select._receivers[0])
self.assertEqual(select._put, subselect.notify)
def test_subselect_nonempty(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass()
subselect = self.klass()
subselect.add(recv)
select.add(subselect)
self.assertEqual(1, len(select._receivers))
self.assertEqual(subselect, select._receivers[0])
self.assertEqual(select._put, subselect.notify)
def test_subselect_loop_direct(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.add(select))
self.assertEqual(str(exc), self.klass.loop_msg)
def test_subselect_loop_indirect(self):
s0 = self.klass()
s1 = self.klass()
s2 = self.klass()
s0.add(s1)
s1.add(s2)
exc = self.assertRaises(mitogen.select.Error,
lambda: s2.add(s0))
self.assertEqual(str(exc), self.klass.loop_msg)
def test_double_add_receiver(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
select.add(recv)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.add(recv))
self.assertEqual(str(exc), self.klass.owned_msg)
def test_double_add_subselect(self):
select = self.klass()
select2 = self.klass()
select.add(select2)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.add(select2))
self.assertEqual(str(exc), self.klass.owned_msg)
class RemoveTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_receiver_empty(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.remove(recv))
self.assertEqual(str(exc), self.klass.not_present_msg)
def test_receiver_absent(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
recv2 = mitogen.core.Receiver(self.router)
select.add(recv2)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.remove(recv))
self.assertEqual(str(exc), self.klass.not_present_msg)
def test_receiver_present(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
select.add(recv)
select.remove(recv)
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, recv.notify)
def test_latch_empty(self):
select = self.klass()
latch = mitogen.core.Latch()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.remove(latch))
self.assertEqual(str(exc), self.klass.not_present_msg)
def test_latch_absent(self):
select = self.klass()
latch = mitogen.core.Latch()
latch2 = mitogen.core.Latch()
select.add(latch2)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.remove(latch))
self.assertEqual(str(exc), self.klass.not_present_msg)
def test_latch_present(self):
select = self.klass()
latch = mitogen.core.Latch()
select.add(latch)
select.remove(latch)
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, latch.notify)
class CloseTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_empty(self):
select = self.klass()
select.close() # No effect.
def test_one_latch(self):
select = self.klass()
latch = mitogen.core.Latch()
select.add(latch)
self.assertEqual(1, len(select._receivers))
self.assertEqual(select._put, latch.notify)
select.close()
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, latch.notify)
def test_one_receiver(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
select.add(recv)
self.assertEqual(1, len(select._receivers))
self.assertEqual(select._put, recv.notify)
select.close()
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, recv.notify)
def test_one_subselect(self):
select = self.klass()
subselect = self.klass()
select.add(subselect)
recv = mitogen.core.Receiver(self.router)
subselect.add(recv)
self.assertEqual(1, len(select._receivers))
self.assertEqual(subselect._put, recv.notify)
select.close()
self.assertEqual(0, len(select._receivers))
self.assertEqual(subselect._put, recv.notify)
subselect.close()
self.assertEqual(None, recv.notify)
class EmptyTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_no_receivers(self):
select = self.klass()
self.assertTrue(select.empty())
def test_empty_receiver(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
self.assertTrue(select.empty())
def test_nonempty_receiver_before_add(self):
recv = mitogen.core.Receiver(self.router)
recv._on_receive(mitogen.core.Message.pickled('123'))
select = self.klass([recv])
self.assertFalse(select.empty())
def test_nonempty__receiver_after_add(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
recv._on_receive(mitogen.core.Message.pickled('123'))
self.assertFalse(select.empty())
def test_empty_latch(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
self.assertTrue(select.empty())
def test_nonempty_latch_before_add(self):
latch = mitogen.core.Latch()
latch.put(123)
select = self.klass([latch])
self.assertFalse(select.empty())
def test_nonempty__latch_after_add(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
latch.put(123)
self.assertFalse(select.empty())
class IterTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_empty(self):
select = self.klass()
self.assertEqual([], list(select))
def test_nonempty_receiver(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
msg = mitogen.core.Message.pickled('123')
recv._on_receive(msg)
self.assertEqual([msg], list(select))
def test_nonempty_latch(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
latch.put(123)
self.assertEqual([123], list(select))
class OneShotTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_true_receiver_removed_after_get(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
msg = mitogen.core.Message.pickled('123')
recv._on_receive(msg)
msg_ = select.get()
self.assertEqual(msg, msg_)
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, recv.notify)
def test_false_receiver_persists_after_get(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv], oneshot=False)
msg = mitogen.core.Message.pickled('123')
recv._on_receive(msg)
self.assertEqual(msg, select.get())
self.assertEqual(1, len(select._receivers))
self.assertEqual(recv, select._receivers[0])
self.assertEqual(select._put, recv.notify)
def test_true_latch_removed_after_get(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
latch.put(123)
self.assertEqual(123, select.get())
self.assertEqual(0, len(select._receivers))
self.assertEqual(None, latch.notify)
def test_false_latch_persists_after_get(self):
latch = mitogen.core.Latch()
select = self.klass([latch], oneshot=False)
latch.put(123)
self.assertEqual(123, select.get())
self.assertEqual(1, len(select._receivers))
self.assertEqual(latch, select._receivers[0])
self.assertEqual(select._put, latch.notify)
class GetReceiverTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_no_receivers(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.get())
self.assertEqual(str(exc), self.klass.empty_msg)
def test_timeout_no_receivers(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.get(timeout=1.0))
self.assertEqual(str(exc), self.klass.empty_msg)
def test_zero_timeout(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.0))
def test_timeout(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.1))
def test_nonempty_before_add(self):
recv = mitogen.core.Receiver(self.router)
recv._on_receive(mitogen.core.Message.pickled('123'))
select = self.klass([recv])
msg = select.get()
self.assertEqual('123', msg.unpickle())
def test_nonempty_multiple_items_before_add(self):
recv = mitogen.core.Receiver(self.router)
recv._on_receive(mitogen.core.Message.pickled('123'))
recv._on_receive(mitogen.core.Message.pickled('234'))
select = self.klass([recv], oneshot=False)
msg = select.get()
self.assertEqual('123', msg.unpickle())
msg = select.get()
self.assertEqual('234', msg.unpickle())
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(block=False))
def test_nonempty_after_add(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
recv._on_receive(mitogen.core.Message.pickled('123'))
msg = select.get()
self.assertEqual('123', msg.unpickle())
def test_nonempty_receiver_attr_set(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
recv._on_receive(mitogen.core.Message.pickled('123'))
msg = select.get()
self.assertEqual(msg.receiver, recv)
def test_drained_by_other_thread(self):
recv = mitogen.core.Receiver(self.router)
recv._on_receive(mitogen.core.Message.pickled('123'))
select = self.klass([recv])
msg = recv.get()
self.assertEqual('123', msg.unpickle())
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.0))
class GetLatchTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_no_latches(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.get())
self.assertEqual(str(exc), self.klass.empty_msg)
def test_timeout_no_receivers(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.get(timeout=1.0))
self.assertEqual(str(exc), self.klass.empty_msg)
def test_zero_timeout(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.0))
def test_timeout(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.1))
def test_nonempty_before_add(self):
latch = mitogen.core.Latch()
latch.put(123)
select = self.klass([latch])
self.assertEqual(123, select.get())
def test_nonempty_multiple_items_before_add(self):
latch = mitogen.core.Latch()
latch.put(123)
latch.put(234)
select = self.klass([latch], oneshot=False)
self.assertEqual(123, select.get())
self.assertEqual(234, select.get())
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(block=False))
def test_nonempty_after_add(self):
latch = mitogen.core.Latch()
select = self.klass([latch])
latch.put(123)
self.assertEqual(123, latch.get())
def test_drained_by_other_thread(self):
latch = mitogen.core.Latch()
latch.put(123)
select = self.klass([latch])
self.assertEqual(123, latch.get())
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.0))
class GetEventTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_empty(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.get())
self.assertEqual(str(exc), self.klass.empty_msg)
def test_latch(self):
latch = mitogen.core.Latch()
latch.put(123)
select = self.klass([latch])
event = select.get_event()
self.assertEqual(latch, event.source)
self.assertEqual(123, event.data)
def test_receiver(self):
recv = mitogen.core.Receiver(self.router)
recv._on_receive(mitogen.core.Message.pickled('123'))
select = self.klass([recv])
event = select.get_event()
self.assertEqual(recv, event.source)
self.assertEqual('123', event.data.unpickle())