Function test_subprocess_unobserved

tests/debugpy/test_multiproc.py:445–499

(pyfile, run, target, wait)

@pytest.mark.parametrize("run", runners.all_attach_connect)
@pytest.mark.parametrize("wait", ["wait", ""])
def test_subprocess_unobserved(pyfile, run, target, wait):
    @pyfile
    def child():
        from debuggee import backchannel  # @bp

        backchannel.send("child running")
        backchannel.receive()

    @pyfile
    def parent():
        import debuggee
        import os
        import subprocess
        import sys

        debuggee.setup()
        args = [sys.executable, sys.argv[1]]
        env = os.environ.copy()
        subprocess.Popen(args, env=env).wait()

    with debug.Session() as parent_session:
        backchannel = parent_session.open_backchannel()

        if not wait:
            # The child process should have started running user code as soon as it's
            # spawned, before there is a client connection.
            def before_connect(address):
                assert backchannel.receive() == "child running"

            parent_session.before_connect = before_connect

        with run(parent_session, target(parent, args=[child]), wait=bool(wait)):
            pass

        child_config = parent_session.wait_for_next_event("debugpyAttach")
        parent_session.proceed()

        with debug.Session(child_config) as child_session:
            with child_session.start():
                child_session.set_breakpoints(child, all)

            if wait:
                # The child process should not have started running user code until
                # there was a client connection, so the breakpoint should be hit.
                child_session.wait_for_stop(
                    expected_frames=[some.dap.frame(child, line="bp")]
                )
                child_session.request_continue()
            else:
                # The breakpoint shouldn't be hit, since that line should have been
                # executed before we attached.
                pass

            backchannel.send("proceed")
            child_session.wait_for_terminated()


@pytest.mark.parametrize("run", runners.all_launch)