Changeset 391 for python/trunk/Lib/test/test_queue.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_queue.py (modified) (13 diffs)

python/trunk/Lib/test/test_queue.py

@@ -2 +2 @@
 # to ensure the Queue locks remain stable.
 import Queue
-import sys
-import threading
 import time
 import unittest
 from test import test_support
+threading = test_support.import_module('threading')
 
 QUEUE_SIZE = 5
@@ -44 +43 @@
 
 class BlockingTestMixin:
+
+    def tearDown(self):
+        self.t = None
 
     def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args):
@@ -81 +83 @@
 
 
-class BaseQueueTest(unittest.TestCase, BlockingTestMixin):
+class BaseQueueTest(BlockingTestMixin):
     def setUp(self):
         self.cum = 0
@@ -97 +99 @@
                             PriorityQueue = [111, 222, 333])
         actual_order = [q.get(), q.get(), q.get()]
-        self.assertEquals(actual_order, target_order[q.__class__.__name__],
-                          "Didn't seem to queue the correct data!")
+        self.assertEqual(actual_order, target_order[q.__class__.__name__],
+                         "Didn't seem to queue the correct data!")
         for i in range(QUEUE_SIZE-1):
             q.put(i)
-            self.assert_(not q.empty(), "Queue should not be empty")
-        self.assert_(not q.full(), "Queue should not be full")
-        q.put("last")
-        self.assert_(q.full(), "Queue should be full")
-        try:
-            q.put("full", block=0)
+            self.assertTrue(not q.empty(), "Queue should not be empty")
+        self.assertTrue(not q.full(), "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")
+        try:
+            q.put(full, block=0)
             self.fail("Didn't appear to block with a full queue")
         except Queue.Full:
             pass
         try:
-            q.put("full", timeout=0.01)
+            q.put(full, timeout=0.01)
             self.fail("Didn't appear to time-out with a full queue")
         except Queue.Full:
             pass
         # Test a blocking put
-        self.do_blocking_test(q.put, ("full",), q.get, ())
-        self.do_blocking_test(q.put, ("full", True, 10), q.get, ())
+        self.do_blocking_test(q.put, (full,), q.get, ())
+        self.do_blocking_test(q.put, (full, True, 10), q.get, ())
         # Empty it
         for i in range(QUEUE_SIZE):
             q.get()
-        self.assert_(q.empty(), "Queue should be empty")
+        self.assertTrue(q.empty(), "Queue should be empty")
         try:
             q.get(block=0)
@@ -154 +158 @@
             q.put(i)
         q.join()
-        self.assertEquals(self.cum, sum(range(100)),
-                          "q.join() did not block until all tasks were done")
+        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
@@ -191 +195 @@
 
 
-class QueueTest(BaseQueueTest):
+class QueueTest(BaseQueueTest, unittest.TestCase):
     type2test = Queue.Queue
 
-class LifoQueueTest(BaseQueueTest):
+class LifoQueueTest(BaseQueueTest, unittest.TestCase):
     type2test = Queue.LifoQueue
 
-class PriorityQueueTest(BaseQueueTest):
+class PriorityQueueTest(BaseQueueTest, unittest.TestCase):
     type2test = Queue.PriorityQueue
 
@@ -222 +226 @@
         return Queue.Queue._get(self)
 
-class FailingQueueTest(unittest.TestCase, BlockingTestMixin):
+class FailingQueueTest(BlockingTestMixin, unittest.TestCase):
 
     def failing_queue_test(self, q):
@@ -243 +247 @@
             pass
         q.put("last")
-        self.assert_(q.full(), "Queue should be full")
+        self.assertTrue(q.full(), "Queue should be full")
         # Test a failing blocking put
         q.fail_next_put = True
@@ -265 +269 @@
         # put failed, but get succeeded - re-add
         q.put("last")
-        self.assert_(q.full(), "Queue should be full")
+        self.assertTrue(q.full(), "Queue should be full")
         q.get()
-        self.assert_(not q.full(), "Queue should not be full")
+        self.assertTrue(not q.full(), "Queue should not be full")
         q.put("last")
-        self.assert_(q.full(), "Queue should be full")
+        self.assertTrue(q.full(), "Queue should be full")
         # Test a blocking put
         self.do_blocking_test(q.put, ("full",), q.get, ())
@@ -275 +279 @@
         for i in range(QUEUE_SIZE):
             q.get()
-        self.assert_(q.empty(), "Queue should be empty")
+        self.assertTrue(q.empty(), "Queue should be empty")
         q.put("first")
         q.fail_next_get = True
@@ -283 +287 @@
         except FailingQueueException:
             pass
-        self.assert_(not q.empty(), "Queue should not be empty")
+        self.assertTrue(not q.empty(), "Queue should not be empty")
         q.fail_next_get = True
         try:
@@ -290 +294 @@
         except FailingQueueException:
             pass
-        self.assert_(not q.empty(), "Queue should not be empty")
+        self.assertTrue(not q.empty(), "Queue should not be empty")
         q.get()
-        self.assert_(q.empty(), "Queue should be empty")
+        self.assertTrue(q.empty(), "Queue should be empty")
         q.fail_next_get = True
         try:
@@ -301 +305 @@
             pass
         # put succeeded, but get failed.
-        self.assert_(not q.empty(), "Queue should not be empty")
+        self.assertTrue(not q.empty(), "Queue should not be empty")
         q.get()
-        self.assert_(q.empty(), "Queue should be empty")
+        self.assertTrue(q.empty(), "Queue should be empty")
 
     def test_failing_queue(self):