Attach to a running process with the given PID.
(pid, commands=())
| 3456 | |
| 3457 | |
| 3458 | def attach(pid, commands=()): |
| 3459 | """Attach to a running process with the given PID.""" |
| 3460 | with ExitStack() as stack: |
| 3461 | server = stack.enter_context( |
| 3462 | closing(socket.create_server(("localhost", 0))) |
| 3463 | ) |
| 3464 | port = server.getsockname()[1] |
| 3465 | |
| 3466 | connect_script = stack.enter_context( |
| 3467 | tempfile.NamedTemporaryFile("w", delete_on_close=False) |
| 3468 | ) |
| 3469 | |
| 3470 | use_signal_thread = sys.platform == "win32" |
| 3471 | colorize = _colorize.can_colorize() |
| 3472 | |
| 3473 | connect_script.write( |
| 3474 | textwrap.dedent( |
| 3475 | f""" |
| 3476 | import pdb, sys |
| 3477 | pdb._connect( |
| 3478 | host="localhost", |
| 3479 | port={port}, |
| 3480 | frame=sys._getframe(1), |
| 3481 | commands={json.dumps("\n".join(commands))}, |
| 3482 | version={_PdbServer.protocol_version()}, |
| 3483 | signal_raising_thread={use_signal_thread!r}, |
| 3484 | colorize={colorize!r}, |
| 3485 | ) |
| 3486 | """ |
| 3487 | ) |
| 3488 | ) |
| 3489 | connect_script.close() |
| 3490 | orig_mode = os.stat(connect_script.name).st_mode |
| 3491 | os.chmod(connect_script.name, orig_mode | stat.S_IROTH | stat.S_IRGRP) |
| 3492 | sys.remote_exec(pid, connect_script.name) |
| 3493 | |
| 3494 | # TODO Add a timeout? Or don't bother since the user can ^C? |
| 3495 | client_sock, _ = server.accept() |
| 3496 | stack.enter_context(closing(client_sock)) |
| 3497 | |
| 3498 | if use_signal_thread: |
| 3499 | interrupt_sock, _ = server.accept() |
| 3500 | stack.enter_context(closing(interrupt_sock)) |
| 3501 | interrupt_sock.setblocking(False) |
| 3502 | else: |
| 3503 | interrupt_sock = None |
| 3504 | |
| 3505 | _PdbClient(pid, client_sock, interrupt_sock).cmdloop() |
| 3506 | |
| 3507 | |
| 3508 | # Post-Mortem interface |
no test coverage detected
searching dependent graphs…