Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Style guide reformats. I saw this test fail on a very heavily loaded 

Win98SE box, but whatever the cause, it had scrolled off the DOS box.
(There was just the "test_queue failed" summary at the end of the
regrtest run.)
This commit is contained in:
Tim Peters 2002-11-15 19:08:50 +00:00
parent 49b33fa4cb
commit a1d004af04
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Lib/test/test_queue.py
View File

@ -22,7 +22,7 @@ class _TriggerThread(threading.Thread):
self.startedEvent.set() self.startedEvent.set()
self.fn(*self.args) self.fn(*self.args)
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()
try: try:
@ -30,10 +30,12 @@ def _doBlockingTest( block_func, block_args, trigger_func, trigger_args):
finally: finally:
# If we unblocked before our thread made the call, we failed! # If we unblocked before our thread made the call, we failed!
if not t.startedEvent.isSet(): if not t.startedEvent.isSet():
raise TestFailed("blocking function '%r' appeared not to block" % (block_func,)) raise TestFailed("blocking function '%r' appeared not to block" %
block_func)
t.join(1) # make sure the thread terminates t.join(1) # make sure the thread terminates
if t.isAlive(): if t.isAlive():
raise TestFailed("trigger function '%r' appeared to not return" % (trigger_func,)) raise TestFailed("trigger function '%r' appeared to not return" %
trigger_func)
# A Queue subclass that can provoke failure at a moment's notice :) # A Queue subclass that can provoke failure at a moment's notice :)
class FailingQueueException(Exception): class FailingQueueException(Exception):
@ -140,7 +142,8 @@ def SimpleQueueTest(q):
# I guess we better check things actually queue correctly a little :) # I guess we better check things actually queue correctly a little :)
q.put(111) q.put(111)
q.put(222) q.put(222)
verify(q.get()==111 and q.get()==222, "Didn't seem to queue the correct data!") verify(q.get() == 111 and q.get() == 222,
"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")
@ -174,8 +177,8 @@ def SimpleQueueTest(q):
except Queue.Empty: except Queue.Empty:
pass pass
# Test a blocking get # Test a blocking get
_doBlockingTest( q.get, (), q.put, ('empty',)) _doBlockingTest(q.get, (), q.put, ('empty',))
_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)