Code Duplication    Length = 12-16 lines in 3 locations

tests/test_manhole.py 3 locations

@@ 234-247 (lines=14) @@
231
                    wait_for_strings(client.read, 1, ">>>")
232
                    client.reset()
233
                    sock.send(b"import sys\n"
234
                              b"sys.stderr.write('OK')\n")
235
                    wait_for_strings(client.read, 1, "OK")
236
237
238
def test_redirect_stderr_default_dump_stacktraces():
239
    with TestProcess(sys.executable, HELPER, 'test_redirect_stderr_default') as proc:
240
        with dump_on_error(proc.read):
241
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')
242
            check_dump_stacktraces(SOCKET_PATH)
243
244
245
def test_redirect_stderr_default_print_tracebacks():
246
    with TestProcess(sys.executable, HELPER, 'test_redirect_stderr_default') as proc:
247
        with dump_on_error(proc.read):
248
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')
249
            check_print_tracebacks(SOCKET_PATH)
250
@@ 206-217 (lines=12) @@
203
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')
204
            proc.reset()
205
            assert_manhole_running(proc, SOCKET_PATH)
206
207
208
def test_socket_path_with_fork():
209
    with TestProcess(sys.executable, '-u', HELPER, 'test_socket_path_with_fork') as proc:
210
        with dump_on_error(proc.read):
211
            wait_for_strings(proc.read, TIMEOUT, 'Not patching os.fork and os.forkpty. Using user socket path')
212
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')
213
            sock = connect_to_manhole(SOCKET_PATH)
214
            with TestSocket(sock) as client:
215
                with dump_on_error(client.read):
216
                    wait_for_strings(client.read, TIMEOUT, "ProcessID", "ThreadID", ">>>")
217
                    sock.send(b"print('BEFORE FORK')\n")
218
                    wait_for_strings(client.read, TIMEOUT, "BEFORE FORK")
219
                    time.sleep(2)
220
                    sock.send(b"print('AFTER FORK')\n")
@@ 484-499 (lines=16) @@
481
            wait_for_strings(proc.read, TIMEOUT, '/tmp/manhole-')
482
            uds_path = re.findall(r"(/tmp/manhole-\d+)", proc.read())[0]
483
            only_on_old_python = ['Waiting for new connection'] if sys.version_info < (3, 5) else []
484
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection', 'Sending signal to manhole thread',
485
                             *only_on_old_python)
486
            assert_manhole_running(proc, uds_path)
487
488
489
def test_environ_variable_activation():
490
    with TestProcess(sys.executable, '-u', HELPER, 'test_environ_variable_activation',
491
                     env=dict(os.environ, PYTHONMANHOLE="oneshot_on='USR2'")) as proc:
492
        with dump_on_error(proc.read):
493
            wait_for_strings(proc.read, TIMEOUT,
494
                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
495
            proc.signal(signal.SIGUSR2)
496
            wait_for_strings(proc.read, TIMEOUT, '/tmp/manhole-')
497
            uds_path = re.findall(r"(/tmp/manhole-\d+)", proc.read())[0]
498
            wait_for_strings(proc.read, TIMEOUT, 'Waiting for new connection')
499
            assert_manhole_running(proc, uds_path, oneshot=True)
500
501
502
@mark.skipif(not is_module_available('signalfd'), reason="signalfd not available")