Mercurial > public > queues
view test.py @ 20:b4184f99ed13
Added functionality for the dummy backend to make it behave more like a production backend for testability sake. Thanks to Travis Cline for the patch. Fixes issue #4.
author | mcroydon |
---|---|
date | Tue, 20 Oct 2009 22:01:12 +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()