Mercurial > public > queues
view test.py @ 28:baedd22fd73a
Implement timeout for Redis backend.
author | btimby |
---|---|
date | Sun, 05 Feb 2012 03:09:04 +0000 |
parents | a0d3e275c885 |
children |
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 >>> unicode(q.read()) u'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()