mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-04 00:30:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Semantic-neutral format and comment changes.

Browse source
This commit is contained in:
Tim Peters 2004-08-20 02:37:25 +00:00
parent 8623b36ee0
commit afe5297b8a
1 changed files with 16 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

30
Lib/test/test_queue.py
View file

@ -1,5 +1,5 @@
# Some simple Queue module tests, plus some failure conditions # Some simple Queue module tests, plus some failure conditions
# to ensure the Queue locks remain stable # to ensure the Queue locks remain stable.
import Queue import Queue
import sys import sys
import threading import threading
@ -7,21 +7,23 @@ import time
from test.test_support import verify, TestFailed, verbose from test.test_support import verify, TestFailed, verbose
queue_size = 5 QUEUE_SIZE = 5
# Execute a function that blocks, and in a seperate thread, a function that # A thread to run a function that unclogs a blocked Queue.
# triggers the release. Returns the result of the blocking function.
class _TriggerThread(threading.Thread): class _TriggerThread(threading.Thread):
def __init__(self, fn, args): def __init__(self, fn, args):
self.fn = fn self.fn = fn
self.args = args self.args = args
self.startedEvent = threading.Event() self.startedEvent = threading.Event()
threading.Thread.__init__(self) threading.Thread.__init__(self)
def run(self): def run(self):
time.sleep(.1) time.sleep(.1)
self.startedEvent.set() self.startedEvent.set()
self.fn(*self.args) self.fn(*self.args)
# Execute a function that blocks, and in a seperate thread, a function that
# triggers the release. Returns the result of the blocking function.
def _doBlockingTest(block_func, block_args, trigger_func, trigger_args): def _doBlockingTest(block_func, block_args, trigger_func, trigger_args):
t = _TriggerThread(trigger_func, trigger_args) t = _TriggerThread(trigger_func, trigger_args)
t.start() t.start()
@ -60,7 +62,7 @@ class FailingQueue(Queue.Queue):
def FailingQueueTest(q): def FailingQueueTest(q):
if not q.empty(): if not q.empty():
raise RuntimeError, "Call this function with an empty queue" raise RuntimeError, "Call this function with an empty queue"
for i in range(queue_size-1): for i in range(QUEUE_SIZE-1):
q.put(i) q.put(i)
# Test a failing non-blocking put. # Test a failing non-blocking put.
q.fail_next_put = True q.fail_next_put = True
@ -80,7 +82,7 @@ def FailingQueueTest(q):
# Test a failing blocking put # Test a failing blocking put
q.fail_next_put = True q.fail_next_put = True
try: try:
_doBlockingTest( q.put, ("full",), q.get, ()) _doBlockingTest(q.put, ("full",), q.get, ())
raise TestFailed("The queue didn't fail when it should have") raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException: except FailingQueueException:
pass pass
@ -90,7 +92,7 @@ def FailingQueueTest(q):
# Test a failing timeout put # Test a failing timeout put
q.fail_next_put = True q.fail_next_put = True
try: try:
_doBlockingTest( q.put, ("full", True, 0.2), q.get, ()) _doBlockingTest(q.put, ("full", True, 0.2), q.get, ())
raise TestFailed("The queue didn't fail when it should have") raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException: except FailingQueueException:
pass pass
@ -105,7 +107,7 @@ def FailingQueueTest(q):
# Test a blocking put # Test a blocking put
_doBlockingTest( q.put, ("full",), q.get, ()) _doBlockingTest( q.put, ("full",), q.get, ())
# Empty it # Empty it
for i in range(queue_size): for i in range(QUEUE_SIZE):
q.get() q.get()
verify(q.empty(), "Queue should be empty") verify(q.empty(), "Queue should be empty")
q.put("first") q.put("first")
@ -144,7 +146,7 @@ def SimpleQueueTest(q):
q.put(222) q.put(222)
verify(q.get() == 111 and q.get() == 222, verify(q.get() == 111 and q.get() == 222,
"Didn't seem to queue the correct data!") "Didn't seem to queue the correct data!")
for i in range(queue_size-1): for i in range(QUEUE_SIZE-1):
q.put(i) q.put(i)
verify(not q.full(), "Queue should not be full") verify(not q.full(), "Queue should not be full")
q.put("last") q.put("last")
@ -160,10 +162,10 @@ def SimpleQueueTest(q):
except Queue.Full: except Queue.Full:
pass pass
# Test a blocking put # Test a blocking put
_doBlockingTest( q.put, ("full",), q.get, ()) _doBlockingTest(q.put, ("full",), q.get, ())
_doBlockingTest( q.put, ("full", True, 0.2), q.get, ()) _doBlockingTest(q.put, ("full", True, 0.2), q.get, ())
# Empty it # Empty it
for i in range(queue_size): for i in range(QUEUE_SIZE):
q.get() q.get()
verify(q.empty(), "Queue should be empty") verify(q.empty(), "Queue should be empty")
try: try:
@ -181,13 +183,13 @@ def SimpleQueueTest(q):
_doBlockingTest(q.get, (True, 0.2), q.put, ('empty',)) _doBlockingTest(q.get, (True, 0.2), q.put, ('empty',))
def test(): def test():
q=Queue.Queue(queue_size) q = Queue.Queue(QUEUE_SIZE)
# Do it a couple of times on the same queue # Do it a couple of times on the same queue
SimpleQueueTest(q) SimpleQueueTest(q)
SimpleQueueTest(q) SimpleQueueTest(q)
if verbose: if verbose:
print "Simple Queue tests seemed to work" print "Simple Queue tests seemed to work"
q = FailingQueue(queue_size) q = FailingQueue(QUEUE_SIZE)
FailingQueueTest(q) FailingQueueTest(q)
FailingQueueTest(q) FailingQueueTest(q)
if verbose: if verbose: