Function test_terminate

tests_python/test_debugger_json.py:5570–5681

(case_setup_dap, scenario, check_subprocesses)

    ],
)
def test_terminate(case_setup_dap, scenario, check_subprocesses):
    import psutil

    def check_test_suceeded_msg(writer, stdout, stderr):
        return "TEST SUCEEDED" not in "".join(stdout)

    def update_command_line_args(writer, args):
        ret = debugger_unittest.AbstractWriterThread.update_command_line_args(writer, args)
        if check_subprocesses in ("kill_subprocesses", "dont_kill_subprocesses"):
            ret.append("check-subprocesses")
        if check_subprocesses in ("kill_subprocesses_ignore_pid",):
            ret.append("check-subprocesses-ignore-pid")
        return ret

    with case_setup_dap.test_file(
        "_debugger_case_terminate.py",
        check_test_suceeded_msg=check_test_suceeded_msg,
        update_command_line_args=update_command_line_args,
        EXPECTED_RETURNCODE="any" if check_subprocesses == "kill_subprocesses_ignore_pid" else 0,
    ) as writer:
        json_facade = JsonFacade(writer)
        if check_subprocesses == "dont_kill_subprocesses":
            json_facade.write_launch(terminateChildProcesses=False)

        json_facade.write_make_initial_run()
        response = json_facade.write_initialize()
        pid = response.to_dict()["body"]["pydevd"]["processId"]

        if check_subprocesses in ("kill_subprocesses", "dont_kill_subprocesses", "kill_subprocesses_ignore_pid"):
            process_ids_to_check = [pid]
            p = psutil.Process(pid)

            def wait_for_child_processes():
                children = p.children(recursive=True)
                found = len(children)
                if found == 8:
                    process_ids_to_check.extend([x.pid for x in children])
                    return True
                return False

            wait_for_condition(wait_for_child_processes)

        if scenario == "terminate_request":
            json_facade.write_terminate()
        elif scenario == "terminate_debugee":
            json_facade.write_disconnect(terminate_debugee=True)
        else:
            raise AssertionError("Unexpected: %s" % (scenario,))
        json_facade.wait_for_terminated()

        if check_subprocesses in ("kill_subprocesses", "dont_kill_subprocesses", "kill_subprocesses_ignore_pid"):

            def is_pid_alive(pid):
                # Note: the process may be a zombie process in Linux
                # (althought it's killed it remains in that state
                # because we're monitoring it).
                try:
                    proc = psutil.Process(pid)