def test_normal(self, publisher_container, consumer_container, tracker):

        # call 1 succeeds
        payload1 = "payload1"
        with entrypoint_waiter(consumer_container, 'recv'):
            with entrypoint_hook(publisher_container, 'send') as send:
                send(payload1)

        assert tracker.call_args_list == [
            call("send", payload1),
            call("recv", payload1),
        ]

        # call 2 succeeds
        payload2 = "payload2"
        with entrypoint_waiter(consumer_container, 'recv'):
            with entrypoint_hook(publisher_container, 'send') as send:
                send(payload2)

        assert tracker.call_args_list == [
            call("send", payload1),
            call("recv", payload1),
            call("send", payload2),
            call("recv", payload2),
        ]

    def test_reuse_when_down(
        self, publisher_container, consumer_container, tracker, toxiproxy,
    ):
        """ Verify we detect stale connections.

        Publish confirms are required for this functionality. Without confirms
        the later messages are silently lost and the test hangs waiting for a
        response.
        """
        # call 1 succeeds
        payload1 = "payload1"
        with entrypoint_waiter(consumer_container, 'recv'):
            with entrypoint_hook(publisher_container, 'send') as send:
                send(payload1)

        assert tracker.call_args_list == [
            call("send", payload1),
            call("recv", payload1),
        ]

        with toxiproxy.disabled():

            # call 2 fails
            payload2 = "payload2"
            with pytest.raises(IOError):
                with entrypoint_hook(publisher_container, 'send') as send:
                    send(payload2)

            assert tracker.call_args_list == [
                call("send", payload1),
                call("recv", payload1),
                call("send", payload2),
            ]

def test_end_to_end(container_factory, tmpdir):

    # create a temporary database
    db_uri = 'sqlite:///{}'.format(tmpdir.join("db").strpath)
    engine = create_engine(db_uri)
    ExampleModel.metadata.create_all(engine)

    config = {
        DB_URIS_KEY: {
            'exampleservice:examplebase': db_uri
        }
    }

    container = container_factory(ExampleService, config)
    container.start()

    # write through the service
    with entrypoint_hook(container, "write") as write:
        pk = write("foobar")

    # verify changes written to disk
    entries = list(engine.execute('SELECT data FROM example LIMIT 1'))
    assert entries == [('foobar',)]

    # read through the service
    with entrypoint_hook(container, "read") as read:
        assert read(pk) == "foobar"

    def test_reuse_when_recovered(
        self, publisher_container, consumer_container, tracker, toxiproxy
    ):
        """ Verify we detect and recover from stale connections.

        Publish confirms are required for this functionality. Without confirms
        the later messages are silently lost and the test hangs waiting for a
        response.
        """
        # call 1 succeeds
        payload1 = "payload1"
        with entrypoint_waiter(consumer_container, 'recv'):
            with entrypoint_hook(publisher_container, 'send') as send:
                send(payload1)

        assert tracker.call_args_list == [
            call("send", payload1),
            call("recv", payload1),
        ]

        with toxiproxy.disabled():

            # call 2 fails
            payload2 = "payload2"
            with pytest.raises(IOError) as exc_info:
                with entrypoint_hook(publisher_container, 'send') as send:
                    send(payload2)
            assert (
                # expect the write to raise a BrokenPipe or, if it succeeds,
                # the socket to be closed on the subsequent confirmation read
                "Broken pipe" in str(exc_info.value) or
                "Socket closed" in str(exc_info.value)
            )

            assert tracker.call_args_list == [
                call("send", payload1),
                call("recv", payload1),
                call("send", payload2),
            ]

        # call 3 succeeds
        payload3 = "payload3"
        with entrypoint_waiter(consumer_container, 'recv'):
            with entrypoint_hook(publisher_container, 'send') as send:
                send(payload3)

        assert tracker.call_args_list == [
            call("send", payload1),
            call("recv", payload1),
            call("send", payload2),
            call("send", payload3),
            call("recv", payload3),
        ]

def test_unsubscribe(container, websocket):
    ws = websocket()
    ws.rpc('subscribe')
    with entrypoint_hook(container, 'broadcast') as broadcast:
        broadcast(value=42)
    assert get_message(ws) == 42
    ws.rpc('unsubscribe')

    with entrypoint_hook(container, 'broadcast') as broadcast:
        broadcast(value=42)
    with eventlet.Timeout(.1, exception=False):
        assert get_message(ws) == 42

def test_entrypoint_hook_with_return(runner_factory, rabbit_config):

    service_classes = (Service, ServiceA, ServiceB, ServiceC)
    runner = runner_factory(rabbit_config, *service_classes)
    runner.start()

    service_container = get_container(runner, Service)

    with entrypoint_hook(service_container, "working") as working:
        assert working("value") == "value-a-b-c"

    with entrypoint_hook(service_container, "broken") as broken:
        with pytest.raises(ExampleError):
            broken("value")

def test_close(container, websocket):
    ws = websocket()

    ws.rpc('subscribe')

    with entrypoint_hook(container, 'list_subscribers') as list_subscribers:
        subscribers1 = list_subscribers()
        assert subscribers1['test_channel']

    ws.app.close()

    with entrypoint_hook(container, 'list_subscribers') as list_subscribers:
        subscribers2 = list_subscribers()
        assert not subscribers2['test_channel']

    def it_should_allow_method_calls_to_beanstalkd(self, bs):
        container = ServiceContainer(ExampleService, {})
        container.start()

        with entrypoint_hook(container, "write") as write:
            write("foobar")

        assert bs.Connection.return_value.put.called
        assert bs.Connection.return_value.put.call_args[0][0] == 'foobar'

        bs.Connection.return_value.reserve.return_value = MockJob('foobar')

        with entrypoint_hook(container, "read") as read:
            assert read() == "foobar"

    def test_expected_exceptions(self, container_factory):

        exceptions = defaultdict(list)

        class CustomException(Exception):
            pass

        class Logger(DependencyProvider):
            """ Example DependencyProvider that interprets
            ``expected_exceptions`` on an entrypoint
            """

            def worker_result(self, worker_ctx, result=None, exc_info=None):
                if exc_info is None:  # nothing to do
                    return  # pragma: no cover

                exc = exc_info[1]
                expected = worker_ctx.entrypoint.expected_exceptions

                if isinstance(exc, expected):
                    exceptions['expected'].append(exc)
                else:
                    exceptions['unexpected'].append(exc)

        class Service(object):
            name = "service"

            logger = Logger()

            @dummy(expected_exceptions=CustomException)
            def expected(self):
                raise CustomException()

            @dummy
            def unexpected(self):
                raise CustomException()

        container = container_factory(Service, {})
        container.start()

        with entrypoint_hook(container, 'expected') as hook:
            with pytest.raises(CustomException) as expected_exc:
                hook()
        assert expected_exc.value in exceptions['expected']

        with entrypoint_hook(container, 'unexpected') as hook:
            with pytest.raises(CustomException) as unexpected_exc:
                hook()
        assert unexpected_exc.value in exceptions['unexpected']

def test_pub_sub(container, websocket):
    ws = websocket()
    assert ws.rpc('subscribe') == 'subscribed!'
    with entrypoint_hook(container, 'broadcast') as broadcast:
        broadcast(value=42)

    assert get_message(ws) == 42

def test_unicast(container, websocket):
    ws = websocket()
    _, connected_data = ws.wait_for_event('connected')
    socket_id = connected_data['socket_id']
    with entrypoint_hook(container, 'unicast') as unicast:
        assert unicast(target_socket_id=socket_id, value=42)
    assert get_message(ws) == 42

def test_expected_exceptions_integration(container_factory, rabbit_config):
    container = container_factory(ExampleService, rabbit_config)
    container.start()

    worker_logger = get_extension(container, WorkerErrorLogger)

    with entrypoint_hook(container, 'broken') as broken:
        with pytest.raises(ExampleError):
            broken()

    with entrypoint_hook(container, 'very_broken') as very_broken:
        with pytest.raises(AttributeError):
            very_broken()

    assert worker_logger.expected == {'broken': ExampleError}
    assert worker_logger.unexpected == {'very_broken': AttributeError}

def test_proxy_disconnect_with_active_worker(container_factory,
                                             rabbit_manager, rabbit_config):
    """ Break the connection to rabbit while a service's queue consumer and
    rabbit while the service has an in-flight rpc request (i.e. it is waiting
    on a reply).
    """
    # ExampleService is the target; ProxyService has the rpc_proxy;
    proxy_container = container_factory(ProxyService, rabbit_config)
    example_container = container_factory(ExampleService, rabbit_config)

    proxy_container.start()

    # get proxyservice's queue consumer connection while we know it's the
    # only active connection
    vhost = rabbit_config['vhost']
    connections = get_rabbit_connections(vhost, rabbit_manager)
    assert len(connections) == 1
    proxy_consumer_conn = connections[0]['name']

    example_container.start()

    # there should now be two connections:
    # 1. the queue consumer from proxyservice
    # 2. the queue consumer from exampleservice
    connections = get_rabbit_connections(vhost, rabbit_manager)
    assert len(connections) == 2

    # disconnect proxyservice's queue consumer while its request is in-flight
    eventlet.spawn(disconnect_on_event, rabbit_manager, proxy_consumer_conn)
    with entrypoint_hook(proxy_container, 'entrypoint') as entrypoint:
        # we should receive a response after reconnection
        assert entrypoint('hello') == 'hello'

    connections = get_rabbit_connections(vhost, rabbit_manager)
    assert proxy_consumer_conn not in [conn['name'] for conn in connections]

    def test_event_dispatcher_over_ssl(
        self, container_factory, rabbit_ssl_config, rabbit_config
    ):
        class Dispatcher(object):
            name = "dispatch"

            dispatch = EventDispatcher()

            @dummy
            def method(self, payload):
                return self.dispatch("event-type", payload)

        class Handler(object):
            name = "handler"

            @event_handler("dispatch", "event-type")
            def echo(self, payload):
                return payload

        dispatcher = container_factory(Dispatcher, rabbit_ssl_config)
        dispatcher.start()

        handler = container_factory(Handler, rabbit_config)
        handler.start()

        with entrypoint_waiter(handler, 'echo') as result:
            with entrypoint_hook(dispatcher, 'method') as dispatch:
                dispatch("payload")
        assert result.get() == "payload"

def test_async_rpc(container_factory, rabbit_config):

    container = container_factory(ExampleService, rabbit_config)
    container.start()

    with entrypoint_hook(container, 'call_async') as call_async:
        assert call_async() == ["result_b", "result_a", [[], {}]]

    def test_publisher_over_ssl(
        self, container_factory, rabbit_ssl_config, rabbit_config, queue
    ):
        class PublisherService(object):
            name = "publisher"

            publish = Publisher()

            @dummy
            def method(self, payload):
                return self.publish(payload, routing_key=queue.name)

        class ConsumerService(object):
            name = "consumer"

            @consume(queue)
            def echo(self, payload):
                return payload

        publisher = container_factory(PublisherService, rabbit_ssl_config)
        publisher.start()

        consumer = container_factory(ConsumerService, rabbit_config)
        consumer.start()

        with entrypoint_waiter(consumer, 'echo') as result:
            with entrypoint_hook(publisher, 'method') as publish:
                publish("payload")
        assert result.get() == "payload"

def test_hello_languages(language, greeting, container_factory, rabbit_config):

    container = container_factory(HelloService, rabbit_config)
    container.start()

    context_data = {'language': language}
    with entrypoint_hook(container, 'hello', context_data) as hook:
        assert hook("Matt") == "{}, Matt!".format(greeting)

def test_entrypoint_hook_context_data(container_factory, rabbit_config, context_data):

    container = container_factory(Service, rabbit_config)
    container.start()

    method = "get_language"
    with entrypoint_hook(container, method, context_data) as get_language:
        assert get_language() == context_data["language"]

def test_rpc_context_data(container_factory, rabbit_config):

    container = container_factory(ExampleService, rabbit_config)
    container.start()

    context_data = {
        'language': 'en',
        'auth_token': '123456789'
    }

    with entrypoint_hook(container, 'say_hello', context_data) as say_hello:
        assert say_hello() == "hello"

    context_data['language'] = 'fr'

    with entrypoint_hook(container, 'say_hello', context_data) as say_hello:
        assert say_hello() == "bonjour"

def test_expected_exceptions_integration(container_factory, rabbit_config):
    container = container_factory(ExampleService, rabbit_config)
    container.start()

    worker_logger = get_dependency(container, WorkerErrorLogger)

    with entrypoint_hook(container, 'broken') as broken:
        with pytest.raises(ExampleError):
            broken()

    with entrypoint_hook(container, 'very_broken') as very_broken:
        with pytest.raises(AttributeError):
            very_broken()

    wait_for_worker_idle(container)  # wait for worker lifecycle to complete
    assert worker_logger.expected == {'broken': ExampleError}
    assert worker_logger.unexpected == {'very_broken': AttributeError}