os_fork: python 3 fixes and tests.

mitogen/os_fork.py

@@ -135,7 +135,7 @@ class Corker(object):
         held. This will not return until each thread acknowledges it has ceased
         execution.
         """
-        s = 'CORK' * ((128 / 4) * 1024)
+        s = mitogen.core.b('CORK') * ((128 // 4) * 1024)
         self._rsocks = []
         for pool in self.pools:
             if not pool.closed:

tests/os_fork_test.py

@@ -0,0 +1,52 @@
+
+import testlib
+import unittest2
+
+import mitogen.os_fork
+import mitogen.service
+
+
+class CorkTest(testlib.RouterMixin, testlib.TestCase):
+    klass = mitogen.os_fork.Corker
+
+    def ping(self, latch):
+        latch.put('pong')
+
+    def test_cork_broker(self):
+        latch = mitogen.core.Latch()
+        self.broker.defer(self.ping, latch)
+        self.assertEquals('pong', latch.get())
+
+        corker = self.klass(brokers=(self.broker,))
+        corker.cork()
+
+        latch = mitogen.core.Latch()
+        self.broker.defer(self.ping, latch)
+        self.assertRaises(mitogen.core.TimeoutError,
+            lambda: latch.get(timeout=0.5))
+        corker.uncork()
+        self.assertEquals('pong', latch.get())
+
+    def test_cork_pool(self):
+        pool = mitogen.service.Pool(self.router, services=(), size=4)
+        try:
+            latch = mitogen.core.Latch()
+            pool.defer(self.ping, latch)
+            self.assertEquals('pong', latch.get())
+
+            corker = self.klass(pools=(pool,))
+            corker.cork()
+
+            latch = mitogen.core.Latch()
+            pool.defer(self.ping, latch)
+            self.assertRaises(mitogen.core.TimeoutError,
+                lambda: latch.get(timeout=0.5))
+            corker.uncork()
+            self.assertEquals('pong', latch.get())
+        finally:
+            pool.stop(join=True)
+
+
+
+if __name__ == '__main__':
+    unittest2.main()