Kombu: How to use in memory transport for unit tests ?

[c4pone]

Hey guys,

I just want to write some functional tests were I produce a message to the queue and see if the message gets correctly handled. Therefore I trying to use the memory transport, but can't find much documentation about it.

So my question is, would it be possible that you guys add an test example to the docs, or just post one directly here ?

[tcwalther]

@c4pone Why did you close it? I was just having the same question. I guess the solution is to simply create an in-memory transport, then to check whether a function produces a message, have a consumer in your test function wait until it consumes a message? Even if this is how one should go about it, a couple of examples in the documentation would be very nice.

[c4pone]

Hey @tcwalther, I found shortly after I asked this question the following unit test https://github.com/celery/kombu/blob/master/kombu/tests/transport/test_memory.py

also
http://stackoverflow.com/questions/38149888/how-to-properly-use-the-in-memory-transport-for-unit-tests

But I agree with you, a simple example in the documentation would be really nice.

[ask]

Feel free to add this example in the documentation, you can even edit the files on Github directly and submit a pull request that way :)

[kindule]

What about an example like this?

"""
Example that use memory transport for in process commuication.
"""
import threading
from time import sleep, time
from kombu import Connection, Exchange, Queue, Consumer
from kombu.utils.debug import setup_logging

media_exchange = Exchange('media', 'direct')
video_queue = Queue('video', exchange=media_exchange, routing_key='video')
task_queues = [video_queue]


class Worker(threading.Thread):

    def run(self):
        connection = Connection("memory:///", transport_options={"polling_interval": 0})
        while True:
            with Consumer(connection, task_queues, callbacks=[self.on_message]):
                connection.drain_events()

    def on_message(self, body, message):
        print("%s RECEIVED MESSAGE: %r" % (time(), body))
        message.ack()


if __name__ == '__main__':
    setup_logging(loglevel='INFO', loggers=[''])

    try:
        worker = Worker()
        worker.start()
    except KeyboardInterrupt:
        print('bye bye')

    while True:
        with Connection('memory:///') as conn:
            producer = conn.Producer(serializer='json')
            producer.publish({"foo": "bar"}, exchange=media_exchange, routing_key='video', declare=task_queues)
        sleep(0.1)