mitogen/tests/os_fork_test.py

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()