|
6 | 6 | from test.support import threading_helper, requires_subprocess, requires_gil_enabled |
7 | 7 | from test.support import verbose, cpython_only, os_helper |
8 | 8 | from test.support.import_helper import ensure_lazy_imports, import_module |
9 | | -from test.support.script_helper import assert_python_ok, assert_python_failure |
| 9 | +from test.support.script_helper import assert_python_ok, assert_python_failure, spawn_python |
10 | 10 | from test.support import force_not_colorized |
11 | 11 |
|
12 | 12 | import random |
@@ -2083,6 +2083,32 @@ def test_dummy_thread_on_interpreter_shutdown(self): |
2083 | 2083 | self.assertEqual(out, b"") |
2084 | 2084 | self.assertEqual(err, b"") |
2085 | 2085 |
|
| 2086 | + @requires_subprocess() |
| 2087 | + def test_keyboard_interrupt_during_threading_shutdown(self): |
| 2088 | + import subprocess |
| 2089 | + source = f""" |
| 2090 | + from threading import Thread |
| 2091 | + import time |
| 2092 | + import os |
| 2093 | +
|
| 2094 | +
|
| 2095 | + def test(): |
| 2096 | + print('a', flush=True, end='') |
| 2097 | + time.sleep(10) |
| 2098 | +
|
| 2099 | +
|
| 2100 | + for _ in range(3): |
| 2101 | + Thread(target=test).start() |
| 2102 | + """ |
| 2103 | + |
| 2104 | + with spawn_python("-c", source, stderr=subprocess.PIPE) as proc: |
| 2105 | + self.assertEqual(proc.stdout.read(3), b'aaa') |
| 2106 | + proc.send_signal(signal.SIGINT) |
| 2107 | + proc.stderr.flush() |
| 2108 | + error = proc.stderr.read() |
| 2109 | + self.assertIn(b"KeyboardInterrupt", error) |
| 2110 | + self.assertIn(b"threading shutdown", error) |
| 2111 | + |
2086 | 2112 |
|
2087 | 2113 | class ThreadRunFail(threading.Thread): |
2088 | 2114 | def run(self): |
|
0 commit comments