import unittest2 import mitogen.select import testlib class BoolTest(testlib.RouterMixin, testlib.TestCase): klass = mitogen.select.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.assertEquals('123', select.get().unpickle()) self.assertFalse(select) 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()