ionelmc / python-manhole

            sock = connect_to_manhole(SOCKET_PATH)

            with TestSocket(sock) as client:

                with dump_on_error(client.read):

                    wait_for_strings(client.read, 1, ">>>")

                    client.reset()

                    sock.send(b"import sys\n"

                              b"sys.stderr.write('OK')\n")

                    wait_for_strings(client.read, 1, "OK")

def test_redirect_stderr_default_dump_stacktraces():

    with TestProcess(sys.executable, HELPER, 'test_redirect_stderr_default') as proc:

        with dump_on_error(proc.read):

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

            check_dump_stacktraces(SOCKET_PATH)

def test_redirect_stderr_default_print_tracebacks():

    with TestProcess(sys.executable, HELPER, 'test_redirect_stderr_default') as proc:

        with dump_on_error(proc.read):

def test_socket_path():

    with TestProcess(sys.executable, HELPER, 'test_socket_path') as proc:

        with dump_on_error(proc.read):

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

            proc.reset()

            assert_manhole_running(proc, SOCKET_PATH)

def test_socket_path_with_fork():

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

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT, 'Not patching os.fork and os.forkpty. Using user socket path')

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

            sock = connect_to_manhole(SOCKET_PATH)

            with TestSocket(sock) as client:

                with dump_on_error(client.read):

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

                    sock.send(b"print('BEFORE FORK')\n")

def test_interrupt_on_accept():

    with TestProcess(sys.executable, '-u', HELPER, 'test_interrupt_on_accept') 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]

            only_on_old_python = ['Waiting for new connection'] if sys.version_info < (3, 5) else []

            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection', 'Sending signal to manhole thread',

                             *only_on_old_python)

            assert_manhole_running(proc, uds_path)

def test_environ_variable_activation():

    with TestProcess(sys.executable, '-u', HELPER, 'test_environ_variable_activation',

                     env=dict(os.environ, PYTHONMANHOLE="oneshot_on='USR2'")) as proc:

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT,

                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')

            proc.signal(signal.SIGUSR2)

            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')

            assert_manhole_running(proc, uds_path, oneshot=True)