import os import socket import subprocess import sys import time import unittest2 import mitogen import mitogen.master import mitogen.service import mitogen.unix import testlib class MyService(mitogen.service.Service): def __init__(self, latch, **kwargs): super(MyService, self).__init__(**kwargs) # used to wake up main thread once client has made its request self.latch = latch @classmethod def name(cls): # Because this is loaded from both __main__ and whatever unit2 does, # specify a fixed name. return 'unix_test.MyService' @mitogen.service.expose(policy=mitogen.service.AllowParents()) def ping(self, msg): self.latch.put(None) return { 'src_id': msg.src_id, 'auth_id': msg.auth_id, } class IsPathDeadTest(testlib.TestCase): func = staticmethod(mitogen.unix.is_path_dead) path = '/tmp/stale-socket' def test_does_not_exist(self): self.assertTrue(self.func('/tmp/does-not-exist')) def make_socket(self): if os.path.exists(self.path): os.unlink(self.path) s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.bind(self.path) return s def test_conn_refused(self): s = self.make_socket() s.close() self.assertTrue(self.func(self.path)) def test_is_alive(self): s = self.make_socket() s.listen(5) self.assertFalse(self.func(self.path)) s.close() os.unlink(self.path) class ListenerTest(testlib.RouterMixin, testlib.TestCase): klass = mitogen.unix.Listener def test_constructor_basic(self): listener = self.klass.build_stream(router=self.router) capture = testlib.LogCapturer() capture.start() try: self.assertFalse(mitogen.unix.is_path_dead(listener.protocol.path)) os.unlink(listener.protocol.path) # ensure we catch 0 byte read error log message self.broker.shutdown() self.broker.join() self.broker_shutdown = True finally: capture.stop() class ClientTest(testlib.TestCase): klass = mitogen.unix.Listener def _try_connect(self, path): # give server a chance to setup listener timeout = time.time() + 30.0 while True: try: return mitogen.unix.connect(path) except socket.error: if time.time() > timeout: raise time.sleep(0.1) def _test_simple_client(self, path): router, context = self._try_connect(path) try: self.assertEquals(0, context.context_id) self.assertEquals(1, mitogen.context_id) self.assertEquals(0, mitogen.parent_id) resp = context.call_service(service_name=MyService, method_name='ping') self.assertEquals(mitogen.context_id, resp['src_id']) self.assertEquals(0, resp['auth_id']) finally: router.broker.shutdown() router.broker.join() os.unlink(path) @classmethod def _test_simple_server(cls, path): router = mitogen.master.Router() latch = mitogen.core.Latch() try: try: listener = cls.klass.build_stream(path=path, router=router) pool = mitogen.service.Pool(router=router, services=[ MyService(latch=latch, router=router), ]) latch.get() # give broker a chance to deliver service resopnse time.sleep(0.1) finally: pool.shutdown() pool.join() router.broker.shutdown() router.broker.join() finally: os._exit(0) def test_simple(self): path = mitogen.unix.make_socket_path() proc = subprocess.Popen( [sys.executable, __file__, 'ClientTest_server', path] ) try: self._test_simple_client(path) finally: # TODO :) mitogen.context_id = 0 mitogen.parent_id = None mitogen.parent_ids = [] proc.wait() if __name__ == '__main__': if len(sys.argv) == 3 and sys.argv[1] == 'ClientTest_server': ClientTest._test_simple_server(path=sys.argv[2]) else: unittest2.main()