ionelmc / python-manhole

            assert_manhole_running(proc, uds_path, oneshot=True,

                                   extra=lambda client: client.sock.send(b"raise SystemExit()\n"))

            proc.reset()

            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)

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 []

def test_activate_on_usr2():

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

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT, 'Not patching os.fork and os.forkpty. Activation is done by signal')

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

            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)

def test_activate_on_with_oneshot_on():

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

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT,