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

Fix flake8 failures

Browse source
This commit is contained in:
Chris Kuehl 2019-02-20 13:00:02 -08:00
parent 1722197e81
commit 9374f3e664
9 changed files with 39 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.pre-commit-config.yaml
View file

@ -24,3 +24,7 @@ repos:
rev: v1.1.5
hooks:
- id: remove-tabs
- repo: https://github.com/asottile/pyupgrade
rev: v1.11.3
hooks:
- id: pyupgrade

12
testing/__init__.py
View file

@ -20,7 +20,7 @@ SUSPEND_SIGNALS = frozenset([
NORMAL_SIGNALS = frozenset(
set(range(1, 32)) -
set([signal.SIGKILL, signal.SIGSTOP, signal.SIGCHLD]) -
{signal.SIGKILL, signal.SIGSTOP, signal.SIGCHLD} -
SUSPEND_SIGNALS
)
@ -37,7 +37,7 @@ def print_signals(args=()):
stdout=PIPE,
)
line = proc.stdout.readline()
m = re.match(b'^ready \(pid: ([0-9]+)\)\n$', line)
m = re.match(b'^ready \\(pid: ([0-9]+)\\)\n$', line)
assert m, line
yield proc, m.group(1).decode('ascii')
@ -52,7 +52,7 @@ def child_pids(pid):
for p in LocalPath('/proc').listdir():
try:
stat = open(p.join('stat').strpath).read()
m = re.match('^\d+ \(.+?\) [a-zA-Z] (\d+) ', stat)
m = re.match(r'^\d+ \(.+?\) [a-zA-Z] (\d+) ', stat)
assert m, stat
ppid = int(m.group(1))
if ppid == pid:
@ -67,11 +67,11 @@ def child_pids(pid):
def pid_tree(pid):
"""Return a list of all descendant PIDs for the given PID."""
children = child_pids(pid)
return set(
return {
pid
for child in children
for pid in pid_tree(child)
) | children
} | children
def is_alive(pid):
@ -82,7 +82,7 @@ def is_alive(pid):
def process_state(pid):
"""Return a process' state, such as "stopped" or "running"."""
status = LocalPath('/proc').join(str(pid), 'status').read()
m = re.search('^State:\s+[A-Z] \(([a-z]+)\)$', status, re.MULTILINE)
m = re.search(r'^State:\s+[A-Z] \(([a-z]+)\)$', status, re.MULTILINE)
return m.group(1)

6
testing/print_signals.py
View file

@ -13,7 +13,7 @@ import time
CATCHABLE_SIGNALS = frozenset(
set(range(1, 32)) - set([signal.SIGKILL, signal.SIGSTOP, signal.SIGCHLD])
set(range(1, 32)) - {signal.SIGKILL, signal.SIGSTOP, signal.SIGCHLD}
)
@ -22,7 +22,7 @@ last_signal = None
def unbuffered_print(line):
sys.stdout.write('{0}\n'.format(line))
sys.stdout.write('{}\n'.format(line))
sys.stdout.flush()
@ -34,7 +34,7 @@ if __name__ == '__main__':
for signum in CATCHABLE_SIGNALS:
signal.signal(signum, print_signal)
unbuffered_print('ready (pid: {0})'.format(os.getpid()))
unbuffered_print('ready (pid: {})'.format(os.getpid()))
# loop forever just printing signals
while True:

4
tests/child_processes_test.py
View file

@ -32,7 +32,7 @@ def spawn_and_kill_pipeline():
def living_pids(pids):
return set(pid for pid in pids if is_alive(pid))
return {pid for pid in pids if is_alive(pid)}
@pytest.mark.usefixtures('both_debug_modes', 'setsid_enabled')
@ -91,7 +91,7 @@ def spawn_process_which_dies_with_children():
# read a line from print_signals, figure out its pid
line = proc.stdout.readline()
match = re.match(b'ready \(pid: ([0-9]+)\)\n', line)
match = re.match(b'ready \\(pid: ([0-9]+)\\)\n', line)
assert match, line
child_pid = int(match.group(1))

22
tests/cli_test.py
View file

@ -85,15 +85,15 @@ def test_verbose(flag):
assert stdout == b'oh, hi\n'
# child/parent race to print output after the fork(), can't guarantee exact order
assert re.search(b'(^|\n)\[dumb-init\] setsid complete\.\n', stderr), stderr # child
assert re.search(b'(^|\n)\\[dumb-init\\] setsid complete\\.\n', stderr), stderr # child
assert re.search( # parent
(
'(^|\n)\[dumb-init\] Child spawned with PID [0-9]+\.\n'
'(^|\n)\\[dumb-init\\] Child spawned with PID [0-9]+\\.\n'
'.*' # child might print here
'\[dumb-init\] Received signal {signal.SIGCHLD}\.\n'
'\[dumb-init\] A child with PID [0-9]+ exited with exit status 0.\n'
'\[dumb-init\] Forwarded signal 15 to children\.\n'
'\[dumb-init\] Child exited with status 0\. Goodbye\.\n$'
'\\[dumb-init\\] Received signal {signal.SIGCHLD}\\.\n'
'\\[dumb-init\\] A child with PID [0-9]+ exited with exit status 0.\n'
'\\[dumb-init\\] Forwarded signal 15 to children\\.\n'
'\\[dumb-init\\] Child exited with status 0\\. Goodbye\\.\n$'
).format(signal=signal).encode('utf8'),
stderr,
re.DOTALL,
@ -110,11 +110,11 @@ def test_verbose_and_single_child(flag1, flag2):
assert stdout == b'oh, hi\n'
assert re.match(
(
'^\[dumb-init\] Child spawned with PID [0-9]+\.\n'
'\[dumb-init\] Received signal {signal.SIGCHLD}\.\n'
'\[dumb-init\] A child with PID [0-9]+ exited with exit status 0.\n'
'\[dumb-init\] Forwarded signal 15 to children\.\n'
'\[dumb-init\] Child exited with status 0\. Goodbye\.\n$'
'^\\[dumb-init\\] Child spawned with PID [0-9]+\\.\n'
'\\[dumb-init\\] Received signal {signal.SIGCHLD}\\.\n'
'\\[dumb-init\\] A child with PID [0-9]+ exited with exit status 0.\n'
'\\[dumb-init\\] Forwarded signal 15 to children\\.\n'
'\\[dumb-init\\] Child exited with status 0\\. Goodbye\\.\n$'
).format(signal=signal).encode('utf8'),
stderr,
)

4
tests/exit_status_test.py
View file

@ -11,7 +11,7 @@ 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.
"""
proc = Popen(('dumb-init', 'sh', '-c', 'exit {0}'.format(exit_status)))
proc = Popen(('dumb-init', 'sh', '-c', 'exit {}'.format(exit_status)))
proc.wait()
assert proc.returncode == exit_status
@ -29,7 +29,7 @@ def test_exit_status_terminated_by_signal(signal):
"""
# We use Python because sh is "dash" on Debian and "bash" on others.
# https://github.com/Yelp/dumb-init/issues/115
proc = Popen(('dumb-init', sys.executable, '-c', 'import os; os.kill(os.getpid(), {0})'.format(
proc = Popen(('dumb-init', sys.executable, '-c', 'import os; os.kill(os.getpid(), {})'.format(
signal,
)))
proc.wait()

14
tests/proxies_signals_test.py
View file

@ -15,12 +15,12 @@ def test_proxies_signals():
with print_signals() as (proc, _):
for signum in NORMAL_SIGNALS:
proc.send_signal(signum)
assert proc.stdout.readline() == '{0}\n'.format(signum).encode('ascii')
assert proc.stdout.readline() == '{}\n'.format(signum).encode('ascii')
def _rewrite_map_to_args(rewrite_map):
return chain.from_iterable(
('-r', '{0}:{1}'.format(src, dst)) for src, dst in rewrite_map.items()
('-r', '{}:{}'.format(src, dst)) for src, dst in rewrite_map.items()
)
@ -60,7 +60,7 @@ def test_proxies_signals_with_rewrite(rewrite_map, sequence, expected):
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')
assert proc.stdout.readline() == '{}\n'.format(expect_receive).encode('ascii')
@pytest.mark.usefixtures('both_debug_modes', 'setsid_enabled')
@ -78,12 +78,12 @@ def test_default_rewrites_can_be_overriden_with_setsid_enabled():
assert process_state(proc.pid) in ['running', 'sleeping']
proc.send_signal(send)
assert proc.stdout.readline() == '{0}\n'.format(expect_receive).encode('ascii')
assert proc.stdout.readline() == '{}\n'.format(expect_receive).encode('ascii')
os.waitpid(proc.pid, os.WUNTRACED)
assert process_state(proc.pid) == 'stopped'
proc.send_signal(signal.SIGCONT)
assert proc.stdout.readline() == '{0}\n'.format(signal.SIGCONT).encode('ascii')
assert proc.stdout.readline() == '{}\n'.format(signal.SIGCONT).encode('ascii')
assert process_state(proc.pid) in ['running', 'sleeping']
@ -98,8 +98,8 @@ def test_ignored_signals_are_not_proxied():
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')
assert proc.stdout.readline() == '{}\n'.format(signal.SIGQUIT).encode('ascii')
proc.send_signal(signal.SIGWINCH)
proc.send_signal(signal.SIGHUP)
assert proc.stdout.readline() == '{0}\n'.format(signal.SIGHUP).encode('ascii')
assert proc.stdout.readline() == '{}\n'.format(signal.SIGHUP).encode('ascii')

6
tests/shell_background_test.py
View file

@ -31,7 +31,7 @@ def test_shell_background_support_setsid():
# and then both wake up again
proc.send_signal(SIGCONT)
assert (
proc.stdout.readline() == '{0}\n'.format(SIGCONT).encode('ascii')
proc.stdout.readline() == '{}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(pid) in ['running', 'sleeping']
assert process_state(proc.pid) in ['running', 'sleeping']
@ -46,12 +46,12 @@ def test_shell_background_support_without_setsid():
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')
assert proc.stdout.readline() == '{}\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')
proc.stdout.readline() == '{}\n'.format(SIGCONT).encode('ascii')
)
assert process_state(proc.pid) in ['running', 'sleeping']

2
tests/tty_test.py
View file

@ -77,7 +77,7 @@ def test_child_gets_controlling_tty_if_we_had_one():
output = readall(sfd)
assert os.waitpid(pid, 0) == (pid, 0), output
m = re.search(b'flags are: \[\[([a-zA-Z]+)\]\]\n', output)
m = re.search(b'flags are: \\[\\[([a-zA-Z]+)\\]\\]\n', output)
assert m, output
# "m" is job control