def test_task_context_relay_fork():
    endpoint1 = random_ipc_endpoint()
    endpoint2 = random_ipc_endpoint()
    srv_ctx = zerorpc.Context()
    srv_relay_ctx = zerorpc.Context()
    cli_ctx = zerorpc.Context()

    srv_tracer = Tracer("[server]")
    srv_ctx.register_middleware(srv_tracer)
    srv_relay_tracer = Tracer("[server_relay]")
    srv_relay_ctx.register_middleware(srv_relay_tracer)
    cli_tracer = Tracer("[client]")
    cli_ctx.register_middleware(cli_tracer)

    class Srv:
        def echo(self, msg):
            return msg

    srv = zerorpc.Server(Srv(), context=srv_ctx)
    srv.bind(endpoint1)
    srv_task = gevent.spawn(srv.run)

    c_relay = zerorpc.Client(context=srv_relay_ctx)
    c_relay.connect(endpoint1)

    class SrvRelay:
        def echo(self, msg):
            def dothework(msg):
                return c_relay.echo(msg) + "relayed"

            g = gevent.spawn(zerorpc.fork_task_context(dothework, srv_relay_ctx), "relay" + msg)
            print "relaying in separate task:", g
            r = g.get()
            print "back to main task"
            return r

    srv_relay = zerorpc.Server(SrvRelay(), context=srv_relay_ctx)
    srv_relay.bind(endpoint2)
    srv_relay_task = gevent.spawn(srv_relay.run)

    c = zerorpc.Client(context=cli_ctx)
    c.connect(endpoint2)

    assert c.echo("hello") == "relayhellorelayed"

    srv_relay.stop()
    srv.stop()
    srv_relay_task.join()
    srv_task.join()

    assert cli_tracer._log == [("new", cli_tracer.trace_id)]
    assert srv_relay_tracer._log == [
        ("load", cli_tracer.trace_id),
        ("reuse", cli_tracer.trace_id),
        ("load", cli_tracer.trace_id),
        ("reuse", cli_tracer.trace_id),
        ("reuse", cli_tracer.trace_id),
    ]
    assert srv_tracer._log == [("load", cli_tracer.trace_id), ("reuse", cli_tracer.trace_id)]

def test_task_context_relay():
    endpoint1 = random_ipc_endpoint()
    endpoint2 = random_ipc_endpoint()
    srv_ctx = zerorpc.Context()
    srv_relay_ctx = zerorpc.Context()
    cli_ctx = zerorpc.Context()

    srv_tracer = Tracer('[server]')
    srv_ctx.register_middleware(srv_tracer)
    srv_relay_tracer = Tracer('[server_relay]')
    srv_relay_ctx.register_middleware(srv_relay_tracer)
    cli_tracer = Tracer('[client]')
    cli_ctx.register_middleware(cli_tracer)

    class Srv:
        def echo(self, msg):
            return msg

    srv = zerorpc.Server(Srv(), context=srv_ctx)
    srv.bind(endpoint1)
    srv_task = gevent.spawn(srv.run)

    c_relay = zerorpc.Client(context=srv_relay_ctx)
    c_relay.connect(endpoint1)

    class SrvRelay:
        def echo(self, msg):
            return c_relay.echo('relay' + msg) + 'relayed'

    srv_relay = zerorpc.Server(SrvRelay(), context=srv_relay_ctx)
    srv_relay.bind(endpoint2)
    srv_relay_task = gevent.spawn(srv_relay.run)

    c = zerorpc.Client(context=cli_ctx)
    c.connect(endpoint2)

    assert c.echo('hello') == 'relayhellorelayed'

    srv_relay.stop()
    srv.stop()
    srv_relay_task.join()
    srv_task.join()

    assert cli_tracer._log == [
            ('new', cli_tracer.trace_id),
            ]
    assert srv_relay_tracer._log == [
            ('load', cli_tracer.trace_id),
            ('reuse', cli_tracer.trace_id),
            ('reuse', cli_tracer.trace_id),
            ]
    assert srv_tracer._log == [
            ('load', cli_tracer.trace_id),
            ('reuse', cli_tracer.trace_id),
            ]

def test_config_multiple_endpoints():
    app = Zask(__name__)
    endpoint = random_ipc_endpoint()
    another_endpoint = random_ipc_endpoint()

    app.config['ZERORPC_SOME_SERVICE'] = {
        '1.0': endpoint,
    }
    app.config['ZERORPC_SOME_SERVICE_2'] = {
        '1.0': another_endpoint
    }
    app.config['ZERORPC_SOME_SERVICE_CLIENT'] = {
        '1.0': [
            endpoint,
            another_endpoint
        ],
        'default': '1.0'
    }
    rpc = ZeroRPC(app, middlewares=[CONFIG_ENDPOINT_MIDDLEWARE])

    class Srv(object):
        __version__ = "1.0"
        __service_name__ = "some_service"

        def hello(self):
            return 'i am server 1'

    class AnotherSrv(object):
        __version__ = "1.0"
        __service_name__ = "some_service_2"

        def hello(self):
            return 'i am server 2'

    srv = rpc.Server(Srv())
    another_srv = rpc.Server(AnotherSrv())
    gevent.spawn(srv.run)
    gevent.spawn(another_srv.run)

    client = rpc.Client('some_service_client')
    for i in range(5):
        who_i_am = client.hello()
        app.logger.debug(who_i_am)
        assert who_i_am == 'i am server 1' or who_i_am == 'i am server 2'

    with pytest.raises(MissingConfigException):
        client = rpc.Client('some_service_client', version='2.0')

    client.close()
    srv.close()
    another_srv.close()

def test_heartbeat_can_open_channel_server_close():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.ROUTER)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.DEALER)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events, ignore_broadcast=True)

    client_channel = client.channel()
    client_hbchan = zerorpc.HeartBeatOnChannel(client_channel, freq=TIME_FACTOR * 2)
    client_bufchan = zerorpc.BufferedChannel(client_hbchan)

    event = server.recv()
    server_channel = server.channel(event)
    server_hbchan = zerorpc.HeartBeatOnChannel(server_channel, freq=TIME_FACTOR * 2)
    server_bufchan = zerorpc.BufferedChannel(server_hbchan)

    gevent.sleep(TIME_FACTOR * 3)
    print 'CLOSE SERVER SOCKET!!!'
    server_bufchan.close()
    if sys.version_info < (2, 7):
        assert_raises(zerorpc.LostRemote, client_bufchan.recv)
    else:
        with assert_raises(zerorpc.LostRemote):
            client_bufchan.recv()
    print 'CLIENT LOST SERVER :)'
    client_bufchan.close()
    server.close()
    client.close()

def test_events_channel_both_side():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.XREP)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.XREQ)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events)

    client_channel = client.channel()
    client_channel.emit("openthat", (42,))

    event = server.recv()
    print event
    assert event.args == (42,)
    assert event.name == "openthat"

    server_channel = server.channel(event)
    server_channel.emit("test", (21,))

    event = client_channel.recv()
    assert event.args == (21,)
    assert event.name == "test"

    server_channel.emit("test", (22,))

    event = client_channel.recv()
    assert event.args == (22,)
    assert event.name == "test"

    server_events.close()
    server_channel.close()
    client_channel.close()
    client_events.close()

def test1():
    endpoint = random_ipc_endpoint()
    def server():
        c = zmq.Context()
        s = c.socket(zmq.REP)
        s.bind(endpoint)
        while True:
            print 'srv recving...'
            r = s.recv()
            print 'srv', r
            print 'srv sending...'
            s.send('world')

        s.close()
        c.term()

    def client():
        c = zmq.Context()
        s = c.socket(zmq.REQ)
        s.connect(endpoint)

        print 'cli sending...'
        s.send('hello')
        print 'cli recving...'
        r = s.recv()
        print 'cli', r

        s.close()
        c.term()

    s = gevent.spawn(server)
    c = gevent.spawn(client)
    c.join()

def test_close_client_hbchan():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.ROUTER)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.DEALER)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events, ignore_broadcast=True)

    client_channel = client.channel()
    client_hbchan = zerorpc.HeartBeatOnChannel(client_channel, freq=2)
    client_hbchan.emit('openthat', None)

    event = server.recv()
    server_channel = server.channel(event)
    server_hbchan = zerorpc.HeartBeatOnChannel(server_channel, freq=2)
    server_hbchan.recv()

    gevent.sleep(3)
    print 'CLOSE CLIENT SOCKET!!!'
    client_hbchan.close()
    if sys.version_info < (2, 7):
        assert_raises(zerorpc.LostRemote, server_hbchan.recv)
    else:
        with assert_raises(zerorpc.LostRemote):
            server_hbchan.recv()
    print 'SERVER LOST CLIENT :)'
    server_hbchan.close()
    server.close()
    client.close()

def test_hook_client_before_request():

    class ClientBeforeRequestMiddleware(object):
        def __init__(self):
            self.called = False
        def client_before_request(self, event):
            self.called = True
            self.method = event.name

    zero_ctx = zerorpc.Context()
    endpoint = random_ipc_endpoint()

    test_server = zerorpc.Server(EchoModule(), context=zero_ctx)
    test_server.bind(endpoint)
    test_server_task = gevent.spawn(test_server.run)

    test_client = zerorpc.Client(context=zero_ctx)
    test_client.connect(endpoint)

    assert test_client.echo("test") == "echo: test"

    test_middleware = ClientBeforeRequestMiddleware()
    zero_ctx.register_middleware(test_middleware)
    assert test_middleware.called == False
    assert test_client.echo("test") == "echo: test"
    assert test_middleware.called == True
    assert test_middleware.method == 'echo'

    test_server.stop()
    test_server_task.join()

def test_events_channel_client_side_server_send_many():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.ROUTER)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.DEALER)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events)

    client_channel = client.channel()
    client_channel.emit('giveme', (10,))

    event = server.recv()
    print event
    assert list(event.args) == [10]
    assert event.header.get('zmqid', None) is not None

    for x in xrange(10):
        server.emit('someanswer', (x,),
                xheader=dict(response_to=event.header['message_id'],
                    zmqid=event.header['zmqid']))
    for x in xrange(10):
        event = client_channel.recv()
        assert list(event.args) == [x]

def test_server_manual():
    endpoint = random_ipc_endpoint()

    class MySrv(zerorpc.Server):

        def lolita(self):
            return 42

        def add(self, a, b):
            return a + b

    srv = MySrv()
    srv.bind(endpoint)
    gevent.spawn(srv.run)

    client_events = zerorpc.Events(zmq.XREQ)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events, ignore_broadcast=True)

    client_channel = client.channel()
    client_channel.emit('lolita', tuple())
    event = client_channel.recv()
    assert event.args == (42,)
    client_channel.close()

    client_channel = client.channel()
    client_channel.emit('add', (1, 2))
    event = client_channel.recv()
    assert event.args == (3,)
    client_channel.close()
    srv.stop()

def test_client_server_exception():
    endpoint = random_ipc_endpoint()

    class MySrv(zerorpc.Server):

        def raise_something(self, a):
            return a[4]

    srv = MySrv()
    srv.bind(endpoint)
    gevent.spawn(srv.run)

    client = zerorpc.Client(timeout=2)
    client.connect(endpoint)

    if sys.version_info < (2, 7):
        def _do_with_assert_raises():
            print client.raise_something(42)
        assert_raises(zerorpc.RemoteError, _do_with_assert_raises)
    else:
        with assert_raises(zerorpc.RemoteError):
            print client.raise_something(42)
    assert client.raise_something(range(5)) == 4
    client.close()
    srv.close()

def test_sub_events():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.ROUTER)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.DEALER)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events, ignore_broadcast=True)

    client_channel = client.channel()
    client_channel_events = zerorpc.WrappedEvents(client_channel)
    client_channel_events.emit("coucou", 42)

    event = server.recv()
    print event
    assert type(event.args) is tuple
    assert event.name == "w"
    subevent = event.args
    print "subevent:", subevent
    server_channel = server.channel(event)
    server_channel_events = zerorpc.WrappedEvents(server_channel)
    server_channel_channel = zerorpc.ChannelMultiplexer(server_channel_events)
    event = server_channel_channel.recv()
    print event
    assert event.name == "coucou"
    assert event.args == 42

    server_events.close()
    client_events.close()

def test_hook_server_before_exec_puller():
    zero_ctx = zerorpc.Context()
    trigger = gevent.event.Event()
    endpoint = random_ipc_endpoint()

    echo_module = EchoModule(trigger)
    test_server = zerorpc.Puller(echo_module, context=zero_ctx)
    test_server.bind(endpoint)
    test_server_task = gevent.spawn(test_server.run)
    test_client = zerorpc.Pusher()
    test_client.connect(endpoint)

    # Test without a middleware
    test_client.echo("test")
    trigger.wait(timeout=TIME_FACTOR * 2)
    assert echo_module.last_msg == "echo: test"
    trigger.clear()

    # Test with a middleware
    test_middleware = ServerBeforeExecMiddleware()
    zero_ctx.register_middleware(test_middleware)
    assert test_middleware.called == False
    test_client.echo("test with a middleware")
    trigger.wait(timeout=TIME_FACTOR * 2)
    assert echo_module.last_msg == "echo: test with a middleware"
    assert test_middleware.called == True

    test_server.stop()
    test_server_task.join()

def test_hook_server_after_exec_on_error_puller():
    zero_ctx = zerorpc.Context()
    trigger = gevent.event.Event()
    endpoint = random_ipc_endpoint()

    echo_module = BrokenEchoModule(trigger)
    test_server = zerorpc.Puller(echo_module, context=zero_ctx)
    test_server.bind(endpoint)
    test_server_task = gevent.spawn(test_server.run)
    test_client = zerorpc.Pusher()
    test_client.connect(endpoint)

    test_middleware = ServerAfterExecMiddleware()
    zero_ctx.register_middleware(test_middleware)
    assert test_middleware.called == False
    try:
        test_client.echo("test with a middleware")
        trigger.wait(timeout=TIME_FACTOR * 2)
    except zerorpc.RemoteError:
        pass
    assert echo_module.last_msg == "Raise"
    assert test_middleware.called == False

    test_server.stop()
    test_server_task.join()

def test_hook_server_after_exec_stream():
    zero_ctx = zerorpc.Context()
    endpoint = random_ipc_endpoint()

    test_server = zerorpc.Server(EchoModule(), context=zero_ctx)
    test_server.bind(endpoint)
    test_server_task = gevent.spawn(test_server.run)
    test_client = zerorpc.Client()
    test_client.connect(endpoint)

    # Test without a middleware
    for echo in test_client.echoes("test"):
        assert echo == "echo: test"

    # Test with a middleware
    test_middleware = ServerAfterExecMiddleware()
    zero_ctx.register_middleware(test_middleware)
    assert test_middleware.called == False
    it = test_client.echoes("test")
    assert next(it) == "echo: test"
    assert test_middleware.called == False
    for echo in it:
        assert echo == "echo: test"
    assert test_middleware.called == True
    assert test_middleware.request_event_name == 'echoes'
    assert test_middleware.reply_event_name == 'STREAM_DONE'

    test_server.stop()
    test_server_task.join()

def test_events_channel_both_side():
    endpoint = random_ipc_endpoint()
    server_events = zerorpc.Events(zmq.ROUTER)
    server_events.bind(endpoint)
    server = zerorpc.ChannelMultiplexer(server_events)

    client_events = zerorpc.Events(zmq.DEALER)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events)

    client_channel = client.channel()
    client_channel.emit('openthat', (42,))

    event = server.recv()
    print event
    assert list(event.args) == [42]
    assert event.name == 'openthat'

    server_channel = server.channel(event)
    server_channel.emit('test', (21,))

    event = client_channel.recv()
    assert list(event.args) == [21]
    assert event.name == 'test'

    server_channel.emit('test', (22,))

    event = client_channel.recv()
    assert list(event.args) == [22]
    assert event.name == 'test'

    server_events.close()
    server_channel.close()
    client_channel.close()
    client_events.close()

def test_client_server_detailed_exception():
    endpoint = random_ipc_endpoint()

    class MySrv(zerorpc.Server):

        def raise_error(self):
            raise RuntimeError('oops!')

    srv = MySrv()
    srv.bind(endpoint)
    gevent.spawn(srv.run)

    client = zerorpc.Client(timeout=2)
    client.connect(endpoint)

    with assert_raises(zerorpc.RemoteError):
        print client.raise_error()
    try:
        client.raise_error()
    except zerorpc.RemoteError as e:
        print 'got that:', e
        print 'name', e.name
        print 'msg', e.msg
        assert e.name == 'RuntimeError'
        assert e.msg == 'oops!'

    client.close()
    srv.close()

def test_resolve_endpoint_events():
    test_endpoint = random_ipc_endpoint()
    c = zerorpc.Context()

    class Resolver():
        def resolve_endpoint(self, endpoint):
            if endpoint == 'some_service':
                return test_endpoint
            return endpoint

    class Srv(zerorpc.Server):
        def hello(self):
            print 'heee'
            return 'world'

    srv = Srv(heartbeat=TIME_FACTOR * 1, context=c)
    if sys.version_info < (2, 7):
        assert_raises(zmq.ZMQError, srv.bind, 'some_service')
    else:
        with assert_raises(zmq.ZMQError):
            srv.bind('some_service')

    cnt = c.register_middleware(Resolver())
    assert cnt == 1
    srv.bind('some_service')
    gevent.spawn(srv.run)

    client = zerorpc.Client(heartbeat=TIME_FACTOR * 1, context=c)
    client.connect('some_service')
    assert client.hello() == 'world'

    client.close()
    srv.close()

def test_exception_compat_v1():
    endpoint = random_ipc_endpoint()

    class MySrv(zerorpc.Server):
        pass

    srv = MySrv()
    srv.bind(endpoint)
    gevent.spawn(srv.run)

    client_events = zerorpc.Events(zmq.XREQ)
    client_events.connect(endpoint)
    client = zerorpc.ChannelMultiplexer(client_events, ignore_broadcast=True)

    rpccall = client.channel()
    rpccall.emit('donotexist', tuple())
    event = rpccall.recv()
    print event
    assert event.name == 'ERR'
    (name, msg, tb) = event.args
    print 'detailed error', name, msg, tb
    assert name == 'NameError'
    assert msg == 'donotexist'

    rpccall = client.channel()
    rpccall.emit('donotexist', tuple(), xheader=dict(v=1))
    event = rpccall.recv()
    print event
    assert event.name == 'ERR'
    (msg,) = event.args
    print 'msg only', msg
    assert msg == "NameError('donotexist',)"

    client_events.close()
    srv.close()

def test_hook_client_after_request_timeout():

    class ClientAfterRequestMiddleware(object):
        def __init__(self):
            self.called = False
        def client_after_request(self, req_event, rep_event, exception):
            self.called = True
            assert req_event is not None
            assert req_event.name == "timeout"
            assert rep_event is None

    zero_ctx = zerorpc.Context()
    test_middleware = ClientAfterRequestMiddleware()
    zero_ctx.register_middleware(test_middleware)
    endpoint = random_ipc_endpoint()

    test_server = zerorpc.Server(EchoModule(), context=zero_ctx)
    test_server.bind(endpoint)
    test_server_task = gevent.spawn(test_server.run)

    test_client = zerorpc.Client(timeout=TIME_FACTOR * 1, context=zero_ctx)
    test_client.connect(endpoint)

    assert test_middleware.called == False
    try:
        test_client.timeout("test")
    except zerorpc.TimeoutExpired as ex:
        assert test_middleware.called == True
        assert "timeout" in ex.args[0]

    test_server.stop()
    test_server_task.join()