MiddleMan: test_middleman.py

import Pyro.core
Pyro.config.PYRO_DETAILED_TRACEBACK = 1
Pyro.config.PYRO_TRACELEVEL = 1
Pyro.config.PYRO_USER_TRACELEVEL = 1

import middleman


class TestPyroService(Pyro.core.ObjBase):
    
    def bar(self, bacon):
        """Return the 'bacon' arg multiplied by 10."""
        return bacon * 10


class TestPyroServer(object):
    
    port = None
    service = None
    daemon = None
    running = False
    
    def __init__(self, port):
        self.port = port
        self.service = TestPyroService()
        self.running = False
    
    def start(self):
        self.thread = threading.Thread(target=self._run)
        self.thread.start()
    
    def _run(self):
        self.daemon = Pyro.core.Daemon(port=self.port)
        self.daemon.connect(self.service, 'foo')
        self.running = True
        while self.running:
            self.daemon.handleRequests()
        self.thread = None
    
    def stop(self):
        self.running = False


class TestMiddleman(object):
    
    def test_transparency(self):
        # Start up the Pyro server
        server = TestPyroServer(port=6000)
        try:
            server.start()
            
            # Start up the proxy
            tp = middleman.TransparentProxy(public=('localhost', 6500),
                                            private=('localhost', 6000))
            try:
                tp.start()
                
                # open a connection
                get_conn = Pyro.core.getProxyForURI(
                    "PYROLOC://%s:%s/foo" % (tp.public.host, tp.public.port))
                conn = get_conn()
                
                # Call the method
                result = conn.bar(3)
                assert result == 30
            finally:
                tp.stop()
        finally:
            server.stop()
    
    def test_no_route_to_host(self):
        # Start up the Pyro server
        server = TestPyroServer(port=6000)
        try:
            server.start()
            
            # Start up the proxy
            tp = middleman.TransparentProxy(
                ['MySQL', ('localhost', 6500), ('localhost', 6000)]
                )
            try:
                tp.start()
                tp.freeze('MySQL')
                
                # Open a connection
                get_conn = Pyro.core.getProxyForURI(
                    "PYROLOC://%s:%s/foo" % (tp.public.host, tp.public.port))
                try:
                    conn = get_conn()
                    
                    # Call the method
                    result = conn.bar(7)
                except RuntimeError, exc:
                    if exc.args != ["oops"]:
                        raise AssertionError("get_conn failure args %s != ['oops']" %
                                             repr(exc.args))
                    pass
                else:
                    raise AssertionError("get_conn didn't fail like it should have.")
            finally:
                tp.stop()
        finally:
            server.stop()