def test_runner_with_duplicate_services(runner_factory, rabbit_config):

    # host Service multiple times
    runner = runner_factory(rabbit_config)
    runner.add_service(Service)
    runner.add_service(Service)  # no-op
    runner.start()

    # it should only be hosted once
    assert len(runner.containers) == 1

    # test events (only one service is hosted)
    event_data = "msg"
    with event_dispatcher('srcservice', rabbit_config) as dispatch:
        dispatch(TestEvent(event_data))

    with eventlet.Timeout(1):
        while len(received) == 0:
            eventlet.sleep()

        assert received == [event_data]

    # test rpc
    arg = "msg"
    del received[:]

    with RpcProxy("service", rabbit_config) as proxy:
        proxy.handle(arg)

    with eventlet.Timeout(1):
        while len(received) == 0:
            eventlet.sleep()

        assert received == [arg]

def test_service_disconnect_with_active_async_worker(
        container_factory, rabbit_manager, rabbit_config):
    """ Break the connection between a service's queue consumer and rabbit
    while the service has an active async worker (e.g. event handler).
    """
    container = container_factory(ExampleService, rabbit_config)
    container.start()

    # get the service'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
    queue_consumer_conn = connections[0]['name']

    # disconnect the service's queue consumer while it's running the worker
    eventlet.spawn(disconnect_on_event, rabbit_manager, queue_consumer_conn)

    # dispatch an event
    data = uuid.uuid4().hex
    dispatch = event_dispatcher(rabbit_config)
    dispatch('srcservice', 'exampleevent', data)

    # `handle` will have been called twice with the same the `data`, because
    # rabbit will have redelivered the un-ack'd message from the first call
    def event_handled_twice():
        assert handle_called.call_args_list == [call(data), call(data)]
    assert_stops_raising(event_handled_twice)

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

def test_entrypoint_waiter(container_factory, rabbit_config):
    container = container_factory(Service, rabbit_config)
    container.start()

    dispatch = event_dispatcher(rabbit_config)
    with entrypoint_waiter(container, "handle"):
        dispatch("srcservice", "eventtype", "")

def test_runner_with_duplicate_services(
    runner_factory, rabbit_config, service_cls, tracker
):

    # host Service multiple times
    runner = runner_factory(rabbit_config)
    runner.add_service(service_cls)
    runner.add_service(service_cls)  # no-op
    runner.start()

    # it should only be hosted once
    assert len(runner.containers) == 1
    container = list(runner.containers)[0]

    # test events (only one service is hosted)
    event_data = "event"
    dispatch = event_dispatcher(rabbit_config)

    with entrypoint_waiter(container, "handle"):
        dispatch('srcservice', "testevent", event_data)
    assert tracker.call_args_list == [call(event_data)]

    # test rpc
    arg = "arg"
    with ServiceRpcProxy("service", rabbit_config) as proxy:
        proxy.handle(arg)

    assert tracker.call_args_list == [call(event_data), call(arg)]

    def test_deadlock_due_to_slow_workers(
        self, service_cls, container_factory, config
    ):
        """ Deadlock will occur if the unack'd messages grows beyond the
        size of the worker pool at any point. The QueueConsumer will block
        waiting for a worker and pending RPC replies will not be ack'd.
        Any running workers therefore never complete, and the worker pool
        remains exhausted.
        """
        container = container_factory(service_cls, config)
        container.start()

        count = 2

        dispatch = event_dispatcher(config)
        for _ in range(count):
            dispatch("service", "event1", 1)
            dispatch("service", "event2", 1)

        counter = itertools.count(start=1)

        def cb(worker_ctx, res, exc_info):
            if next(counter) == count:
                return True

        with entrypoint_waiter(
            container, 'handle_event1', timeout=5, callback=cb
        ):
            pass

def test_entrypoint_waiter(container_factory, rabbit_config):
    container = container_factory(Service, rabbit_config)
    container.start()

    class ExampleEvent(Event):
        type = "eventtype"

    with event_dispatcher('srcservice', rabbit_config) as dispatch:
        with entrypoint_waiter(container, 'handle'):
            dispatch(ExampleEvent(""))

    def get_dependency(self, worker_ctx):
        """ Inject a dispatch method onto the service instance
        """
        headers = self.get_message_headers(worker_ctx)
        kwargs = self.kwargs
        dispatcher = event_dispatcher(self.config, headers=headers, **kwargs)

        def dispatch(event_type, event_data):
            dispatcher(self.service_name, event_type, event_data)
        return dispatch

def test_restrict_entrypoints(container_factory, rabbit_config):

    method_called = Mock()

    class OnceProvider(EntrypointProvider):
        """ Entrypoint that spawns a worker exactly once, as soon as
        the service container started.
        """
        def __init__(self, *args, **kwargs):
            self.args = args
            self.kwargs = kwargs

        def start(self):
            self.container.spawn_worker(self, self.args, self.kwargs)

    @entrypoint
    def once(*args, **kwargs):
        return DependencyFactory(OnceProvider, args, kwargs)

    class ExampleEvent(Event):
        type = "eventtype"

    class Service(object):

        @rpc
        @once("assert not seen")
        def handler_one(self, arg):
            method_called(arg)

        @event_handler('srcservice', 'eventtype')
        def handler_two(self, msg):
            method_called(msg)

    container = container_factory(Service, rabbit_config)

    # disable the entrypoints on handler_one
    restrict_entrypoints(container, "handler_two")
    container.start()

    # verify the rpc entrypoint on handler_one is disabled
    with RpcProxy("service", rabbit_config) as service_proxy:
        with pytest.raises(MethodNotFound) as exc_info:
            service_proxy.handler_one("msg")
        assert exc_info.value.message == "handler_one"

    # dispatch an event to handler_two
    msg = "msg"
    with event_dispatcher('srcservice', rabbit_config) as dispatch:

        with entrypoint_waiter(container, 'handler_two'):
            dispatch(ExampleEvent(msg))

    # method_called should have exactly one call, derived from the event
    # handler and not from the disabled @once entrypoint
    method_called.assert_called_once_with(msg)

def test_event_interface(container_factory, rabbit_config):

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

    dispatch = event_dispatcher(rabbit_config)

    # prints "service b received payload" before "exited"
    with entrypoint_waiter(container, 'handle_event'):
        dispatch("service_a", "event_type", "payload")
    print("exited")

def test_update_is_called_when_battle_finishes(rabbit_config):
    container = ServiceContainer(ScoreService, rabbit_config)
    container.start()

    dispatch = event_dispatcher(rabbit_config)

    with mock.patch.object(ScoreService, 'update_players_score') as mock_method:
        with entrypoint_waiter(container, 'update_players_score'):
            dispatch('battle_service', 'battle_finished', [0, 1, 2])
        mock_method.assert_called_once_with([0, 1, 2])

    container.stop()

def test_dispatch(container_factory, rabbit_config):
    config = rabbit_config

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

    msg = "msg"

    dispatch = event_dispatcher(config)
    with entrypoint_waiter(container, 'handler', timeout=1):
        dispatch('srcservice', 'testevent', msg)
    handler_called.assert_called_once_with(msg)

def test_crawler_triggers_webhook(runner_factory, web_container_config):
    """Is crawler_container dispatching to webhook_container?"""
    runner = runner_factory(web_container_config, CrawlerService,
                            WebhookService)
    webhook_container = get_container(runner, WebhookService)
    storage_w = replace_dependencies(webhook_container, 'storage')
    dispatch = event_dispatcher(web_container_config)
    runner.start()
    with entrypoint_waiter(webhook_container, 'send_response'):
        dispatch('http_server', 'url_to_check',
                 ['http://example.org/test_crawling_group',
                  'datagouvfr', None])
    assert storage_w.get_webhooks_for_url.call_count == 1

def test_dispatch(container_factory, rabbit_config):
    config = rabbit_config

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

    msg = "msg"

    with event_dispatcher('srcservice', config) as dispatch:
        dispatch(TestEvent(msg))

        with wait_for_call(1, handler_called):
            handler_called.assert_called_once_with(msg)

def test_crawling_url(container_factory, rabbit_config, web_container_config):
    crawler_container = container_factory(CrawlerService, web_container_config)
    storage, dispatch_dep = replace_dependencies(crawler_container, 'storage',
                                                 'dispatch')
    crawler_container.start()
    dispatch = event_dispatcher(rabbit_config)
    with entrypoint_waiter(crawler_container, 'check_url'):
        dispatch('http_server', 'url_to_check',
                 ['http://example.org/test_crawling_url', None, None])
    assert storage.store_url.call_count == 1
    assert storage.store_group.call_count == 0
    assert storage.store_metadata.call_count == 1
    # fired 'url_crawled'
    assert dispatch_dep.call_count == 1

def test_multiple_runners_coexist(
    runner_factory, rabbit_config, rabbit_manager
):

    runner1 = runner_factory(rabbit_config, Service)
    runner1.start()

    runner2 = runner_factory(rabbit_config, Service)
    runner2.start()

    vhost = rabbit_config['vhost']

    # verify there are two event queues with a single consumer each
    def check_consumers():
        evt_queues = [queue for queue in rabbit_manager.get_queues(vhost)
                      if queue['name'].startswith('evt-srcservice-testevent')]
        assert len(evt_queues) == 2
        for queue in evt_queues:
            assert queue['consumers'] == 1

    # rabbit's management API seems to lag
    assert_stops_raising(check_consumers)

    # test events (both services will receive if in "broadcast" mode)
    event_data = "msg"
    dispatch = event_dispatcher(rabbit_config)
    dispatch('srcservice', "testevent", event_data)

    with eventlet.Timeout(1):
        while len(received) < 2:
            eventlet.sleep()

        assert received == [event_data, event_data]

    # verify there are two consumers on the rpc queue
    rpc_queue = rabbit_manager.get_queue(vhost, 'rpc-service')
    assert rpc_queue['consumers'] == 2

    # test rpc (only one service will respond)
    del received[:]
    arg = "msg"
    with ServiceRpcProxy('service', rabbit_config) as proxy:
        proxy.handle(arg)

    with eventlet.Timeout(1):
        while len(received) == 0:
            eventlet.sleep()

        assert received == [arg]

def test_entrypoint_waiter_result_teardown_race(
    container_factory, rabbit_config, counter
):
    tracker = Mock()

    class TrackingDependency(DependencyProvider):

        def worker_result(self, worker_ctx, res, exc_info):
            tracker.worker_result()

        def worker_teardown(self, worker_ctx):
            tracker.worker_teardown()

    class Service(object):
        name = "service"

        tracker = TrackingDependency()

        @event_handler('srcservice', 'eventtype')
        def handle(self, msg):
            tracker.handle(msg)

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

    def wait_for_two_calls(worker_ctx, res, exc_info):
        return counter.count() > 1

    dispatch = event_dispatcher(rabbit_config)
    with entrypoint_waiter(container, 'handle', callback=wait_for_two_calls):

        # dispatch the first message
        # wait until teardown has happened
        with wait_for_call(TrackingDependency, 'worker_teardown'):
            dispatch('srcservice', 'eventtype', "msg")

        assert tracker.worker_teardown.call_count == 1
        assert tracker.worker_result.call_count == 1
        assert tracker.handle.call_count == 1

        # dispatch the second event
        dispatch('srcservice', 'eventtype', "msg")

    # we should wait for the second teardown to complete before exiting
    # the entrypoint waiter
    assert tracker.worker_teardown.call_count == 2
    assert tracker.worker_result.call_count == 2
    assert tracker.handle.call_count == 2

def test_multiple_runners_coexist(
    runner_factory, rabbit_config, rabbit_manager, service_cls, tracker
):

    runner1 = runner_factory(rabbit_config, service_cls)
    runner1.start()

    runner2 = runner_factory(rabbit_config, service_cls)
    runner2.start()

    vhost = rabbit_config['vhost']

    # verify there are two event queues with a single consumer each
    def check_consumers():
        evt_queues = [queue for queue in rabbit_manager.get_queues(vhost)
                      if queue['name'].startswith('evt-srcservice-testevent')]
        assert len(evt_queues) == 2
        for queue in evt_queues:
            assert queue['consumers'] == 1

    # rabbit's management API seems to lag
    assert_stops_raising(check_consumers)

    # test events (both services will receive if in "broadcast" mode)
    event_data = "event"
    dispatch = event_dispatcher(rabbit_config)

    container1 = list(runner1.containers)[0]
    container2 = list(runner2.containers)[0]

    with entrypoint_waiter(container1, "handle"):
        with entrypoint_waiter(container2, "handle"):
            dispatch('srcservice', "testevent", event_data)
    assert tracker.call_args_list == [call(event_data), call(event_data)]

    # verify there are two consumers on the rpc queue
    rpc_queue = rabbit_manager.get_queue(vhost, 'rpc-service')
    assert rpc_queue['consumers'] == 2

    # test rpc (only one service will respond)
    arg = "arg"
    with ServiceRpcProxy('service', rabbit_config) as proxy:
        proxy.handle(arg)

    assert tracker.call_args_list == [
        call(event_data), call(event_data), call(arg)
    ]

def test_entrypoint_waiter_duplicate(container_factory, rabbit_config):

    class Service(object):
        name = "service"

        @event_handler('srcservice', 'eventtype')
        def handle_event(self, msg):
            handle_event(msg)

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

    dispatch = event_dispatcher(rabbit_config)
    with entrypoint_waiter(container, 'handle_event'):
        with entrypoint_waiter(container, 'handle_event'):
            dispatch('srcservice', 'eventtype', "msg")

    assert handle_event.call_args_list == [call("msg")]

def test_entrypoint_waiter_result(container_factory, rabbit_config):

    class Service(object):
        name = "service"

        @event_handler('srcservice', 'eventtype')
        def handle_event(self, msg):
            return msg.upper()

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

    dispatch = event_dispatcher(rabbit_config)
    with entrypoint_waiter(container, 'handle_event') as result:
        dispatch('srcservice', 'eventtype', "msg")

    res = result.get()
    assert res == "MSG"

def test_crawling_head_offender_url(
        container_factory, web_container_config, rmock=None):
    url_to_check = 'http://example-head.com/test_crawling_url'
    rmock.head(url_to_check)
    rmock.get(url_to_check, text='xxx')
    web_container_config['HEAD_DOMAINS_BLACKLIST'] = ['example-head.com']
    crawler_container = container_factory(CrawlerService, web_container_config)
    storage = replace_dependencies(crawler_container, 'storage')
    crawler_container.start()
    dispatch = event_dispatcher(web_container_config)
    with entrypoint_waiter(crawler_container, 'check_url'):
        dispatch('http_server', 'url_to_check',
                 [url_to_check, None, None])
    assert storage.store_url.call_count == 1
    assert storage.store_group.call_count == 0
    assert storage.store_metadata.call_count == 1
    # check that no HEAD method was called
    requests_l = filter_mock_requests(url_to_check, rmock.request_history)
    assert len(requests_l) == 1
    assert requests_l[0].method == 'GET'