NeonXP/dumb-init
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #89 from chriskuehl/clean-up-tests

Browse source 
Clean up tests a bit
This commit is contained in:
Anthony Sottile 2016-06-17 15:58:46 -07:00 committed by GitHub
commit 843626f152
9 changed files with 118 additions and 131 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
tests/lib/testing.py → testing/__init__.py
View file

@ -1,5 +1,10 @@
import os
import re
import signal
import sys
from contextlib import contextmanager
from subprocess import PIPE
from subprocess import Popen
from py._path.local import LocalPath
@ -18,6 +23,27 @@ NORMAL_SIGNALS = frozenset(
)
@contextmanager
def print_signals(args=()):
"""Start print_signals and yield dumb-init process and print_signals PID."""
proc = Popen(
(
('dumb-init',) +
tuple(args) +
(sys.executable, '-m', 'testing.print_signals')
),
stdout=PIPE,
)
line = proc.stdout.readline()
m = re.match(b'^ready \(pid: ([0-9]+)\)\n$', line)
assert m, line
yield proc, m.group(1).decode('ascii')
for pid in pid_tree(proc.pid):
os.kill(pid, signal.SIGKILL)
def child_pids(pid):
"""Return a list of direct child PIDs for the given PID."""
pid = str(pid)

0
tests/lib/print_signals.py → testing/print_signals.py
View file

30
tests/child_processes_test.py
View file

@ -8,8 +8,8 @@ from subprocess import Popen
import pytest
from tests.lib.testing import is_alive
from tests.lib.testing import pid_tree
from testing import is_alive
from testing import pid_tree
def spawn_and_kill_pipeline():
@ -34,7 +34,8 @@ def living_pids(pids):
return set(pid for pid in pids if is_alive(pid))
def test_setsid_signals_entire_group(both_debug_modes, setsid_enabled):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_enabled')
def test_setsid_signals_entire_group():
"""When dumb-init is running in setsid mode, it should only signal the
entire process group rooted at it.
"""
@ -42,10 +43,8 @@ def test_setsid_signals_entire_group(both_debug_modes, setsid_enabled):
assert len(living_pids(pids)) == 0
def test_no_setsid_doesnt_signal_entire_group(
both_debug_modes,
setsid_disabled,
):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_disabled')
def test_no_setsid_doesnt_signal_entire_group():
"""When dumb-init is not running in setsid mode, it should only signal its
immediate child.
"""
@ -73,7 +72,7 @@ def spawn_process_which_dies_with_children():
# we need to sleep before the shell exits, or dumb-init might send
# TERM to print_signals before it has had time to register custom
# signal handlers
'{python} -m tests.lib.print_signals & sleep 0.1'.format(
'{python} -m testing.print_signals & sleep 0.1'.format(
python=sys.executable,
),
),
@ -96,10 +95,8 @@ def spawn_process_which_dies_with_children():
return child_pid, proc.stdout
def test_all_processes_receive_term_on_exit_if_setsid(
both_debug_modes,
setsid_enabled,
):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_enabled')
def test_all_processes_receive_term_on_exit_if_setsid():
"""If the child exits for some reason, dumb-init should send TERM to all
processes in its session if setsid mode is enabled."""
child_pid, child_stdout = spawn_process_which_dies_with_children()
@ -110,10 +107,8 @@ def test_all_processes_receive_term_on_exit_if_setsid(
os.kill(child_pid, signal.SIGKILL)
def test_processes_dont_receive_term_on_exit_if_no_setsid(
both_debug_modes,
setsid_disabled,
):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_disabled')
def test_processes_dont_receive_term_on_exit_if_no_setsid():
"""If the child exits for some reason, dumb-init should not send TERM to
any other processes if setsid mode is disabled."""
child_pid, child_stdout = spawn_process_which_dies_with_children()
@ -133,7 +128,8 @@ def test_processes_dont_receive_term_on_exit_if_no_setsid(
('-c', '/doesnotexist'),
('--single-child', '--', '/doesnotexist'),
])
def test_fails_nonzero_with_bad_exec(args, both_debug_modes, both_setsid_modes):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_fails_nonzero_with_bad_exec(args):
"""If dumb-init can't exec as requested, it should exit nonzero."""
proc = Popen(('dumb-init',) + args, stderr=PIPE)
proc.wait()

9
tests/cli_test.py
View file

@ -11,7 +11,8 @@ def current_version():
return open('VERSION').read().strip()
def test_no_arguments_prints_usage(both_debug_modes, both_setsid_modes):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_no_arguments_prints_usage():
proc = Popen(('dumb-init'), stderr=PIPE)
_, stderr = proc.communicate()
assert proc.returncode != 0
@ -33,7 +34,8 @@ def test_exits_invalid_with_invalid_args():
@pytest.mark.parametrize('flag', ['-h', '--help'])
def test_help_message(flag, both_debug_modes, both_setsid_modes, current_version):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_help_message(flag, current_version):
"""dumb-init should say something useful when called with the help flag,
and exit zero.
"""
@ -63,7 +65,8 @@ def test_help_message(flag, both_debug_modes, both_setsid_modes, current_version
@pytest.mark.parametrize('flag', ['-V', '--version'])
def test_version_message(flag, both_debug_modes, both_setsid_modes, current_version):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_version_message(flag, current_version):
"""dumb-init should print its version when asked to."""
proc = Popen(('dumb-init', flag), stderr=PIPE)

6
tests/exit_status_test.py
View file

@ -5,7 +5,8 @@ import pytest
@pytest.mark.parametrize('exit_status', [0, 1, 2, 32, 64, 127, 254, 255])
def test_exit_status_regular_exit(exit_status, both_debug_modes, both_setsid_modes):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_exit_status_regular_exit(exit_status):
"""dumb-init should exit with the same exit status as the process that it
supervises when that process exits normally.
"""
@ -20,7 +21,8 @@ def test_exit_status_regular_exit(exit_status, both_debug_modes, both_setsid_mod
signal.SIGQUIT,
signal.SIGKILL,
])
def test_exit_status_terminated_by_signal(signal, both_debug_modes, both_setsid_modes):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_exit_status_terminated_by_signal(signal):
"""dumb-init should exit with status 128 + signal when the child process is
terminated by a signal.
"""

0
tests/lib/__init__.py
View file

38
tests/proxies_signals_test.py
View file

@ -1,42 +1,18 @@
import os
import re
import signal
import sys
from contextlib import contextmanager
from itertools import chain
from subprocess import PIPE
from subprocess import Popen
import pytest
from tests.lib.testing import NORMAL_SIGNALS
from tests.lib.testing import pid_tree
from tests.lib.testing import process_state
@contextmanager
def _print_signals(args=()):
"""Start print_signals and return dumb-init process."""
proc = Popen(
(
('dumb-init',) +
tuple(args) +
(sys.executable, '-m', 'tests.lib.print_signals')
),
stdout=PIPE,
)
assert re.match(b'^ready \(pid: (?:[0-9]+)\)\n$', proc.stdout.readline())
yield proc
for pid in pid_tree(proc.pid):
os.kill(pid, signal.SIGKILL)
from testing import NORMAL_SIGNALS
from testing import print_signals
from testing import process_state
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_proxies_signals():
"""Ensure dumb-init proxies regular signals to its child."""
with _print_signals() as proc:
with print_signals() as (proc, _):
for signum in NORMAL_SIGNALS:
proc.send_signal(signum)
assert proc.stdout.readline() == '{0}\n'.format(signum).encode('ascii')
@ -81,7 +57,7 @@ def _rewrite_map_to_args(rewrite_map):
@pytest.mark.usefixtures('both_debug_modes', 'both_setsid_modes')
def test_proxies_signals_with_rewrite(rewrite_map, sequence, expected):
"""Ensure dumb-init can rewrite signals."""
with _print_signals(_rewrite_map_to_args(rewrite_map)) as proc:
with print_signals(_rewrite_map_to_args(rewrite_map)) as (proc, _):
for send, expect_receive in zip(sequence, expected):
proc.send_signal(send)
assert proc.stdout.readline() == '{0}\n'.format(expect_receive).encode('ascii')
@ -97,7 +73,7 @@ def test_default_rewrites_can_be_overriden_with_setsid_enabled():
signal.SIGTTOU: signal.SIGINT,
signal.SIGTSTP: signal.SIGHUP,
}
with _print_signals(_rewrite_map_to_args(rewrite_map)) as proc:
with print_signals(_rewrite_map_to_args(rewrite_map)) as (proc, _):
for send, expect_receive in rewrite_map.items():
assert process_state(proc.pid) in ['running', 'sleeping']
proc.send_signal(send)
@ -119,7 +95,7 @@ def test_ignored_signals_are_not_proxied():
signal.SIGINT: 0,
signal.SIGWINCH: 0,
}
with _print_signals(_rewrite_map_to_args(rewrite_map)) as proc:
with print_signals(_rewrite_map_to_args(rewrite_map)) as (proc, _):
proc.send_signal(signal.SIGTERM)
proc.send_signal(signal.SIGINT)
assert proc.stdout.readline() == '{0}\n'.format(signal.SIGQUIT).encode('ascii')

109
tests/shell_background_test.py
View file

@ -1,83 +1,64 @@
import os
import re
import sys
import time
from signal import SIGCONT
from signal import SIGKILL
from subprocess import PIPE
from subprocess import Popen
from tests.lib.testing import pid_tree
from tests.lib.testing import process_state
from tests.lib.testing import SUSPEND_SIGNALS
import pytest
from testing import print_signals
from testing import process_state
from testing import SUSPEND_SIGNALS
def test_shell_background_support_setsid(both_debug_modes, setsid_enabled):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_enabled')
def test_shell_background_support_setsid():
"""In setsid mode, dumb-init should suspend itself and its children when it
receives SIGTSTP, SIGTTOU, or SIGTTIN.
"""
proc = Popen(
('dumb-init', sys.executable, '-m', 'tests.lib.print_signals'),
stdout=PIPE,
)
match = re.match(b'^ready \(pid: ([0-9]+)\)\n$', proc.stdout.readline())
pid = match.group(1).decode('ascii')
with print_signals() as (proc, pid):
for signum in SUSPEND_SIGNALS:
# both dumb-init and print_signals should be running or sleeping
assert process_state(pid) in ['running', 'sleeping']
assert process_state(proc.pid) in ['running', 'sleeping']
for signum in SUSPEND_SIGNALS:
# both dumb-init and print_signals should be running or sleeping
assert process_state(pid) in ['running', 'sleeping']
assert process_state(proc.pid) in ['running', 'sleeping']
# both should now suspend
proc.send_signal(signum)
# both should now suspend
proc.send_signal(signum)
for _ in range(1000):
time.sleep(0.001)
try:
assert process_state(proc.pid) == 'stopped'
assert process_state(pid) == 'stopped'
except AssertionError:
pass
for _ in range(1000):
time.sleep(0.001)
try:
assert process_state(proc.pid) == 'stopped'
assert process_state(pid) == 'stopped'
except AssertionError:
pass
else:
break
else:
break
else:
raise RuntimeError('Timed out waiting for processes to stop.')
raise RuntimeError('Timed out waiting for processes to stop.')
# and then both wake up again
proc.send_signal(SIGCONT)
assert (
proc.stdout.readline() == '{0}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(pid) in ['running', 'sleeping']
assert process_state(proc.pid) in ['running', 'sleeping']
for pid in pid_tree(proc.pid):
os.kill(pid, SIGKILL)
# and then both wake up again
proc.send_signal(SIGCONT)
assert (
proc.stdout.readline() == '{0}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(pid) in ['running', 'sleeping']
assert process_state(proc.pid) in ['running', 'sleeping']
def test_shell_background_support_without_setsid(both_debug_modes, setsid_disabled):
@pytest.mark.usefixtures('both_debug_modes', 'setsid_disabled')
def test_shell_background_support_without_setsid():
"""In non-setsid mode, dumb-init should forward the signals SIGTSTP,
SIGTTOU, and SIGTTIN, and then suspend itself.
"""
proc = Popen(
('dumb-init', sys.executable, '-m', 'tests.lib.print_signals'),
stdout=PIPE,
)
with print_signals() as (proc, _):
for signum in SUSPEND_SIGNALS:
assert process_state(proc.pid) in ['running', 'sleeping']
proc.send_signal(signum)
assert proc.stdout.readline() == '{0}\n'.format(signum).encode('ascii')
os.waitpid(proc.pid, os.WUNTRACED)
assert process_state(proc.pid) == 'stopped'
assert re.match(b'^ready \(pid: (?:[0-9]+)\)\n$', proc.stdout.readline())
for signum in SUSPEND_SIGNALS:
assert process_state(proc.pid) in ['running', 'sleeping']
proc.send_signal(signum)
assert proc.stdout.readline() == '{0}\n'.format(signum).encode('ascii')
os.waitpid(proc.pid, os.WUNTRACED)
assert process_state(proc.pid) == 'stopped'
proc.send_signal(SIGCONT)
assert (
proc.stdout.readline() == '{0}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(proc.pid) in ['running', 'sleeping']
for pid in pid_tree(proc.pid):
os.kill(pid, SIGKILL)
proc.send_signal(SIGCONT)
assert (
proc.stdout.readline() == '{0}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(proc.pid) in ['running', 'sleeping']

31
tests/tty_test.py
View file

@ -1,3 +1,10 @@
import os
import pty
import termios
import pytest
EOF = b'\x04'
@ -5,20 +12,18 @@ def ttyflags(fd):
"""normalize tty i/o for testing"""
# see:
# http://www.gnu.org/software/libc/manual/html_mono/libc.html#Output-Modes
import termios as T
attrs = T.tcgetattr(fd)
attrs[1] &= ~T.OPOST # don't munge output
attrs[3] &= ~T.ECHO # don't echo input
T.tcsetattr(fd, T.TCSANOW, attrs)
attrs = termios.tcgetattr(fd)
attrs[1] &= ~termios.OPOST # don't munge output
attrs[3] &= ~termios.ECHO # don't echo input
termios.tcsetattr(fd, termios.TCSANOW, attrs)
def readall(fd):
"""read until EOF"""
from os import read
result = b''
while True:
try:
chunk = read(fd, 1 << 10)
chunk = os.read(fd, 1 << 10)
except OSError as error:
if error.errno == 5: # linux pty EOF
return result
@ -33,23 +38,21 @@ def readall(fd):
def _test(fd):
"""write to tac via the pty and verify its output"""
ttyflags(fd)
from os import write
assert write(fd, b'1\n2\n3\n') == 6
assert write(fd, EOF * 2) == 2
assert os.write(fd, b'1\n2\n3\n') == 6
assert os.write(fd, EOF * 2) == 2
output = readall(fd)
assert output == b'3\n2\n1\n', repr(output)
print('PASS')
# disable debug output so it doesn't break our assertion
def test_tty(debug_disabled):
@pytest.mark.usefixtures('debug_disabled')
def test_tty():
"""
Ensure processes wrapped by dumb-init can write successfully, given a tty
"""
import pty
pid, fd = pty.fork()
if pid == 0:
from os import execvp
execvp('dumb-init', ('dumb-init', 'tac'))
os.execvp('dumb-init', ('dumb-init', 'tac'))
else:
_test(fd)