test_pid() - Code Metrics - ionelmc/python-manhole - Measure and Improve Code Quality continuously with Scrutinizer

test_pid() B
last analyzed 2017-12-28 16:33 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	5
c	2
b	0
f	0
dl	0
loc	9
rs	8.5454

from __future__ import print_function

import os
import signal
import sys

import pytest
from process_tests import TestProcess
from process_tests import dump_on_error
from process_tests import wait_for_strings

try:
    import subprocess32 as subprocess
except ImportError:
    import subprocess

TIMEOUT = int(os.getenv('MANHOLE_TEST_TIMEOUT', 10))
HELPER = os.path.join(os.path.dirname(__file__), 'helper.py')


def test_pid_validation():
    exc = pytest.raises(subprocess.CalledProcessError, subprocess.check_output, ['manhole-cli', 'asdfasdf'],
                        stderr=subprocess.STDOUT)
    assert exc.value.output == b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 | -2 | -s SIGNAL] PID
manhole-cli: error: argument PID: PID must be in one of these forms: 1234 or /tmp/manhole-1234
"""


def test_sig_number_validation():
    exc = pytest.raises(subprocess.CalledProcessError, subprocess.check_output,
                        ['manhole-cli', '-s', '12341234', '12341234'], stderr=subprocess.STDOUT)
    assert exc.value.output.startswith(b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 | -2 | -s SIGNAL] PID
manhole-cli: error: argument -s/--signal: Invalid signal number 12341234. Expected one of: """)


def test_help():
    output = subprocess.check_output(['manhole-cli', '--help'])
    print(output)
    assert output == b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 | -2 | -s SIGNAL] PID

Connect to a manhole.

positional arguments:
  PID                   A numerical process id, or a path in the form:
                        /tmp/manhole-1234

optional arguments:
  -h, --help            show this help message and exit
  -t TIMEOUT, --timeout TIMEOUT
                        Timeout to use. Default: 1 seconds.
  -1, -USR1             Send USR1 (10) to the process before connecting.
  -2, -USR2             Send USR2 (12) to the process before connecting.
  -s SIGNAL, --signal SIGNAL
                        Send the given SIGNAL to the process before
                        connecting.
"""


def test_usr2():
    with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT,
                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
            with TestProcess('manhole-cli', '-USR2', str(service.proc.pid), bufsize=0, stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


def test_pid():
    with TestProcess(sys.executable, HELPER, 'test_simple') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT, '/tmp/manhole-')
            with TestProcess('manhole-cli', str(service.proc.pid), bufsize=0, stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


def test_path():

    with TestProcess(sys.executable, HELPER, 'test_simple') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT, '/tmp/manhole-')
            with TestProcess('manhole-cli', '/tmp/manhole-%s' % service.proc.pid, bufsize=0,
                             stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


def test_sig_usr2():

    with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT,
                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
            with TestProcess('manhole-cli', '--signal=USR2', str(service.proc.pid), bufsize=0,
                             stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


def test_sig_usr2_full():
    with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT,
                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
            with TestProcess('manhole-cli', '-s', 'SIGUSR2', str(service.proc.pid), bufsize=0,
                             stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


def test_sig_usr2_number():

    with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
        with dump_on_error(service.read):
            wait_for_strings(service.read, TIMEOUT,
                             'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
            with TestProcess('manhole-cli', '-s', str(int(signal.SIGUSR2)), str(service.proc.pid), bufsize=0,
                             stdin=subprocess.PIPE) as client:
                with dump_on_error(client.read):
                    wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
                    client.proc.stdin.write(b"1234+2345\n")
                    wait_for_strings(client.read, TIMEOUT, '3579')


1		from __future__ import print_function
2
3		import os
4		import signal
5		import sys
6
7		import pytest
8		from process_tests import TestProcess
9		from process_tests import dump_on_error
10		from process_tests import wait_for_strings
11
12		try:
13		import subprocess32 as subprocess
14		except ImportError:
15		import subprocess
16
17		TIMEOUT = int(os.getenv('MANHOLE_TEST_TIMEOUT', 10))
18		HELPER = os.path.join(os.path.dirname(__file__), 'helper.py')
19
20
21		def test_pid_validation():
22		exc = pytest.raises(subprocess.CalledProcessError, subprocess.check_output, ['manhole-cli', 'asdfasdf'],
23		stderr=subprocess.STDOUT)
24		assert exc.value.output == b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 \| -2 \| -s SIGNAL] PID
25		manhole-cli: error: argument PID: PID must be in one of these forms: 1234 or /tmp/manhole-1234
26		"""
27
28
29		def test_sig_number_validation():
30		exc = pytest.raises(subprocess.CalledProcessError, subprocess.check_output,
31		['manhole-cli', '-s', '12341234', '12341234'], stderr=subprocess.STDOUT)
32		assert exc.value.output.startswith(b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 \| -2 \| -s SIGNAL] PID
33		manhole-cli: error: argument -s/--signal: Invalid signal number 12341234. Expected one of: """)
34
35
36		def test_help():
37		output = subprocess.check_output(['manhole-cli', '--help'])
38		print(output)
39		assert output == b"""usage: manhole-cli [-h] [-t TIMEOUT] [-1 \| -2 \| -s SIGNAL] PID
40
41		Connect to a manhole.
42
43		positional arguments:
44		PID A numerical process id, or a path in the form:
45		/tmp/manhole-1234
46
47		optional arguments:
48		-h, --help show this help message and exit
49		-t TIMEOUT, --timeout TIMEOUT
50		Timeout to use. Default: 1 seconds.
51		-1, -USR1 Send USR1 (10) to the process before connecting.
52		-2, -USR2 Send USR2 (12) to the process before connecting.
53		-s SIGNAL, --signal SIGNAL
54		Send the given SIGNAL to the process before
55		connecting.
56		"""
57
58
59		def test_usr2():
60		with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
61		with dump_on_error(service.read):
62		wait_for_strings(service.read, TIMEOUT,
63		'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
64		with TestProcess('manhole-cli', '-USR2', str(service.proc.pid), bufsize=0, stdin=subprocess.PIPE) as client:
65		with dump_on_error(client.read):
66		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
67		client.proc.stdin.write(b"1234+2345\n")
68		wait_for_strings(client.read, TIMEOUT, '3579')
69
70
71		def test_pid():
72		with TestProcess(sys.executable, HELPER, 'test_simple') as service:
73		with dump_on_error(service.read):
74		wait_for_strings(service.read, TIMEOUT, '/tmp/manhole-')
75		with TestProcess('manhole-cli', str(service.proc.pid), bufsize=0, stdin=subprocess.PIPE) as client:
76		with dump_on_error(client.read):
77		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
78		client.proc.stdin.write(b"1234+2345\n")
79		wait_for_strings(client.read, TIMEOUT, '3579')
80
81
82	View Code Duplication	def test_path():
		0 ignored issues – show Duplication introduced 2016-06-11 00:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
83		with TestProcess(sys.executable, HELPER, 'test_simple') as service:
84		with dump_on_error(service.read):
85		wait_for_strings(service.read, TIMEOUT, '/tmp/manhole-')
86		with TestProcess('manhole-cli', '/tmp/manhole-%s' % service.proc.pid, bufsize=0,
87		stdin=subprocess.PIPE) as client:
88		with dump_on_error(client.read):
89		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
90		client.proc.stdin.write(b"1234+2345\n")
91		wait_for_strings(client.read, TIMEOUT, '3579')
92
93
94	View Code Duplication	def test_sig_usr2():
		0 ignored issues – show Duplication introduced 2016-06-11 00:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
95		with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
96		with dump_on_error(service.read):
97		wait_for_strings(service.read, TIMEOUT,
98		'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
99		with TestProcess('manhole-cli', '--signal=USR2', str(service.proc.pid), bufsize=0,
100		stdin=subprocess.PIPE) as client:
101		with dump_on_error(client.read):
102		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
103		client.proc.stdin.write(b"1234+2345\n")
104		wait_for_strings(client.read, TIMEOUT, '3579')
105
106
107		def test_sig_usr2_full():
108		with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
109		with dump_on_error(service.read):
110		wait_for_strings(service.read, TIMEOUT,
111		'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
112		with TestProcess('manhole-cli', '-s', 'SIGUSR2', str(service.proc.pid), bufsize=0,
113		stdin=subprocess.PIPE) as client:
114		with dump_on_error(client.read):
115		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
116		client.proc.stdin.write(b"1234+2345\n")
117		wait_for_strings(client.read, TIMEOUT, '3579')
118
119
120	View Code Duplication	def test_sig_usr2_number():
		0 ignored issues – show Duplication introduced 2016-06-11 00:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
121		with TestProcess(sys.executable, '-u', HELPER, 'test_oneshot_on_usr2') as service:
122		with dump_on_error(service.read):
123		wait_for_strings(service.read, TIMEOUT,
124		'Not patching os.fork and os.forkpty. Oneshot activation is done by signal')
125		with TestProcess('manhole-cli', '-s', str(int(signal.SIGUSR2)), str(service.proc.pid), bufsize=0,
126		stdin=subprocess.PIPE) as client:
127		with dump_on_error(client.read):
128		wait_for_strings(client.read, TIMEOUT, '(ManholeConsole)', '>>>')
129		client.proc.stdin.write(b"1234+2345\n")
130		wait_for_strings(client.read, TIMEOUT, '3579')
131

ionelmc / python-manhole

test_pid() B last analyzed 2017-12-28 16:33 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

test_pid() B
last analyzed 2017-12-28 16:33 UTC