|
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,31 @@ 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) |
437 | 463 |
|
438 | 464 |
|
439 | 465 | class TestInterpreterIsRunning(TestBase):
|
|
0 commit comments