Method _run_remote_exec_test

Lib/test/test_sys.py:1995–2089

(self, script_code, python_args=None, env=None,
                              prologue='',
                              script_path=os_helper.TESTFN + '_remote.py')

        test.support.reap_children()

    def _run_remote_exec_test(self, script_code, python_args=None, env=None,
                              prologue='',
                              script_path=os_helper.TESTFN + '_remote.py'):
        # Create the script that will be remotely executed
        self.addCleanup(os_helper.unlink, script_path)

        with open(script_path, 'w') as f:
            f.write(script_code)

        # Create and run the target process
        target = os_helper.TESTFN + '_target.py'
        self.addCleanup(os_helper.unlink, target)

        port = find_unused_port()

        with open(target, 'w') as f:
            f.write(f'''
import sys
import time
import socket

# Connect to the test process
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('localhost', {port}))

{prologue}

# Signal that the process is ready
sock.sendall(b"ready")

print("Target process running...")

# Wait for remote script to be executed
# (the execution will happen as the following
# code is processed as soon as the recv call
# unblocks)
sock.recv(1024)

# Do a bunch of work to give the remote script time to run
x = 0
for i in range(100):
    x += i

# Write confirmation back
sock.sendall(b"executed")
sock.close()
''')

        # Start the target process and capture its output
        cmd = [sys.executable]
        if python_args:
            cmd.extend(python_args)
        cmd.append(target)

        # Create a socket server to communicate with the target process
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.bind(('localhost', port))
        server_socket.settimeout(SHORT_TIMEOUT)