Function test_no_subprocess_patching

tests_python/test_debugger_json.py:5422–5493

(case_setup_multiprocessing_dap, apply_multiprocessing_patch)

@pytest.mark.parametrize("apply_multiprocessing_patch", [True])
def test_no_subprocess_patching(case_setup_multiprocessing_dap, apply_multiprocessing_patch):
    import threading
    from tests_python.debugger_unittest import AbstractWriterThread

    def update_command_line_args(writer, args):
        ret = debugger_unittest.AbstractWriterThread.update_command_line_args(writer, args)
        ret.insert(ret.index("--client"), "--multiprocess")
        ret.insert(ret.index("--client"), "--debug-mode")
        ret.insert(ret.index("--client"), "debugpy-dap")
        ret.insert(ret.index("--client"), "--json-dap-http")

        if apply_multiprocessing_patch:
            ret.append("apply-multiprocessing-patch")
        return ret

    with case_setup_multiprocessing_dap.test_file(
        "_debugger_case_no_subprocess_patching.py", update_command_line_args=update_command_line_args
    ) as writer:
        json_facade = JsonFacade(writer)
        json_facade.write_launch()

        break1_line = writer.get_line_index_with_content("break 1 here")
        break2_line = writer.get_line_index_with_content("break 2 here")
        json_facade.write_set_breakpoints([break1_line, break2_line])

        server_socket = writer.server_socket

        class SecondaryProcessWriterThread(AbstractWriterThread):
            TEST_FILE = writer.get_main_filename()
            _sequence = -1

        class SecondaryProcessThreadCommunication(threading.Thread):

            def run(self):
                from tests_python.debugger_unittest import ReaderThread

                expected_connections = 1

                for _ in range(expected_connections):
                    server_socket.listen(1)
                    self.server_socket = server_socket
                    new_sock, addr = server_socket.accept()

                    reader_thread = ReaderThread(new_sock)
                    reader_thread.name = "  *** Multiprocess Reader Thread"
                    reader_thread.start()

                    writer2 = SecondaryProcessWriterThread()
                    writer2.reader_thread = reader_thread
                    writer2.sock = new_sock
                    json_facade2 = JsonFacade(writer2)

                    json_facade2.write_set_breakpoints([break1_line, break2_line])
                    json_facade2.write_make_initial_run()

                json_facade2.wait_for_thread_stopped()
                json_facade2.write_continue()