bpo-32604: Implement force-closing channels. (gh-6937)

This will make it easier to clean up channels (e.g. when used in tests).

1 files changed, 96 insertions, 4 deletions

diff --git a/Lib/test/test__xxsubinterpreters.py b/Lib/test/test__xxsubinterpreters.py
index 118f2e4..f66cc95 100644
--- a/Lib/test/test__xxsubinterpreters.py
+++ b/Lib/test/test__xxsubinterpreters.py
@@ -1379,13 +1379,105 @@ class ChannelTests(TestBase):
         with self.assertRaises(interpreters.ChannelClosedError):
             interpreters.channel_close(cid)
 
-    def test_close_with_unused_items(self):
+    def test_close_empty(self):
+        tests = [
+            (False, False),
+            (True, False),
+            (False, True),
+            (True, True),
+            ]
+        for send, recv in tests:
+            with self.subTest((send, recv)):
+                cid = interpreters.channel_create()
+                interpreters.channel_send(cid, b'spam')
+                interpreters.channel_recv(cid)
+                interpreters.channel_close(cid, send=send, recv=recv)
+
+                with self.assertRaises(interpreters.ChannelClosedError):
+                    interpreters.channel_send(cid, b'eggs')
+                with self.assertRaises(interpreters.ChannelClosedError):
+                    interpreters.channel_recv(cid)
+
+    def test_close_defaults_with_unused_items(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+
+        with self.assertRaises(interpreters.ChannelNotEmptyError):
+            interpreters.channel_close(cid)
+        interpreters.channel_recv(cid)
+        interpreters.channel_send(cid, b'eggs')
+
+    def test_close_recv_with_unused_items_unforced(self):
         cid = interpreters.channel_create()
         interpreters.channel_send(cid, b'spam')
         interpreters.channel_send(cid, b'ham')
-        interpreters.channel_close(cid)
+
+        with self.assertRaises(interpreters.ChannelNotEmptyError):
+            interpreters.channel_close(cid, recv=True)
+        interpreters.channel_recv(cid)
+        interpreters.channel_send(cid, b'eggs')
+        interpreters.channel_recv(cid)
+        interpreters.channel_recv(cid)
+        interpreters.channel_close(cid, recv=True)
+
+    def test_close_send_with_unused_items_unforced(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+        interpreters.channel_close(cid, send=True)
 
         with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_send(cid, b'eggs')
+        interpreters.channel_recv(cid)
+        interpreters.channel_recv(cid)
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_recv(cid)
+
+    def test_close_both_with_unused_items_unforced(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+
+        with self.assertRaises(interpreters.ChannelNotEmptyError):
+            interpreters.channel_close(cid, recv=True, send=True)
+        interpreters.channel_recv(cid)
+        interpreters.channel_send(cid, b'eggs')
+        interpreters.channel_recv(cid)
+        interpreters.channel_recv(cid)
+        interpreters.channel_close(cid, recv=True)
+
+    def test_close_recv_with_unused_items_forced(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+        interpreters.channel_close(cid, recv=True, force=True)
+
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_send(cid, b'eggs')
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_recv(cid)
+
+    def test_close_send_with_unused_items_forced(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+        interpreters.channel_close(cid, send=True, force=True)
+
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_send(cid, b'eggs')
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_recv(cid)
+
+    def test_close_both_with_unused_items_forced(self):
+        cid = interpreters.channel_create()
+        interpreters.channel_send(cid, b'spam')
+        interpreters.channel_send(cid, b'ham')
+        interpreters.channel_close(cid, send=True, recv=True, force=True)
+
+        with self.assertRaises(interpreters.ChannelClosedError):
+            interpreters.channel_send(cid, b'eggs')
+        with self.assertRaises(interpreters.ChannelClosedError):
             interpreters.channel_recv(cid)
 
     def test_close_never_used(self):
@@ -1403,7 +1495,7 @@ class ChannelTests(TestBase):
         interp = interpreters.create()
         interpreters.run_string(interp, dedent(f"""
             import _xxsubinterpreters as _interpreters
-            _interpreters.channel_close({cid})
+            _interpreters.channel_close({cid}, force=True)
             """))
         with self.assertRaises(interpreters.ChannelClosedError):
             interpreters.channel_recv(cid)
@@ -1416,7 +1508,7 @@ class ChannelTests(TestBase):
         interpreters.channel_send(cid, b'spam')
         interpreters.channel_send(cid, b'spam')
         interpreters.channel_recv(cid)
-        interpreters.channel_close(cid)
+        interpreters.channel_close(cid, force=True)
 
         with self.assertRaises(interpreters.ChannelClosedError):
             interpreters.channel_send(cid, b'eggs')