ionelmc / python-manhole

@mark.skipif(is_module_available('eventlet'), reason="evenlet can't deal with extra threads at process exit")

def test_daemon_connection():

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

            def terminate_and_read(client):

                proc.proc.send_signal(signal.SIGINT)

                wait_for_strings(proc.read, TIMEOUT, 'Died with KeyboardInterrupt', 'DIED.')

                for _ in range(5):

                    client.sock.send(b'bogus()\n')

                    time.sleep(0.05)

                    print(repr(client.sock.recv(1024)))

            raises((socket.error, OSError), assert_manhole_running, proc, uds_path, extra=terminate_and_read)

            wait_for_strings(proc.read, TIMEOUT, 'In atexit handler')

@mark.skipif(is_module_available('eventlet'), reason="evenlet can't deal with extra threads at process exit")

def test_non_daemon_connection():

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

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

                        wait_for_strings(proc.read, TIMEOUT, 'FOOBAR')

def test_install_once():

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

        with dump_on_error(proc.read):

            wait_for_strings(proc.read, TIMEOUT, 'ALREADY_INSTALLED')

def test_install_twice_not_strict():

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

            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)

@mark.skipif(is_module_available('eventlet'), reason="evenlet can't deal with extra threads at process exit")

def test_daemon_connection():

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