diff options
Diffstat (limited to 'Lib/test/test_queue.py')
| -rw-r--r-- | Lib/test/test_queue.py | 122 |
1 files changed, 80 insertions, 42 deletions
diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index f821db689c1..86ad9c07655 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -1,13 +1,16 @@ # Some simple queue module tests, plus some failure conditions # to ensure the Queue locks remain stable. -import Queue +import queue import time import unittest -from test import test_support -threading = test_support.import_module('threading') +from test import support +threading = support.import_module('threading') QUEUE_SIZE = 5 +def qfull(q): + return q.maxsize > 0 and q.qsize() == q.maxsize + # A thread to run a function that unclogs a blocked Queue. class _TriggerThread(threading.Thread): def __init__(self, fn, args): @@ -79,14 +82,16 @@ class BlockingTestMixin: self.fail("trigger thread ended but event never set") -class BaseQueueTest(BlockingTestMixin): +class BaseQueueTestMixin(BlockingTestMixin): def setUp(self): self.cum = 0 self.cumlock = threading.Lock() def simple_queue_test(self, q): - if not q.empty(): - raise RuntimeError, "Call this function with an empty queue" + if q.qsize(): + raise RuntimeError("Call this function with an empty queue") + self.assertTrue(q.empty()) + self.assertFalse(q.full()) # I guess we better check things actually queue correctly a little :) q.put(111) q.put(333) @@ -99,21 +104,23 @@ class BaseQueueTest(BlockingTestMixin): "Didn't seem to queue the correct data!") for i in range(QUEUE_SIZE-1): q.put(i) - self.assertTrue(not q.empty(), "Queue should not be empty") - self.assertTrue(not q.full(), "Queue should not be full") + self.assertTrue(q.qsize(), "Queue should not be empty") + self.assertTrue(not qfull(q), "Queue should not be full") last = 2 * QUEUE_SIZE full = 3 * 2 * QUEUE_SIZE q.put(last) - self.assertTrue(q.full(), "Queue should be full") + self.assertTrue(qfull(q), "Queue should be full") + self.assertFalse(q.empty()) + self.assertTrue(q.full()) try: q.put(full, block=0) self.fail("Didn't appear to block with a full queue") - except Queue.Full: + except queue.Full: pass try: q.put(full, timeout=0.01) self.fail("Didn't appear to time-out with a full queue") - except Queue.Full: + except queue.Full: pass # Test a blocking put self.do_blocking_test(q.put, (full,), q.get, ()) @@ -121,16 +128,16 @@ class BaseQueueTest(BlockingTestMixin): # Empty it for i in range(QUEUE_SIZE): q.get() - self.assertTrue(q.empty(), "Queue should be empty") + self.assertTrue(not q.qsize(), "Queue should be empty") try: q.get(block=0) self.fail("Didn't appear to block with an empty queue") - except Queue.Empty: + except queue.Empty: pass try: q.get(timeout=0.01) self.fail("Didn't appear to time-out with an empty queue") - except Queue.Empty: + except queue.Empty: pass # Test a blocking get self.do_blocking_test(q.get, (), q.put, ('empty',)) @@ -140,7 +147,7 @@ class BaseQueueTest(BlockingTestMixin): def worker(self, q): while True: x = q.get() - if x is None: + if x < 0: q.task_done() return with self.cumlock: @@ -151,13 +158,13 @@ class BaseQueueTest(BlockingTestMixin): self.cum = 0 for i in (0,1): threading.Thread(target=self.worker, args=(q,)).start() - for i in xrange(100): + for i in range(100): q.put(i) q.join() self.assertEqual(self.cum, sum(range(100)), "q.join() did not block until all tasks were done") for i in (0,1): - q.put(None) # instruct the threads to close + q.put(-1) # instruct the threads to close q.join() # verify that you can join twice def test_queue_task_done(self): @@ -190,15 +197,46 @@ class BaseQueueTest(BlockingTestMixin): self.simple_queue_test(q) self.simple_queue_test(q) + def test_negative_timeout_raises_exception(self): + q = self.type2test(QUEUE_SIZE) + with self.assertRaises(ValueError): + q.put(1, timeout=-1) + with self.assertRaises(ValueError): + q.get(1, timeout=-1) + + def test_nowait(self): + q = self.type2test(QUEUE_SIZE) + for i in range(QUEUE_SIZE): + q.put_nowait(1) + with self.assertRaises(queue.Full): + q.put_nowait(1) + + for i in range(QUEUE_SIZE): + q.get_nowait() + with self.assertRaises(queue.Empty): + q.get_nowait() + + def test_shrinking_queue(self): + # issue 10110 + q = self.type2test(3) + q.put(1) + q.put(2) + q.put(3) + with self.assertRaises(queue.Full): + q.put_nowait(4) + self.assertEqual(q.qsize(), 3) + q.maxsize = 2 # shrink the queue + with self.assertRaises(queue.Full): + q.put_nowait(4) -class QueueTest(BaseQueueTest, unittest.TestCase): - type2test = Queue.Queue +class QueueTest(BaseQueueTestMixin, unittest.TestCase): + type2test = queue.Queue -class LifoQueueTest(BaseQueueTest, unittest.TestCase): - type2test = Queue.LifoQueue +class LifoQueueTest(BaseQueueTestMixin, unittest.TestCase): + type2test = queue.LifoQueue -class PriorityQueueTest(BaseQueueTest, unittest.TestCase): - type2test = Queue.PriorityQueue +class PriorityQueueTest(BaseQueueTestMixin, unittest.TestCase): + type2test = queue.PriorityQueue @@ -206,27 +244,27 @@ class PriorityQueueTest(BaseQueueTest, unittest.TestCase): class FailingQueueException(Exception): pass -class FailingQueue(Queue.Queue): +class FailingQueue(queue.Queue): def __init__(self, *args): self.fail_next_put = False self.fail_next_get = False - Queue.Queue.__init__(self, *args) + queue.Queue.__init__(self, *args) def _put(self, item): if self.fail_next_put: self.fail_next_put = False - raise FailingQueueException, "You Lose" - return Queue.Queue._put(self, item) + raise FailingQueueException("You Lose") + return queue.Queue._put(self, item) def _get(self): if self.fail_next_get: self.fail_next_get = False - raise FailingQueueException, "You Lose" - return Queue.Queue._get(self) + raise FailingQueueException("You Lose") + return queue.Queue._get(self) class FailingQueueTest(unittest.TestCase, BlockingTestMixin): def failing_queue_test(self, q): - if not q.empty(): - raise RuntimeError, "Call this function with an empty queue" + if q.qsize(): + raise RuntimeError("Call this function with an empty queue") for i in range(QUEUE_SIZE-1): q.put(i) # Test a failing non-blocking put. @@ -243,7 +281,7 @@ class FailingQueueTest(unittest.TestCase, BlockingTestMixin): except FailingQueueException: pass q.put("last") - self.assertTrue(q.full(), "Queue should be full") + self.assertTrue(qfull(q), "Queue should be full") # Test a failing blocking put q.fail_next_put = True try: @@ -265,17 +303,17 @@ class FailingQueueTest(unittest.TestCase, BlockingTestMixin): # Check the Queue isn't damaged. # put failed, but get succeeded - re-add q.put("last") - self.assertTrue(q.full(), "Queue should be full") + self.assertTrue(qfull(q), "Queue should be full") q.get() - self.assertTrue(not q.full(), "Queue should not be full") + self.assertTrue(not qfull(q), "Queue should not be full") q.put("last") - self.assertTrue(q.full(), "Queue should be full") + self.assertTrue(qfull(q), "Queue should be full") # Test a blocking put self.do_blocking_test(q.put, ("full",), q.get, ()) # Empty it for i in range(QUEUE_SIZE): q.get() - self.assertTrue(q.empty(), "Queue should be empty") + self.assertTrue(not q.qsize(), "Queue should be empty") q.put("first") q.fail_next_get = True try: @@ -283,16 +321,16 @@ class FailingQueueTest(unittest.TestCase, BlockingTestMixin): self.fail("The queue didn't fail when it should have") except FailingQueueException: pass - self.assertTrue(not q.empty(), "Queue should not be empty") + self.assertTrue(q.qsize(), "Queue should not be empty") q.fail_next_get = True try: q.get(timeout=0.1) self.fail("The queue didn't fail when it should have") except FailingQueueException: pass - self.assertTrue(not q.empty(), "Queue should not be empty") + self.assertTrue(q.qsize(), "Queue should not be empty") q.get() - self.assertTrue(q.empty(), "Queue should be empty") + self.assertTrue(not q.qsize(), "Queue should be empty") q.fail_next_get = True try: self.do_exceptional_blocking_test(q.get, (), q.put, ('empty',), @@ -301,9 +339,9 @@ class FailingQueueTest(unittest.TestCase, BlockingTestMixin): except FailingQueueException: pass # put succeeded, but get failed. - self.assertTrue(not q.empty(), "Queue should not be empty") + self.assertTrue(q.qsize(), "Queue should not be empty") q.get() - self.assertTrue(q.empty(), "Queue should be empty") + self.assertTrue(not q.qsize(), "Queue should be empty") def test_failing_queue(self): # Test to make sure a queue is functioning correctly. @@ -314,7 +352,7 @@ class FailingQueueTest(unittest.TestCase, BlockingTestMixin): def test_main(): - test_support.run_unittest(QueueTest, LifoQueueTest, PriorityQueueTest, + support.run_unittest(QueueTest, LifoQueueTest, PriorityQueueTest, FailingQueueTest) |