view test.py @ 13:e09cc844ff10

Added zenqueue backend as zenqueued based heavily on a patch by Daniel Lindsley. To use the zenqueued backend you'll need to set QUEUE_BACKEND to zenqueued, and also set QUEUE_ZENQUEUE_CONNECTION. The zenqueued backend defaults to using the client for HTTP but you can use either the HTTP or native method by setting QUEUE_ZENQUEUE_METHOD to http or native. Note that zenqueue does not allow for multiple named queues, so the name argument to the Queue constructor is ignored. Several other options (__len__, delete_queue, and get_list) are not supported by zenqueue.
author mcroydon
date Wed, 20 May 2009 19:25:14 +0000
parents 32222d11961f
children a0d3e275c885
line wrap: on
line source
"""
Test basic queue functionality

>>> from queues import queues
>>> import time
>>> queue_name = 'test_queues_%.f' % time.time()

Verify that the queue does not exist
>>> queue_name in queues.get_list()
False

Create the queue
>>> q = queues.Queue(queue_name)

Write to the queue
>>> q.write('test')
True

Verify that it is indeed in the list
>>> queue_name in queues.get_list()
True

Get the length of the queue

Note that SQS doesn't guarantee that the message
we just wrote will be immediately available
>>> len(q)
1

Read from the queue
>>> q.read()
'test'

The queue should now be empty
Note that SQS doesn't guarantee an accurate count
>>> len(q)
0

>>> try:
...     queues.delete_queue(queue_name)
... except NotImplementedError:
...     print True
True
"""

if __name__ == "__main__":
    import doctest
    doctest.testmod()