Method _target_process

Lib/test/test_external_inspection.py:2708–2739

Context manager for running a target process with socket sync.

(self, script_body)

    @contextmanager
    def _target_process(self, script_body):
        """Context manager for running a target process with socket sync."""
        port = find_unused_port()
        script = f"""\
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('localhost', {port}))
{textwrap.dedent(script_body)}
"""

        with os_helper.temp_dir() as work_dir:
            script_dir = os.path.join(work_dir, "script_pkg")
            os.mkdir(script_dir)

            server_socket = _create_server_socket(port)
            script_name = _make_test_script(script_dir, "script", script)
            client_socket = None

            try:
                with _managed_subprocess([sys.executable, script_name]) as p:
                    client_socket, _ = server_socket.accept()
                    server_socket.close()
                    server_socket = None

                    def make_unwinder(cache_frames=True):
                        return RemoteUnwinder(
                            p.pid, all_threads=True, cache_frames=cache_frames
                        )

                    yield p, client_socket, make_unwinder
            finally:
                _cleanup_sockets(client_socket, server_socket)

    def _get_frames_with_retry(self, unwinder, required_funcs):
        """Get frames containing required_funcs, with retry for transient errors."""