Show More
@@ -11,6 +11,8 import warnings | |||
|
11 | 11 | from tempfile import NamedTemporaryFile |
|
12 | 12 | from subprocess import check_output, CalledProcessError, PIPE |
|
13 | 13 | import subprocess |
|
14 | from unittest.mock import patch | |
|
15 | import builtins | |
|
14 | 16 | |
|
15 | 17 | import nose.tools as nt |
|
16 | 18 | |
@@ -227,72 +229,20 def can_exit(): | |||
|
227 | 229 | ''' |
|
228 | 230 | |
|
229 | 231 | |
|
230 | interruptible_debugger = """\ | |
|
231 | import sys | |
|
232 | import threading | |
|
233 | import time | |
|
234 | from os import _exit | |
|
235 | from bdb import BdbQuit | |
|
236 | ||
|
237 | from IPython.core.debugger import set_trace | |
|
238 | ||
|
239 | # Timeout if the interrupt doesn't happen: | |
|
240 | def timeout(): | |
|
241 | time.sleep(5) | |
|
242 | _exit(7) | |
|
243 | threading.Thread(target=timeout, daemon=True).start() | |
|
244 | ||
|
245 | def break_handler(*args): | |
|
246 | print("BREAK!") | |
|
247 | raise KeyboardInterrupt() | |
|
248 | ||
|
249 | def main(): | |
|
250 | import signal | |
|
251 | signal.signal(signal.SIGBREAK, break_handler) | |
|
252 | set_trace() | |
|
253 | ||
|
254 | if __name__ == '__main__': | |
|
255 | try: | |
|
256 | print("Starting debugger...") | |
|
257 | main() | |
|
258 | print("Debugger exited without error.") | |
|
259 | except (KeyboardInterrupt, BdbQuit): | |
|
260 | print("Caught KeyboardInterrupt or BdbQuit, PASSED") | |
|
261 | except Exception as e: | |
|
262 | print("Got wrong exception...") | |
|
263 | raise e | |
|
264 | """ | |
|
265 | ||
|
266 | ||
|
267 | 232 | def test_interruptible_core_debugger(): |
|
268 | 233 | """The debugger can be interrupted. |
|
269 | ||
|
270 | See https://stackoverflow.com/a/35792192 for details on Windows. | |
|
234 | ||
|
235 | The presumption is there is some mechanism that causes a KeyboardInterrupt | |
|
236 | (this is implemented in ipykernel). We want to ensure the | |
|
237 | KeyboardInterrupt cause debugging to cease. | |
|
271 | 238 | """ |
|
272 | with NamedTemporaryFile("w", delete=False) as f: | |
|
273 | f.write(interruptible_debugger) | |
|
274 | f.flush() | |
|
275 | start = time.time() | |
|
276 | ||
|
277 | p = subprocess.Popen([sys.executable, "-u", f.name], | |
|
278 | creationflags=subprocess.CREATE_NEW_PROCESS_GROUP, # TODO disable on posix | |
|
279 | encoding=sys.getdefaultencoding(), | |
|
280 | stderr=PIPE, stdout=PIPE) | |
|
281 | time.sleep(1) # wait for it to hit pdb | |
|
282 | if sys.platform == "win32": | |
|
283 | # Yes, this has to happen once. I have no idea why. | |
|
284 | p.send_signal(signal.CTRL_BREAK_EVENT) | |
|
285 | p.send_signal(signal.CTRL_BREAK_EVENT) | |
|
286 | else: | |
|
287 | p.send_signal(signal.SIGINT) | |
|
288 | exit_code = p.wait() | |
|
289 | stdout = p.stdout.read() | |
|
290 | stderr = p.stderr.read() | |
|
291 | print("STDOUT", stdout, file=sys.__stderr__) | |
|
292 | print("STDERR", stderr, file=sys.__stderr__) | |
|
293 | assert exit_code == 0 | |
|
294 | print("SUCCESS!", file=sys.__stderr__) | |
|
295 | # Make sure it exited cleanly, and quickly: | |
|
296 | end = time.time() | |
|
297 | assert end - start < 2 # timeout is 5 seconds | |
|
298 | assert "PASSED" in stdout No newline at end of file | |
|
239 | def raising_input(msg=""): | |
|
240 | raise KeyboardInterrupt() | |
|
241 | ||
|
242 | with patch.object(builtins, "input", raising_input): | |
|
243 | debugger.set_trace() | |
|
244 | # The way this test will fail is by set_trace() never exiting, | |
|
245 | # resulting in a timeout by the test runner. The alternative | |
|
246 | # implementation would involve a subprocess, but that adds issues with | |
|
247 | # interrupting subprocesses that are rather complex, so it's simpler | |
|
248 | # just to do it this way. |
General Comments 0
You need to be logged in to leave comments.
Login now