1 files changed, 40 insertions, 6 deletions
diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py
index c65827d..86ad9c0 100644
--- a/Lib/test/test_queue.py
+++ b/Lib/test/test_queue.py
@@ -1,11 +1,10 @@
 # Some simple queue module tests, plus some failure conditions
 # to ensure the Queue locks remain stable.
 import queue
-import sys
-import threading
 import time
 import unittest
 from test import support
+threading = support.import_module('threading')
 
 QUEUE_SIZE = 5
 
@@ -83,7 +82,7 @@ class BlockingTestMixin:
                 self.fail("trigger thread ended but event never set")
 
 
-class BaseQueueTest(unittest.TestCase, BlockingTestMixin):
+class BaseQueueTestMixin(BlockingTestMixin):
     def setUp(self):
         self.cum = 0
         self.cumlock = threading.Lock()
@@ -91,6 +90,8 @@ class BaseQueueTest(unittest.TestCase, BlockingTestMixin):
     def simple_queue_test(self, q):
         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)
@@ -109,6 +110,8 @@ class BaseQueueTest(unittest.TestCase, BlockingTestMixin):
         full = 3 * 2 * QUEUE_SIZE
         q.put(last)
         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")
@@ -194,14 +197,45 @@ class BaseQueueTest(unittest.TestCase, 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)
 
-class QueueTest(BaseQueueTest):
+    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(BaseQueueTestMixin, unittest.TestCase):
     type2test = queue.Queue
 
-class LifoQueueTest(BaseQueueTest):
+class LifoQueueTest(BaseQueueTestMixin, unittest.TestCase):
     type2test = queue.LifoQueue
 
-class PriorityQueueTest(BaseQueueTest):
+class PriorityQueueTest(BaseQueueTestMixin, unittest.TestCase):
     type2test = queue.PriorityQueue