Function test_multiprocessing_simple

tests_python/test_debugger.py:2793–2846

(case_setup_multiprocessing, file_to_check)

    ],
)
def test_multiprocessing_simple(case_setup_multiprocessing, file_to_check):
    import threading
    from tests_python.debugger_unittest import AbstractWriterThread

    with case_setup_multiprocessing.test_file(file_to_check) as writer:
        break1_line = writer.get_line_index_with_content("break 1 here")
        break2_line = writer.get_line_index_with_content("break 2 here")

        writer.write_add_breakpoint(break1_line)
        writer.write_add_breakpoint(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 = 2 if sys.platform == "darwin" else 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

                    writer2.write_version()
                    writer2.write_add_breakpoint(break1_line)
                    writer2.write_add_breakpoint(break2_line)
                    writer2.write_make_initial_run()

                hit = writer2.wait_for_breakpoint_hit()
                writer2.write_run_thread(hit.thread_id)

        secondary_process_thread_communication = SecondaryProcessThreadCommunication()
        secondary_process_thread_communication.start()
        writer.write_make_initial_run()
        hit2 = writer.wait_for_breakpoint_hit()
        secondary_process_thread_communication.join(10)
        if secondary_process_thread_communication.is_alive():
            raise AssertionError("The SecondaryProcessThreadCommunication did not finish")
        writer.write_run_thread(hit2.thread_id)
        writer.finished_ok = True


@pytest.mark.skipif(not IS_CPYTHON, reason="CPython only test.")