issue #446: basic tests for Receiver._on_receive().

tests/receiver_test.py

@@ -76,5 +76,54 @@ class CloseTest(testlib.RouterMixin, testlib.TestCase):
             t.join()
 
 
+class OnReceiveTest(testlib.RouterMixin, testlib.TestCase):
+    # Verify behaviour of _on_receive dead message handling. A dead message
+    # should unregister the receiver and wake all threads.
+
+    def wait(self, latch, wait_recv):
+        try:
+            latch.put(wait_recv.get())
+        except Exception:
+            latch.put(sys.exc_info()[1])
+
+    def test_sender_closes_one_thread(self):
+        latch = mitogen.core.Latch()
+        wait_recv = mitogen.core.Receiver(self.router)
+        t = threading.Thread(target=lambda: self.wait(latch, wait_recv))
+        t.start()
+        sender = wait_recv.to_sender()
+        sender.close()
+        def throw():
+            raise latch.get()
+        t.join()
+        e = self.assertRaises(mitogen.core.ChannelError, throw)
+        self.assertEquals(e.args[0], sender.explicit_close_msg)
+
+    @unittest2.skip(reason=(
+        'Unclear if a asingle dead message received from remote should '
+        'cause all threads to wake up.'
+    ))
+    def test_sender_closes_all_threads(self):
+        latch = mitogen.core.Latch()
+        wait_recv = mitogen.core.Receiver(self.router)
+        ts = [
+            threading.Thread(target=lambda: self.wait(latch, wait_recv))
+            for x in range(5)
+        ]
+        for t in ts:
+            t.start()
+        sender = wait_recv.to_sender()
+        sender.close()
+        def throw():
+            raise latch.get()
+        for x in range(5):
+            e = self.assertRaises(mitogen.core.ChannelError, throw)
+            self.assertEquals(e.args[0], mitogen.core.Receiver.closed_msg)
+        for t in ts:
+            t.join()
+
+    # TODO: what happens to a Select subscribed to the receiver in this case?
+
+
 if __name__ == '__main__':
     unittest2.main()