def test_prefetch_count(rabbit_manager, rabbit_config, mock_container):
    container = mock_container
    container.shared_extensions = {}
    container.config = rabbit_config
    container.max_workers = 1
    container.spawn_managed_thread = spawn_thread
    content_type = 'application/data'
    container.accept = [content_type]

    class NonShared(QueueConsumer):
        @property
        def sharing_key(self):
            return uuid.uuid4()

    queue_consumer1 = NonShared().bind(container)
    queue_consumer1.setup()
    queue_consumer2 = NonShared().bind(container)
    queue_consumer2.setup()

    consumer_continue = Event()

    class Handler1(object):
        queue = ham_queue

        def handle_message(self, body, message):
            consumer_continue.wait()
            queue_consumer1.ack_message(message)

    messages = []

    class Handler2(object):
        queue = ham_queue

        def handle_message(self, body, message):
            messages.append(body)
            queue_consumer2.ack_message(message)

    handler1 = Handler1()
    handler2 = Handler2()

    queue_consumer1.register_provider(handler1)
    queue_consumer2.register_provider(handler2)

    queue_consumer1.start()
    queue_consumer2.start()

    vhost = rabbit_config['vhost']
    # the first consumer only has a prefetch_count of 1 and will only
    # consume 1 message and wait in handler1()
    rabbit_manager.publish(vhost, 'spam', '', 'ham',
                           properties=dict(content_type=content_type))
    # the next message will go to handler2() no matter of any prefetch_count
    rabbit_manager.publish(vhost, 'spam', '', 'eggs',
                           properties=dict(content_type=content_type))
    # the third message is only going to handler2 because the first consumer
    # has a prefetch_count of 1 and thus is unable to deal with another message
    # until having ACKed the first one
    rabbit_manager.publish(vhost, 'spam', '', 'bacon',
                           properties=dict(content_type=content_type))

    with eventlet.Timeout(TIMEOUT):
        while 2 < len(messages):
            eventlet.sleep()

    # allow the waiting consumer to ack its message
    consumer_continue.send(None)

    assert messages == ['eggs', 'bacon']

    queue_consumer1.unregister_provider(handler1)
    queue_consumer2.unregister_provider(handler2)

    queue_consumer1.kill()
    queue_consumer2.kill()
dataset/ETHPy150Open onefinestay/nameko/test/test_queue_consumer.py/test_prefetch_count/swapped operands of `<`