|
1 | 1 | import contextlib |
2 | 2 | import os |
3 | 3 | import pickle |
| 4 | +import signal |
4 | 5 | import sys |
5 | 6 | from textwrap import dedent |
6 | 7 | import threading |
|
11 | 12 | from test.support import os_helper |
12 | 13 | from test.support import script_helper |
13 | 14 | from test.support import import_helper |
14 | | -from test.support.script_helper import assert_python_ok |
| 15 | +from test.support.script_helper import assert_python_ok, spawn_python |
15 | 16 | # Raise SkipTest if subinterpreters not supported. |
16 | 17 | _interpreters = import_helper.import_module('_interpreters') |
17 | 18 | from concurrent import interpreters |
@@ -434,6 +435,32 @@ def test_cleanup_in_repl(self): |
434 | 435 | self.assertIn(b"remaining subinterpreters", stdout) |
435 | 436 | self.assertNotIn(b"Traceback", stdout) |
436 | 437 |
|
| 438 | + @support.requires_subprocess() |
| 439 | + @unittest.skipIf(os.name == 'nt', "signals don't work well on windows") |
| 440 | + def test_keyboard_interrupt_in_thread_running_interp(self): |
| 441 | + import subprocess |
| 442 | + source = f"""if True: |
| 443 | + from concurrent import interpreters |
| 444 | + from threading import Thread |
| 445 | +
|
| 446 | + def test(): |
| 447 | + import time |
| 448 | + print('a', flush=True, end='') |
| 449 | + time.sleep(10) |
| 450 | +
|
| 451 | + interp = interpreters.create() |
| 452 | + interp.call_in_thread(test) |
| 453 | + """ |
| 454 | + |
| 455 | + with spawn_python("-c", source, stderr=subprocess.PIPE) as proc: |
| 456 | + self.assertEqual(proc.stdout.read(1), b'a') |
| 457 | + proc.send_signal(signal.SIGINT) |
| 458 | + proc.stderr.flush() |
| 459 | + error = proc.stderr.read() |
| 460 | + self.assertIn(b"KeyboardInterrupt", error) |
| 461 | + retcode = proc.wait() |
| 462 | + self.assertEqual(retcode, 0) |
| 463 | + |
437 | 464 |
|
438 | 465 |
|
439 | 466 | class TestInterpreterIsRunning(TestBase): |
|
0 commit comments