Mercurial > public > queues
view test.py @ 21:331d96e6a133
Fixed NameError in the delete_queue method in the dummy backend. Thanks for the report, cleifer.
author | mcroydon |
---|---|
date | Fri, 30 Apr 2010 15:58:07 +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()