+++ /dev/null
-# Some simple queue module tests, plus some failure conditions\r
-# to ensure the Queue locks remain stable.\r
-import Queue\r
-import time\r
-import unittest\r
-from test import test_support\r
-threading = test_support.import_module('threading')\r
-\r
-QUEUE_SIZE = 5\r
-\r
-# A thread to run a function that unclogs a blocked Queue.\r
-class _TriggerThread(threading.Thread):\r
- def __init__(self, fn, args):\r
- self.fn = fn\r
- self.args = args\r
- self.startedEvent = threading.Event()\r
- threading.Thread.__init__(self)\r
-\r
- def run(self):\r
- # The sleep isn't necessary, but is intended to give the blocking\r
- # function in the main thread a chance at actually blocking before\r
- # we unclog it. But if the sleep is longer than the timeout-based\r
- # tests wait in their blocking functions, those tests will fail.\r
- # So we give them much longer timeout values compared to the\r
- # sleep here (I aimed at 10 seconds for blocking functions --\r
- # they should never actually wait that long - they should make\r
- # progress as soon as we call self.fn()).\r
- time.sleep(0.1)\r
- self.startedEvent.set()\r
- self.fn(*self.args)\r
-\r
-\r
-# Execute a function that blocks, and in a separate thread, a function that\r
-# triggers the release. Returns the result of the blocking function. Caution:\r
-# block_func must guarantee to block until trigger_func is called, and\r
-# trigger_func must guarantee to change queue state so that block_func can make\r
-# enough progress to return. In particular, a block_func that just raises an\r
-# exception regardless of whether trigger_func is called will lead to\r
-# timing-dependent sporadic failures, and one of those went rarely seen but\r
-# undiagnosed for years. Now block_func must be unexceptional. If block_func\r
-# is supposed to raise an exception, call do_exceptional_blocking_test()\r
-# instead.\r
-\r
-class BlockingTestMixin:\r
-\r
- def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args):\r
- self.t = _TriggerThread(trigger_func, trigger_args)\r
- self.t.start()\r
- self.result = block_func(*block_args)\r
- # If block_func returned before our thread made the call, we failed!\r
- if not self.t.startedEvent.is_set():\r
- self.fail("blocking function '%r' appeared not to block" %\r
- block_func)\r
- self.t.join(10) # make sure the thread terminates\r
- if self.t.is_alive():\r
- self.fail("trigger function '%r' appeared to not return" %\r
- trigger_func)\r
- return self.result\r
-\r
- # Call this instead if block_func is supposed to raise an exception.\r
- def do_exceptional_blocking_test(self,block_func, block_args, trigger_func,\r
- trigger_args, expected_exception_class):\r
- self.t = _TriggerThread(trigger_func, trigger_args)\r
- self.t.start()\r
- try:\r
- try:\r
- block_func(*block_args)\r
- except expected_exception_class:\r
- raise\r
- else:\r
- self.fail("expected exception of kind %r" %\r
- expected_exception_class)\r
- finally:\r
- self.t.join(10) # make sure the thread terminates\r
- if self.t.is_alive():\r
- self.fail("trigger function '%r' appeared to not return" %\r
- trigger_func)\r
- if not self.t.startedEvent.is_set():\r
- self.fail("trigger thread ended but event never set")\r
-\r
-\r
-class BaseQueueTest(unittest.TestCase, BlockingTestMixin):\r
- def setUp(self):\r
- self.cum = 0\r
- self.cumlock = threading.Lock()\r
-\r
- def simple_queue_test(self, q):\r
- if not q.empty():\r
- raise RuntimeError, "Call this function with an empty queue"\r
- # I guess we better check things actually queue correctly a little :)\r
- q.put(111)\r
- q.put(333)\r
- q.put(222)\r
- target_order = dict(Queue = [111, 333, 222],\r
- LifoQueue = [222, 333, 111],\r
- PriorityQueue = [111, 222, 333])\r
- actual_order = [q.get(), q.get(), q.get()]\r
- self.assertEqual(actual_order, target_order[q.__class__.__name__],\r
- "Didn't seem to queue the correct data!")\r
- for i in range(QUEUE_SIZE-1):\r
- q.put(i)\r
- self.assertTrue(not q.empty(), "Queue should not be empty")\r
- self.assertTrue(not q.full(), "Queue should not be full")\r
- last = 2 * QUEUE_SIZE\r
- full = 3 * 2 * QUEUE_SIZE\r
- q.put(last)\r
- self.assertTrue(q.full(), "Queue should be full")\r
- try:\r
- q.put(full, block=0)\r
- self.fail("Didn't appear to block with a full queue")\r
- except Queue.Full:\r
- pass\r
- try:\r
- q.put(full, timeout=0.01)\r
- self.fail("Didn't appear to time-out with a full queue")\r
- except Queue.Full:\r
- pass\r
- # Test a blocking put\r
- self.do_blocking_test(q.put, (full,), q.get, ())\r
- self.do_blocking_test(q.put, (full, True, 10), q.get, ())\r
- # Empty it\r
- for i in range(QUEUE_SIZE):\r
- q.get()\r
- self.assertTrue(q.empty(), "Queue should be empty")\r
- try:\r
- q.get(block=0)\r
- self.fail("Didn't appear to block with an empty queue")\r
- except Queue.Empty:\r
- pass\r
- try:\r
- q.get(timeout=0.01)\r
- self.fail("Didn't appear to time-out with an empty queue")\r
- except Queue.Empty:\r
- pass\r
- # Test a blocking get\r
- self.do_blocking_test(q.get, (), q.put, ('empty',))\r
- self.do_blocking_test(q.get, (True, 10), q.put, ('empty',))\r
-\r
-\r
- def worker(self, q):\r
- while True:\r
- x = q.get()\r
- if x is None:\r
- q.task_done()\r
- return\r
- with self.cumlock:\r
- self.cum += x\r
- q.task_done()\r
-\r
- def queue_join_test(self, q):\r
- self.cum = 0\r
- for i in (0,1):\r
- threading.Thread(target=self.worker, args=(q,)).start()\r
- for i in xrange(100):\r
- q.put(i)\r
- q.join()\r
- self.assertEqual(self.cum, sum(range(100)),\r
- "q.join() did not block until all tasks were done")\r
- for i in (0,1):\r
- q.put(None) # instruct the threads to close\r
- q.join() # verify that you can join twice\r
-\r
- def test_queue_task_done(self):\r
- # Test to make sure a queue task completed successfully.\r
- q = self.type2test()\r
- try:\r
- q.task_done()\r
- except ValueError:\r
- pass\r
- else:\r
- self.fail("Did not detect task count going negative")\r
-\r
- def test_queue_join(self):\r
- # Test that a queue join()s successfully, and before anything else\r
- # (done twice for insurance).\r
- q = self.type2test()\r
- self.queue_join_test(q)\r
- self.queue_join_test(q)\r
- try:\r
- q.task_done()\r
- except ValueError:\r
- pass\r
- else:\r
- self.fail("Did not detect task count going negative")\r
-\r
- def test_simple_queue(self):\r
- # Do it a couple of times on the same queue.\r
- # Done twice to make sure works with same instance reused.\r
- q = self.type2test(QUEUE_SIZE)\r
- self.simple_queue_test(q)\r
- self.simple_queue_test(q)\r
-\r
-\r
-class QueueTest(BaseQueueTest):\r
- type2test = Queue.Queue\r
-\r
-class LifoQueueTest(BaseQueueTest):\r
- type2test = Queue.LifoQueue\r
-\r
-class PriorityQueueTest(BaseQueueTest):\r
- type2test = Queue.PriorityQueue\r
-\r
-\r
-\r
-# A Queue subclass that can provoke failure at a moment's notice :)\r
-class FailingQueueException(Exception):\r
- pass\r
-\r
-class FailingQueue(Queue.Queue):\r
- def __init__(self, *args):\r
- self.fail_next_put = False\r
- self.fail_next_get = False\r
- Queue.Queue.__init__(self, *args)\r
- def _put(self, item):\r
- if self.fail_next_put:\r
- self.fail_next_put = False\r
- raise FailingQueueException, "You Lose"\r
- return Queue.Queue._put(self, item)\r
- def _get(self):\r
- if self.fail_next_get:\r
- self.fail_next_get = False\r
- raise FailingQueueException, "You Lose"\r
- return Queue.Queue._get(self)\r
-\r
-class FailingQueueTest(unittest.TestCase, BlockingTestMixin):\r
-\r
- def failing_queue_test(self, q):\r
- if not q.empty():\r
- raise RuntimeError, "Call this function with an empty queue"\r
- for i in range(QUEUE_SIZE-1):\r
- q.put(i)\r
- # Test a failing non-blocking put.\r
- q.fail_next_put = True\r
- try:\r
- q.put("oops", block=0)\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- q.fail_next_put = True\r
- try:\r
- q.put("oops", timeout=0.1)\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- q.put("last")\r
- self.assertTrue(q.full(), "Queue should be full")\r
- # Test a failing blocking put\r
- q.fail_next_put = True\r
- try:\r
- self.do_blocking_test(q.put, ("full",), q.get, ())\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- # Check the Queue isn't damaged.\r
- # put failed, but get succeeded - re-add\r
- q.put("last")\r
- # Test a failing timeout put\r
- q.fail_next_put = True\r
- try:\r
- self.do_exceptional_blocking_test(q.put, ("full", True, 10), q.get, (),\r
- FailingQueueException)\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- # Check the Queue isn't damaged.\r
- # put failed, but get succeeded - re-add\r
- q.put("last")\r
- self.assertTrue(q.full(), "Queue should be full")\r
- q.get()\r
- self.assertTrue(not q.full(), "Queue should not be full")\r
- q.put("last")\r
- self.assertTrue(q.full(), "Queue should be full")\r
- # Test a blocking put\r
- self.do_blocking_test(q.put, ("full",), q.get, ())\r
- # Empty it\r
- for i in range(QUEUE_SIZE):\r
- q.get()\r
- self.assertTrue(q.empty(), "Queue should be empty")\r
- q.put("first")\r
- q.fail_next_get = True\r
- try:\r
- q.get()\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- self.assertTrue(not q.empty(), "Queue should not be empty")\r
- q.fail_next_get = True\r
- try:\r
- q.get(timeout=0.1)\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- self.assertTrue(not q.empty(), "Queue should not be empty")\r
- q.get()\r
- self.assertTrue(q.empty(), "Queue should be empty")\r
- q.fail_next_get = True\r
- try:\r
- self.do_exceptional_blocking_test(q.get, (), q.put, ('empty',),\r
- FailingQueueException)\r
- self.fail("The queue didn't fail when it should have")\r
- except FailingQueueException:\r
- pass\r
- # put succeeded, but get failed.\r
- self.assertTrue(not q.empty(), "Queue should not be empty")\r
- q.get()\r
- self.assertTrue(q.empty(), "Queue should be empty")\r
-\r
- def test_failing_queue(self):\r
- # Test to make sure a queue is functioning correctly.\r
- # Done twice to the same instance.\r
- q = FailingQueue(QUEUE_SIZE)\r
- self.failing_queue_test(q)\r
- self.failing_queue_test(q)\r
-\r
-\r
-def test_main():\r
- test_support.run_unittest(QueueTest, LifoQueueTest, PriorityQueueTest,\r
- FailingQueueTest)\r
-\r
-\r
-if __name__ == "__main__":\r
- test_main()\r
projects / mirror_edk2.git / blobdiff