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

274 lines
8.7 KiB
Python

import unittest2
import mitogen.select
import testlib
class AddTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_receiver(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass()
select.add(recv)
self.assertEquals(1, len(select._receivers))
self.assertEquals(recv, select._receivers[0])
self.assertEquals(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.assertEquals(1, len(select._receivers))
self.assertEquals(chan, select._receivers[0])
self.assertEquals(select._put, chan.notify)
def test_subselect_empty(self):
select = self.klass()
subselect = self.klass()
select.add(subselect)
self.assertEquals(1, len(select._receivers))
self.assertEquals(subselect, select._receivers[0])
self.assertEquals(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.assertEquals(1, len(select._receivers))
self.assertEquals(subselect, select._receivers[0])
self.assertEquals(select._put, subselect.notify)
def test_subselect_loop_direct(self):
select = self.klass()
exc = self.assertRaises(mitogen.select.Error,
lambda: select.add(select))
self.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(str(exc), self.klass.owned_msg)
class RemoveTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_empty(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
exc = self.assertRaises(mitogen.select.Error,
lambda: select.remove(recv))
self.assertEquals(str(exc), self.klass.not_present_msg)
def test_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.assertEquals(str(exc), self.klass.not_present_msg)
def test_present(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
select.add(recv)
select.remove(recv)
self.assertEquals(0, len(select._receivers))
self.assertEquals(None, recv.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_receiver(self):
select = self.klass()
recv = mitogen.core.Receiver(self.router)
select.add(recv)
self.assertEquals(1, len(select._receivers))
self.assertEquals(select._put, recv.notify)
select.close()
self.assertEquals(0, len(select._receivers))
self.assertEquals(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.assertEquals(1, len(select._receivers))
self.assertEquals(subselect._put, recv.notify)
select.close()
self.assertEquals(0, len(select._receivers))
self.assertEquals(subselect._put, recv.notify)
subselect.close()
self.assertEquals(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_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_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())
class IterTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_empty(self):
select = self.klass()
self.assertEquals([], list(select))
def test_nonempty(self):
recv = mitogen.core.Receiver(self.router)
select = self.klass([recv])
msg = mitogen.core.Message.pickled('123')
recv._on_receive(msg)
self.assertEquals([msg], list(select))
class OneShotTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.select.Select
def test_true_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.assertEquals(msg, msg_)
self.assertEquals(0, len(select._receivers))
self.assertEquals(None, recv.notify)
def test_false_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.assertEquals(msg, select.get())
self.assertEquals(1, len(select._receivers))
self.assertEquals(recv, select._receivers[0])
self.assertEquals(select._put, recv.notify)
class GetTest(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.assertEquals(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.assertEquals(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.assertEquals('123', msg.unpickle())
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.assertEquals('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.assertEquals(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.assertEquals('123', msg.unpickle())
self.assertRaises(mitogen.core.TimeoutError,
lambda: select.get(timeout=0.0))
if __name__ == '__main__':
unittest2.main()