

def test_produce_consume_noack(self):
    channel = self.c.channel()
    producer = Producer(channel, self.e)
    consumer = Consumer(channel, self.q, no_ack=True)
    for i in range(10):
        producer.publish({
            'foo': i,
        }, routing_key='test_transport_memory')
    _received = []

    def callback(message_data, message):
        _received.append(message)
    consumer.register_callback(callback)
    consumer.consume()
    while 1:
        if (len(_received) == 10):
            break
        self.c.drain_events()
    self.assertEqual(len(_received), 10)

dataset/ETHPy150Open celery/kombu/kombu/tests/transport/test_memory.py test_MemoryTransport.test_produce_consume_noack/original