Function test_attach_pid_client

tests/debugpy/test_attach.py:209–268

(pyfile, target, pid_type)

)
@pytest.mark.flaky(retries=2, delay=1)
def test_attach_pid_client(pyfile, target, pid_type):
    @pyfile
    def code_to_debug():
        import debuggee
        import time

        debuggee.setup()

        def do_something(i):
            time.sleep(0.2)
            proceed = True
            print(i)  # @bp
            return proceed

        for i in range(500):
            if not do_something(i):
                break

    def before_request(command, arguments):
        if command == "attach":
            assert isinstance(arguments["processId"], int)
            if pid_type == "str":
                arguments["processId"] = str(arguments["processId"])

    session1 = debug.Session()

    session1.before_request = before_request
    session1.config["redirectOutput"] = True

    session1.captured_output = set()
    session1.expected_exit_code = None  # not expected to exit on disconnect

    with session1.attach_pid(target(code_to_debug), wait=False):
        session1.set_breakpoints(code_to_debug, all)

    session1.wait_for_stop(expected_frames=[some.dap.frame(code_to_debug, "bp")])

    pid = session1.config["processId"]

    # Note: don't call session1.disconnect because it'd deadlock in channel.close()
    # (because the fd is in a read() in a different thread, we can't call close() on it).
    session1.request("disconnect")
    session1.wait_for_terminated()

    with debug.Session() as session2:
        with session2.attach_pid(pid, wait=False):
            session2.set_breakpoints(code_to_debug, all)

        stop = session2.wait_for_stop(
            expected_frames=[some.dap.frame(code_to_debug, "bp")]
        )

        # Remove breakpoint and continue.
        session2.set_breakpoints(code_to_debug, [])
        session2.request(
            "setExpression",
            {"frameId": stop.frame_id, "expression": "proceed", "value": "False"},
        )