Mercurial > public > queues
view test.py @ 1:32222d11961f
Initial commit with two tested backends: memcached protocol and Amazon SQS. Because of the non-guaranteed nature of SQS queues, some tests may fail even though the library is working properly.
author | mcroydon |
---|---|
date | Thu, 08 Jan 2009 07:49:35 +0000 |
parents | |
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()