bpo-23057: Use 'raise' to emulate ctrl-c in proactor tests (#11274)

2 files changed, 19 insertions, 76 deletions

diff --git a/Lib/test/test_asyncio/test_ctrl_c_in_proactor_loop_helper.py b/Lib/test/test_asyncio/test_ctrl_c_in_proactor_loop_helper.py
deleted file mode 100644
index 9aeb58a..0000000
--- a/Lib/test/test_asyncio/test_ctrl_c_in_proactor_loop_helper.py
+++ /dev/null
@@ -1,63 +0,0 @@
-import sys
-
-
-def do_in_child_process():
-    import asyncio
-
-    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
-    l = asyncio.get_event_loop()
-
-    def step(n):
-        try:
-            print(n)
-            sys.stdout.flush()
-            l.run_forever()
-            sys.exit(100)
-        except KeyboardInterrupt:
-            # ok
-            pass
-        except:
-            # error - use default exit code
-            sys.exit(200)
-
-    step(1)
-    step(2)
-    sys.exit(255)
-
-
-def do_in_main_process():
-    import os
-    import signal
-    import subprocess
-    import time
-    from test.support.script_helper import spawn_python
-
-    ok = False
-
-    def step(p, expected):
-        s = p.stdout.readline()
-        if s != expected:
-            raise Exception(f"Unexpected line: got {s}, expected '{expected}'")
-        # ensure that child process gets to run_forever
-        time.sleep(0.5)
-        os.kill(p.pid, signal.CTRL_C_EVENT)
-
-    with spawn_python(__file__, "--child") as p:
-        try:
-            # ignore ctrl-c in current process
-            signal.signal(signal.SIGINT, signal.SIG_IGN)
-            step(p, b"1\r\n")
-            step(p, b"2\r\n")
-            exit_code = p.wait(timeout=5)
-            ok = exit_code = 255
-        except Exception as e:
-            sys.stderr.write(repr(e))
-            p.kill()
-    sys.exit(255 if ok else 1)
-
-
-if __name__ == "__main__":
-    if len(sys.argv) == 1:
-        do_in_main_process()
-    else:
-        do_in_child_process()
diff --git a/Lib/test/test_asyncio/test_windows_events.py b/Lib/test/test_asyncio/test_windows_events.py
index 05d875a..a200a8a 100644
--- a/Lib/test/test_asyncio/test_windows_events.py
+++ b/Lib/test/test_asyncio/test_windows_events.py
@@ -4,6 +4,7 @@ import socket
 import sys
 import subprocess
 import time
+import threading
 import unittest
 from unittest import mock
 
@@ -11,6 +12,7 @@ if sys.platform != 'win32':
     raise unittest.SkipTest('Windows only')
 
 import _overlapped
+import _testcapi
 import _winapi
 
 import asyncio
@@ -38,20 +40,24 @@ class UpperProto(asyncio.Protocol):
 
 
 class ProactorLoopCtrlC(test_utils.TestCase):
+
     def test_ctrl_c(self):
-        from .test_ctrl_c_in_proactor_loop_helper import __file__ as f
-
-        # ctrl-c will be sent to all processes that share the same console
-        # in order to isolate the effect of raising ctrl-c we'll create
-        # a process with a new console
-        flags = subprocess.CREATE_NEW_CONSOLE
-        with spawn_python(f, creationflags=flags) as p:
-            try:
-                exit_code = p.wait(timeout=5)
-                self.assertEqual(exit_code, 255)
-            except:
-                p.kill()
-                raise
+
+        def SIGINT_after_delay():
+            time.sleep(1)
+            _testcapi.raise_signal(signal.SIGINT)
+
+        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+        l = asyncio.get_event_loop()
+        try:
+            t = threading.Thread(target=SIGINT_after_delay)
+            t.start()
+            l.run_forever()
+            self.fail("should not fall through 'run_forever'")
+        except KeyboardInterrupt:
+            pass
+        finally:
+            l.close()
 
 
 class ProactorTests(test_utils.TestCase):