gh-135698: Fix Cross-interpreter Queue.full() With Negative/Default max_size

Lib/test/test_interpreters/test_queues.py

@@ -208,18 +208,64 @@ def test_empty(self):
         self.assertIs(after, True)
 
     def test_full(self):
-        expected = [False, False, False, True, False, False, False]
-        actual = []
-        queue = queues.create(3)
-        for _ in range(3):
-            actual.append(queue.full())
-            queue.put(None)
-        actual.append(queue.full())
-        for _ in range(3):
-            queue.get()
-            actual.append(queue.full())
+        for maxsize in [1, 3, 11]:
+            with self.subTest(f'maxsize={maxsize}'):
+                num_to_add = maxsize
+                expected = [False] * (num_to_add * 2 + 3)
+                expected[maxsize] = True
+                expected[maxsize + 1] = True
+
+                queue = queues.create(maxsize)
+                actual = []
+                empty = [queue.empty()]
+
+                for _ in range(num_to_add):
+                    actual.append(queue.full())
+                    queue.put_nowait(None)
+                actual.append(queue.full())
+                with self.assertRaises(queues.QueueFull):
+                    queue.put_nowait(None)
+                empty.append(queue.empty())
+
+                for _ in range(num_to_add):
+                    actual.append(queue.full())
+                    queue.get_nowait()
+                actual.append(queue.full())
+                with self.assertRaises(queues.QueueEmpty):
+                    queue.get_nowait()
+                actual.append(queue.full())
+                empty.append(queue.empty())
 
-        self.assertEqual(actual, expected)
+                self.assertEqual(actual, expected)
+                self.assertEqual(empty, [True, False, True])
+
+        # no max size
+        for args in [(), (0,), (-1,), (-10,)]:
+            with self.subTest(f'maxsize={args[0]}' if args else '<default>'):
+                num_to_add = 13
+                expected = [False] * (num_to_add * 2 + 3)
+
+                queue = queues.create(*args)
+                actual = []
+                empty = [queue.empty()]
+
+                for _ in range(num_to_add):
+                    actual.append(queue.full())
+                    queue.put_nowait(None)
+                actual.append(queue.full())
+                empty.append(queue.empty())
+
+                for _ in range(num_to_add):
+                    actual.append(queue.full())
+                    queue.get_nowait()
+                actual.append(queue.full())
+                with self.assertRaises(queues.QueueEmpty):
+                    queue.get_nowait()
+                actual.append(queue.full())
+                empty.append(queue.empty())
+
+                self.assertEqual(actual, expected)
+                self.assertEqual(empty, [True, False, True])
 
     def test_qsize(self):
         expected = [0, 1, 2, 3, 2, 3, 2, 1, 0, 1, 0]

Modules/_interpqueuesmodule.c

@@ -707,8 +707,11 @@ _queue_is_full(_queue *queue, int *p_is_full)
         return err;
     }
 
-    assert(queue->items.count <= queue->items.maxsize);
-    *p_is_full = queue->items.count == queue->items.maxsize;
+    assert(queue->items.maxsize <= 0
+           || queue->items.count <= queue->items.maxsize);
+    *p_is_full = queue->items.maxsize > 0
+        ? queue->items.count == queue->items.maxsize
+        : 0;
 
     _queue_unlock(queue);
     return 0;