ionelmc / python-manhole

def test_with_fork():

    with TestProcess(sys.executable, '-u', HELPER, 'test_with_fork') as proc:

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT, '/tmp/manhole-')

            uds_path = re.findall(r"(/tmp/manhole-\d+)", proc.read())[0]

            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')

            for _ in range(2):

                proc.reset()

                assert_manhole_running(proc, uds_path)

            proc.reset()

            wait_for_strings(proc.read, TIMEOUT, 'Fork detected')

            wait_for_strings(proc.read, TIMEOUT, '/tmp/manhole-')

            new_uds_path = re.findall(r"(/tmp/manhole-\d+)", proc.read())[0]

            assert uds_path != new_uds_path

            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')

            for _ in range(2):

                proc.reset()

                assert_manhole_running(proc, new_uds_path)

@mark.skipif(hasattr(sys, 'pypy_version_info'), reason="pypy doesn't support forkpty")

def test_with_forkpty():

    with TestProcess(sys.executable, '-u', HELPER, 'test_with_forkpty') as proc:

            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

            sock.settimeout(0.05)

            sock.connect(uds_path)

            with TestSocket(sock) as client:

                with dump_on_error(client.read):

                    wait_for_strings(client.read, TIMEOUT, "ThreadID", "ProcessID", ">>>")

                    sock.send(b"print('FOOBAR')\n")

                    wait_for_strings(client.read, TIMEOUT, "FOOBAR")

                    wait_for_strings(proc.read, TIMEOUT, 'UID:%s' % os.getuid())

                    sock.shutdown(socket.SHUT_WR)

                    select.select([sock], [], [], 5)

                    sock.recv(1024)

                    try:

                        sock.shutdown(socket.SHUT_RD)

                    except Exception as exc:

                        print("Failed to SHUT_RD: %s" % exc)

                    try:

                        sock.close()

                    except Exception as exc:

                        print("Failed to close socket: %s" % exc)

            wait_for_strings(proc.read, TIMEOUT, 'DONE.', 'Cleaned up.', 'Waiting for new connection')

def test_with_fork():

    with TestProcess(sys.executable, '-u', HELPER, 'test_with_fork') as proc: