mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 08:34:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Patch #572628: Optional timeouts for put and get.

Browse source
This commit is contained in:
Martin v. Löwis 2002-10-15 15:11:13 +00:00
parent d98d25e22d
commit 77ac429eff
4 changed files with 137 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

39
Lib/test/test_queue.py
View file

@ -60,17 +60,23 @@ def FailingQueueTest(q):
raise RuntimeError, "Call this function with an empty queue"
for i in range(queue_size-1):
q.put(i)
q.fail_next_put = True
# Test a failing non-blocking put.
q.fail_next_put = True
try:
q.put("oops", block=0)
raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException:
pass
q.fail_next_put = True
try:
q.put("oops", timeout=0.1)
raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException:
pass
q.put("last")
verify(q.full(), "Queue should be full")
q.fail_next_put = True
# Test a failing blocking put
q.fail_next_put = True
try:
_doBlockingTest( q.put, ("full",), q.get, ())
raise TestFailed("The queue didn't fail when it should have")
@ -79,6 +85,16 @@ def FailingQueueTest(q):
# Check the Queue isn't damaged.
# put failed, but get succeeded - re-add
q.put("last")
# Test a failing timeout put
q.fail_next_put = True
try:
_doBlockingTest( q.put, ("full", True, 0.2), q.get, ())
raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException:
pass
# Check the Queue isn't damaged.
# put failed, but get succeeded - re-add
q.put("last")
verify(q.full(), "Queue should be full")
q.get()
verify(not q.full(), "Queue should not be full")
@ -98,6 +114,13 @@ def FailingQueueTest(q):
except FailingQueueException:
pass
verify(not q.empty(), "Queue should not be empty")
q.fail_next_get = True
try:
q.get(timeout=0.1)
raise TestFailed("The queue didn't fail when it should have")
except FailingQueueException:
pass
verify(not q.empty(), "Queue should not be empty")
q.get()
verify(q.empty(), "Queue should be empty")
q.fail_next_get = True
@ -128,8 +151,14 @@ def SimpleQueueTest(q):
raise TestFailed("Didn't appear to block with a full queue")
except Queue.Full:
pass
try:
q.put("full", timeout=0.1)
raise TestFailed("Didn't appear to time-out with a full queue")
except Queue.Full:
pass
# Test a blocking put
_doBlockingTest( q.put, ("full",), q.get, ())
_doBlockingTest( q.put, ("full", True, 0.2), q.get, ())
# Empty it
for i in range(queue_size):
q.get()
@ -139,8 +168,14 @@ def SimpleQueueTest(q):
raise TestFailed("Didn't appear to block with an empty queue")
except Queue.Empty:
pass
try:
q.get(timeout=0.1)
raise TestFailed("Didn't appear to time-out with an empty queue")
except Queue.Empty:
pass
# Test a blocking get
_doBlockingTest( q.get, (), q.put, ('empty',))
_doBlockingTest( q.get, (True, 0.2), q.put, ('empty',))
def test():
q=Queue.Queue(queue_size)