# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Tests for the execution-control options (--time, -0/--zero-exit) and the
subprocess-execution primitives (run_cli and the interrupt machinery)."""
from __future__ import annotations
import _thread
import ast
import logging
import os
import re
import shlex
import signal
import subprocess
import sys
import threading
from pathlib import Path
from textwrap import dedent
from time import monotonic, sleep
import click
import cloup
import pytest
from extra_platforms import is_windows
from extra_platforms.pytest import skip_windows
from click_extra import (
BUILTIN_THEMES,
command,
context,
echo,
format_cli_prompt,
get_current_theme,
group,
highlight_bin_name,
pass_context,
run_cli,
strip_ansi,
timer_option,
zero_exit_option,
)
from click_extra.execution import (
_GROUP_LEADERS,
_KILL_SIGNAL,
_LIVE_PROCESSES,
_LIVE_PROCESSES_LOCK,
_TERMINATE_GRACE,
PROMPT,
_descendant_pids,
_die_by_sigint,
_escaped_process_groups,
_parse_proc_stat,
_posix_process_table,
install_interrupt_handler,
terminate_live_processes,
)
from click_extra.pytest import command_decorators
[docs]
def test_die_by_sigint_skips_a_stream_refusing_a_reentrant_flush(monkeypatch):
"""The process still dies when a stream is mid-write in the interrupted code.
The handler of a second Ctrl+C can run inside a write to `stderr`, whose buffer
then refuses the flush with a `RuntimeError`.
"""
class BusyStream:
def flush(self):
raise RuntimeError("reentrant call inside <_io.BufferedWriter>")
class Exited(Exception):
pass
def fake_exit(code):
raise Exited(code)
signals = []
monkeypatch.setattr("sys.stdout", BusyStream())
monkeypatch.setattr("sys.stderr", BusyStream())
monkeypatch.setattr("signal.signal", lambda *args: None)
monkeypatch.setattr("os.kill", lambda pid, signum: signals.append(signum))
monkeypatch.setattr("os._exit", fake_exit)
with pytest.raises(Exited):
_die_by_sigint()
assert signals == ([] if is_windows() else [signal.SIGINT])
NAP_CLI = """
import time
import click_extra
@click_extra.command
def nap():
print("ready", flush=True)
time.sleep(30)
nap()
"""
"""A CLI that announces it is running, then sleeps until interrupted."""
PARSE_NAP_CLI = """
import time
import click_extra
def soak(ctx, param, value):
print("ready", flush=True)
time.sleep(30)
@click_extra.command
@click_extra.option("--beans", expose_value=False, callback=soak)
def stew():
pass
stew()
"""
"""A CLI that sleeps while it parses its options, before its context is entered."""
[docs]
def interrupt_cli(
tmp_path: Path,
script: str,
presses: int = 1,
launcher: tuple[str, ...] = (),
) -> tuple[int, str]:
"""Run `script` in a child, press Ctrl+C `presses` times once it is ready.
`launcher` holds the interpreter arguments that go before the script, to run
it through another CLI.
A further press waits for the child to say what one does, so it lands once
the child handles it. A child that never says so is killed.
:return: the child's return code and its `stderr`.
"""
script_path = tmp_path / "cli.py"
script_path.write_text(script, encoding="UTF-8")
process = subprocess.Popen(
(sys.executable, *launcher, str(script_path)),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="UTF-8",
# Keeps a `-m` launcher on the installed package, not on a checkout the
# tests happen to run from.
cwd=tmp_path,
)
watchdog = threading.Timer(20, process.kill)
watchdog.start()
stderr = ""
try:
assert process.stdout is not None
assert process.stderr is not None
assert process.stdout.readline() == "ready\n"
process.send_signal(signal.SIGINT)
for _ in range(presses - 1):
for line in process.stderr:
stderr += line
if "Press Ctrl+C again" in line:
break
process.send_signal(signal.SIGINT)
process.wait()
stderr += process.stderr.read()
finally:
watchdog.cancel()
if process.poll() is None:
process.kill()
process.wait()
return process.returncode, stderr
[docs]
@skip_windows
@pytest.mark.parametrize(
"script",
(
pytest.param(NAP_CLI, id="in_the_command"),
pytest.param(PARSE_NAP_CLI, id="while_parsing"),
),
)
def test_ctrl_c_ends_the_process_by_sigint(tmp_path, script):
"""A real Ctrl+C prints Click's `Aborted!`, then ends the process by `SIGINT`,
which a calling shell needs to stop its own loop: a status of `1`, or even
`130`, tells the shell the program handled the interrupt itself."""
returncode, stderr = interrupt_cli(tmp_path, script)
assert returncode == -signal.SIGINT, stderr
assert "Aborted!" in stderr
PLAIN_NAP_CLI = """
import time
import click
@click.command()
def nap():
print("ready", flush=True)
time.sleep(30)
if __name__ == "__main__":
nap()
"""
"""A plain Click CLI that sleeps until interrupted, for the `wrap` command to run."""
[docs]
@skip_windows
def test_ctrl_c_in_a_wrapped_cli_ends_the_process_by_sigint(tmp_path):
"""A CLI run by `wrap` aborts under Click's own `main()`, and the wrapping
process still ends by `SIGINT`."""
returncode, stderr = interrupt_cli(
tmp_path, PLAIN_NAP_CLI, launcher=("-m", "click_extra", "wrap", "--")
)
assert returncode == -signal.SIGINT, stderr
assert "Aborted!" in stderr
SLOW_JOBS_CLI = """
import threading
import time
import click_extra
from click_extra.jobs import run_jobs
BOTH_IN = threading.Barrier(2)
def bake(tray):
# Reports once both trays are in, so the interrupt finds two running tasks.
if BOTH_IN.wait(timeout=10) == 0:
print("ready", flush=True)
time.sleep(30)
return tray
@click_extra.command
def oven():
list(run_jobs(bake, [0, 1], jobs=2))
oven()
"""
"""A CLI whose two parallel tasks outlast any reasonable wait."""
SLOW_THREAD_CLI = """
import threading
import time
import click_extra
@click_extra.command
def kettle():
threading.Thread(target=time.sleep, args=(30,)).start()
print("ready", flush=True)
time.sleep(30)
kettle()
"""
"""A CLI whose single background thread outlasts any reasonable wait."""
[docs]
@skip_windows
@pytest.mark.parametrize(
("script", "notice"),
(
pytest.param(
SLOW_JOBS_CLI, "Waiting for 2 running tasks to finish.", id="two_tasks"
),
pytest.param(
SLOW_THREAD_CLI, "Waiting for 1 running task to finish.", id="one_task"
),
),
)
def test_second_ctrl_c_quits_while_tasks_finish(tmp_path, script, notice):
"""After a Ctrl+C, the run says it waits for its running tasks, and what a
second Ctrl+C does: quit at once, instead of waiting for the tasks."""
returncode, stderr = interrupt_cli(tmp_path, script, presses=2)
assert returncode == -signal.SIGINT, stderr
assert "Aborted!" in stderr
assert notice in stderr
assert "Press Ctrl+C again to quit now." in stderr
assert "Traceback" not in stderr
LOCKED_PRESS = """
import signal
import click
from click_extra import execution
{setup}
with execution._LIVE_PROCESSES_LOCK:
signal.raise_signal(signal.SIGINT)
"""
"""A Ctrl+C that lands while the main thread holds the process registry lock."""
[docs]
@skip_windows
@pytest.mark.parametrize(
"setup",
(
pytest.param(
'execution.install_interrupt_handler(click.Context(click.Command("tea")))',
id="first_press",
),
pytest.param(
"signal.signal(signal.SIGINT, execution._second_interrupt)",
id="second_press",
),
),
)
def test_ctrl_c_while_the_registry_lock_is_held(tmp_path, setup):
"""A Ctrl+C ends the process even when the main thread holds the registry lock.
Both handlers stop the running subprocesses, which takes that lock. The main
thread takes it too, as when the grace period of an interrupted run ends,
and a plain lock would block the handler of a press landing in that moment
for good.
"""
script = tmp_path / "press.py"
script.write_text(LOCKED_PRESS.format(setup=setup), encoding="UTF-8")
try:
result = subprocess.run(
(sys.executable, str(script)),
capture_output=True,
text=True,
encoding="UTF-8",
timeout=20,
check=False,
)
except subprocess.TimeoutExpired:
pytest.fail("The Ctrl+C handler hung on the registry lock.")
assert result.returncode == -signal.SIGINT, result.stderr
STUBBORN_CHILDREN_CLI = """
import os
import sys
import threading
import time
import click_extra
from click_extra import execution
from click_extra.execution import run_cli
from click_extra.jobs import run_jobs
execution._TERMINATE_GRACE = 0.5
CHILD = (
"import os, signal, sys, time;"
"signal.signal(signal.SIGINT, signal.SIG_IGN);"
"signal.signal(signal.SIGTERM, signal.SIG_IGN);"
"open(sys.argv[1], 'w', encoding='UTF-8').write(str(os.getpid()));"
"time.sleep(30)"
)
PID_FILES = [os.path.join(FOLDER, f"child-{n}.pid") for n in range(2)]
def spawn(pid_file):
return run_cli((sys.executable, "-c", CHILD, pid_file))
def announce():
while not all(os.path.exists(path) for path in PID_FILES):
time.sleep(0.05)
print("ready", flush=True)
@click_extra.command
def stubborn():
threading.Thread(target=announce, daemon=True).start()
list(run_jobs(spawn, PID_FILES, jobs=2))
stubborn()
"""
"""A CLI fanning out two children that ignore both `SIGINT` and `SIGTERM`."""
[docs]
@skip_windows
def test_interrupted_run_kills_children_ignoring_sigterm(tmp_path):
"""One Ctrl+C is enough: the aborting run sends `SIGTERM` to the children of
`run_cli`, then `SIGKILL` to the ones still running after the grace."""
script = STUBBORN_CHILDREN_CLI.replace("FOLDER", repr(str(tmp_path)))
start = monotonic()
try:
returncode, stderr = interrupt_cli(tmp_path, script)
assert returncode == -signal.SIGINT, stderr
assert monotonic() - start < 10
pids = [
int(path.read_text(encoding="UTF-8"))
for path in tmp_path.glob("child-*.pid")
]
assert len(pids) == 2
for pid in pids:
deadline = monotonic() + 5
while monotonic() < deadline:
try:
os.kill(pid, 0)
except ProcessLookupError:
break
sleep(0.05)
else:
pytest.fail(f"child {pid} survived the interrupt")
finally:
for path in tmp_path.glob("child-*.pid"):
try:
os.kill(int(path.read_text(encoding="UTF-8")), signal.SIGKILL)
except (ProcessLookupError, ValueError):
pass
STUBBORN_GRANDCHILDREN_CLI = """
import os
import sys
import threading
import time
import click_extra
from click_extra import execution
from click_extra.execution import run_cli
from click_extra.jobs import run_jobs
execution._TERMINATE_GRACE = 0.5
GRANDCHILD = (
"import os, signal, sys, time;"
"signal.signal(signal.SIGINT, signal.SIG_IGN);"
"signal.signal(signal.SIGTERM, signal.SIG_IGN);"
"open(sys.argv[1] + '.tmp', 'w', encoding='UTF-8').write(str(os.getpid()));"
"os.replace(sys.argv[1] + '.tmp', sys.argv[1]);"
"time.sleep(30)"
)
CHILD = (
"import signal, subprocess, sys, time;"
"signal.signal(signal.SIGINT, signal.SIG_IGN);"
"signal.signal(signal.SIGTERM, signal.CHILD_ON_SIGTERM);"
"subprocess.Popen((sys.executable, '-c', sys.argv[2], sys.argv[1]));"
"time.sleep(30)"
)
PID_FILES = [os.path.join(FOLDER, f"grandchild-{n}.pid") for n in range(2)]
def spawn(pid_file):
return run_cli(
(sys.executable, "-c", CHILD, pid_file, GRANDCHILD),
start_new_session=NEW_SESSION,
)
def announce():
while not all(os.path.exists(path) for path in PID_FILES):
time.sleep(0.05)
print("ready", flush=True)
@click_extra.command
def stubborn():
threading.Thread(target=announce, daemon=True).start()
list(run_jobs(spawn, PID_FILES, jobs=2))
stubborn()
"""
"""A CLI fanning out two children, each with a grandchild that ignores both
`SIGINT` and `SIGTERM` and holds the output its child was given."""
[docs]
@skip_windows
@pytest.mark.parametrize("new_session", (False, True), ids=("same-group", "session"))
@pytest.mark.parametrize(
"child_on_sigterm",
# A child that outlives `SIGTERM` is killed with its grandchild. One that
# exits on it leaves the grandchild behind, with no parent to find it by.
("SIG_IGN", "SIG_DFL"),
ids=("stubborn-child", "child-exits-first"),
)
@pytest.mark.usefixtures("requires_process_table")
def test_interrupted_run_kills_grandchildren_ignoring_sigterm(
tmp_path, new_session, child_on_sigterm
):
"""The `SIGKILL` of an aborting run reaches what the children of `run_cli`
started, so no grandchild holds the run open or outlives it."""
script = (
STUBBORN_GRANDCHILDREN_CLI
.replace("FOLDER", repr(str(tmp_path)))
.replace("NEW_SESSION", repr(new_session))
.replace("CHILD_ON_SIGTERM", child_on_sigterm)
)
pid_files = [tmp_path / f"grandchild-{n}.pid" for n in range(2)]
start = monotonic()
try:
returncode, stderr = interrupt_cli(tmp_path, script)
assert returncode == -signal.SIGINT, stderr
assert monotonic() - start < 10
for pid_file in pid_files:
_assert_process_dies(int(pid_file.read_text(encoding="UTF-8")))
finally:
for pid_file in pid_files:
try:
os.kill(int(pid_file.read_text(encoding="UTF-8")), signal.SIGKILL)
except (OSError, ValueError):
pass
SELF_INTERRUPT_CLI = """
import signal
import click_extra
@click_extra.command
def nap():
signal.raise_signal(signal.SIGINT)
nap()
"""
"""A CLI that interrupts itself, as a Ctrl+C does, on every platform."""
SECOND_INTERRUPT_SCRIPT = """
import signal
from click_extra import execution
execution._second_interrupt(signal.SIGINT, None)
"""
"""A process handling the Ctrl+C pressed while an interrupted run waits."""
OUTLASTED_GRACE_SCRIPT = """
import threading
import time
from click_extra import execution
execution._TERMINATE_GRACE = 0.1
threading.Thread(target=time.sleep, args=(1,)).start()
execution._exit_interrupted()
"""
"""An interrupted run whose only task outlasts the grace given to children."""
[docs]
@pytest.mark.parametrize(
("script", "notice"),
(
pytest.param(SELF_INTERRUPT_CLI, "Aborted!", id="ctrl_c"),
# The handler of a second Ctrl+C exits without a word.
pytest.param(SECOND_INTERRUPT_SCRIPT, "", id="second_ctrl_c"),
pytest.param(
OUTLASTED_GRACE_SCRIPT,
"Waiting for 1 running task to finish.",
id="kill_after_the_grace",
),
),
)
def test_interrupted_run_exits_like_an_unhandled_ctrl_c(tmp_path, script, notice):
"""An interrupted run ends the way an unhandled Ctrl+C does, on Windows too.
The process dies by `SIGINT` on POSIX, and exits with the status Windows
reports for a Ctrl+C. The tests sending a real Ctrl+C to a child are skipped
on Windows, so each script here reaches one step of that exit on its own: the
whole path of a command, the second Ctrl+C, and the kill of the children still
running after the grace, for which Windows has no `SIGKILL`.
"""
result = subprocess.run(
(sys.executable, "-c", script),
capture_output=True,
text=True,
encoding="UTF-8",
cwd=tmp_path,
timeout=20,
check=False,
)
assert "Traceback" not in result.stderr
assert notice in result.stderr
# 0xC000013A is STATUS_CONTROL_C_EXIT, as a parent process reads it.
assert result.returncode == (0xC000013A if is_windows() else -signal.SIGINT)
PACKAGE_ROOT = Path(__file__).parent.parent / "click_extra"
PORTABLE_SIGNALS = frozenset({
"SIGABRT",
"SIGFPE",
"SIGILL",
"SIGINT",
"SIGSEGV",
"SIGTERM",
})
"""The signal numbers Python defines on every platform, Windows included."""
POSIX_BRANCHES = {"is_windows()": "orelse", "not is_windows()": "body"}
"""The branch only POSIX takes, for each platform check an `if` can test."""
[docs]
def unguarded_posix_signals(source: str) -> list[int]:
"""Lines of `source` that read a signal number Windows does not define.
`signal.SIGKILL` raises `AttributeError` on Windows, as every signal number
outside `PORTABLE_SIGNALS` does. So a module may read one only where Windows
never goes: in the `else` of an `is_windows()` check, or under a
`not is_windows()` one, of an `if` statement or a conditional expression.
"""
lines: list[int] = []
def scan(node: ast.AST, guarded: bool) -> None:
posix_branch = None
if isinstance(node, (ast.If, ast.IfExp)):
posix_branch = POSIX_BRANCHES.get(ast.unparse(node.test))
elif (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "signal"
and re.fullmatch(r"SIG[A-Z0-9]+", node.attr)
and node.attr not in PORTABLE_SIGNALS
and not guarded
):
lines.append(node.lineno)
for branch, children in ast.iter_fields(node):
for child in children if isinstance(children, list) else [children]:
if isinstance(child, ast.AST):
scan(child, guarded or branch == posix_branch)
scan(ast.parse(source), guarded=False)
return lines
[docs]
def test_posix_only_signal_scan_finds_the_package_and_an_unguarded_read():
"""The scan above reads the package, and tells a guarded read from a bare one."""
assert PACKAGE_ROOT / "execution.py" in PACKAGE_ROOT.rglob("*.py")
for unguarded in (
"os.kill(pid, signal.SIGKILL)\n",
# The branch Windows takes is no guard.
"if is_windows():\n os.kill(pid, signal.SIGKILL)\n",
"signal.SIGKILL if is_windows() else signal.SIGTERM\n",
):
assert unguarded_posix_signals(unguarded), unguarded
for guarded in (
"if is_windows():\n pass\nelse:\n os.kill(pid, signal.SIGKILL)\n",
"if not is_windows():\n os.kill(pid, signal.SIGKILL)\n",
"signal.SIGTERM if is_windows() else signal.SIGKILL\n",
# A portable signal number and a handler need no guard.
"signal.signal(signal.SIGINT, signal.SIG_DFL)\n",
):
assert not unguarded_posix_signals(guarded), guarded
[docs]
def test_simulated_interrupt_keeps_click_exit_status(invoke):
"""A `KeyboardInterrupt` raised by code, as a test runner does, keeps Click's
`Aborted!` and status `1`: only a real Ctrl+C ends the process by `SIGINT`."""
@command
def nap():
raise KeyboardInterrupt
result = invoke(nap)
assert result.exit_code == 1
assert "Aborted!" in result.stderr
[docs]
@pytest.mark.parametrize(
("abort", "notice"),
(
pytest.param("interrupt", "\nAborted!\n", id="interrupt"),
pytest.param("abort", "Aborted!\n", id="abort"),
),
)
def test_abort_notice_precedes_close_callbacks(invoke, abort, notice):
"""`Aborted!` prints as soon as the abort leaves the command: before the close
callbacks of its context and of its parents, where Click prints it after them."""
@group
@pass_context
def farm(ctx):
ctx.call_on_close(lambda: echo("Barn closed.", err=True))
@farm.command()
@pass_context
def harvest(ctx):
ctx.call_on_close(lambda: echo("Field closed.", err=True))
if abort == "interrupt":
raise KeyboardInterrupt
ctx.abort()
result = invoke(farm, "harvest")
assert result.exit_code == 1
assert result.stderr == f"{notice}Field closed.\nBarn closed.\n"
[docs]
def test_abort_caught_around_an_invoked_command_prints_no_notice(invoke):
"""A command run through `ctx.invoke()` leaves its caller free to catch the
abort: a run that goes on prints no `Aborted!`."""
@group
def farm():
pass
@farm.command()
@pass_context
def harvest(ctx):
ctx.abort()
@farm.command()
@pass_context
def visit(ctx):
try:
ctx.invoke(harvest)
except click.Abort:
echo("The visit goes on.")
result = invoke(farm, "visit")
assert result.exit_code == 0
assert result.stdout == "The visit goes on.\n"
assert not result.stderr
[docs]
def test_abort_without_standalone_mode_prints_no_notice(invoke):
"""Without standalone mode, Click raises the abort to the caller, which reports
it its own way: no `Aborted!` is printed."""
@command
@pass_context
def nap(ctx):
ctx.call_on_close(lambda: echo("Closed.", err=True))
raise KeyboardInterrupt
result = invoke(nap, standalone_mode=False)
assert isinstance(result.exception, click.Abort)
assert result.stderr == "Closed.\n\n"
# --- Timer ------------------------------------------------------------------
@group
def integrated_timer():
echo("Start of CLI")
@integrated_timer.command()
def fast_subcommand():
sleep(0.02)
echo("End of fast subcommand")
@integrated_timer.command()
def slow_subcommand():
sleep(0.2)
echo("End of slow subcommand")
[docs]
@pytest.mark.parametrize(
("subcommand_id", "time_min"),
(
("fast", 0.01),
("slow", 0.1),
),
)
def test_integrated_time_option(invoke, subcommand_id, time_min):
result = invoke(integrated_timer, "--time", f"{subcommand_id}-subcommand")
group = re.fullmatch(
rf"Start of CLI\nEnd of {subcommand_id} subcommand\n"
r"Execution time: (?P<time>[0-9.]+) seconds.\n",
result.stdout,
)
assert group
# Hard-code upper bound to avoid flakiness on slow platforms like macOS.
assert time_min < float(group.groupdict()["time"]) < 80
assert not result.stderr
assert result.exit_code == 0
[docs]
@pytest.mark.parametrize("subcommand_id", ("fast", "slow"))
def test_integrated_notime_option(invoke, subcommand_id):
result = invoke(integrated_timer, "--no-time", f"{subcommand_id}-subcommand")
assert result.stdout == f"Start of CLI\nEnd of {subcommand_id} subcommand\n"
assert not result.stderr
assert result.exit_code == 0
[docs]
@pytest.mark.parametrize(
"cmd_decorator",
# Skip click extra's commands, as timer option is already part of the default.
command_decorators(no_groups=True, no_extra=True),
)
@pytest.mark.parametrize("option_decorator", (timer_option, timer_option()))
def test_standalone_timer_option(
invoke, cmd_decorator, option_decorator, assert_output_regex
):
@cmd_decorator
@option_decorator
def standalone_timer():
echo("It works!")
result = invoke(standalone_timer, "--help")
assert result.stdout == dedent(
"""\
Usage: standalone-timer [OPTIONS]
Options:
--time / --no-time Measure and print elapsed execution time.
--help Show this message and exit.
""",
)
assert not result.stderr
assert result.exit_code == 0
result = invoke(standalone_timer, "--time")
assert_output_regex(
result.stdout,
r"It works!\nExecution time: [0-9.]+ seconds.\n",
)
assert not result.stderr
assert result.exit_code == 0
result = invoke(standalone_timer, "--no-time")
assert result.stdout == "It works!\n"
assert not result.stderr
assert result.exit_code == 0
[docs]
def test_nested_run_of_a_timed_command_keeps_the_outer_start(invoke):
"""A run started from within another run of the same command times itself,
and the outer run reports the time since its own start."""
@command
@click.option("--nested", is_flag=True)
def oven(nested):
if nested:
sleep(0.5)
oven.main(["--time"], standalone_mode=False)
result = invoke(oven, "--time", "--nested")
assert result.exit_code == 0
inner, outer = (
float(seconds)
for seconds in re.findall(r"Execution time: ([0-9.]+) seconds\.", result.stdout)
)
# The outer run started half a second before the inner one did.
assert outer - inner >= 0.45
[docs]
def test_time_with_short_circuit_sibling_still_prints(invoke):
"""``--time --version`` still emits a duration.
``--version`` is an eager option that calls ``ctx.exit()`` before the
user command body runs, but ``--time`` is intentionally measured even
on short-circuit paths so it can probe the cost of Click Extra's own
machinery (eager callbacks, config loading, option parsing).
"""
@command
def short_circuit_cli():
echo("body ran")
result = invoke(short_circuit_cli, "--time", "--version")
assert re.search(r"Execution time: [0-9.]+ seconds\.", result.output)
assert result.exit_code == 0
# --- Zero exit --------------------------------------------------------------
[docs]
@pytest.mark.parametrize(
"cmd_decorator",
(click.command, click.command(), cloup.command(), command),
)
@pytest.mark.parametrize("option_decorator", (zero_exit_option, zero_exit_option()))
def test_standalone_zero_exit_option(invoke, cmd_decorator, option_decorator):
@cmd_decorator
@option_decorator
@pass_context
def cli(ctx):
echo("It works!")
echo(f"Zero-exit value: {context.get(ctx, context.ZERO_EXIT)}")
result = invoke(cli, "--help", color=False)
assert "-0, --zero-exit" in result.stdout
assert "Always exit with a status code of 0" in result.stdout
assert not result.stderr
assert result.exit_code == 0
# Defaults to False.
result = invoke(cli)
assert result.stdout == "It works!\nZero-exit value: False\n"
assert not result.stderr
assert result.exit_code == 0
# The long form enables the flag.
result = invoke(cli, "--zero-exit")
assert result.stdout == "It works!\nZero-exit value: True\n"
assert not result.stderr
assert result.exit_code == 0
# The -0 short form enables the flag.
result = invoke(cli, "-0")
assert result.stdout == "It works!\nZero-exit value: True\n"
assert not result.stderr
assert result.exit_code == 0
[docs]
def test_zero_exit_auto_envvar(invoke):
@command
@zero_exit_option
@pass_context
def cli(ctx):
echo(f"Zero-exit value: {context.get(ctx, context.ZERO_EXIT)}")
result = invoke(cli, env={"CLI_ZERO_EXIT": "1"})
assert result.stdout == "Zero-exit value: True\n"
assert not result.stderr
assert result.exit_code == 0
# --- Subprocess execution -----------------------------------------------------
[docs]
def test_run_cli_returns_completed_process(caplog):
"""run_cli mirrors subprocess.run's result shape, with separate streams."""
code = "import sys; print('to out'); print('to err', file=sys.stderr)"
with caplog.at_level(logging.DEBUG):
result = run_cli((sys.executable, "-c", code))
assert isinstance(result, subprocess.CompletedProcess)
assert result.returncode == 0
assert result.stdout == "to out\n"
assert result.stderr == "to err\n"
[docs]
def test_run_cli_cwd_moves_the_child(tmp_path):
"""`cwd` runs the child elsewhere; omitted, it inherits the caller's."""
code = "import os; print(os.path.basename(os.getcwd()))"
result = run_cli((sys.executable, "-c", code), cwd=tmp_path)
assert result.stdout.strip() == tmp_path.name
# The caller's own directory is never changed by the call.
assert Path.cwd() != tmp_path
assert run_cli((sys.executable, "-c", code)).stdout.strip() == Path.cwd().name
[docs]
def test_run_cli_flattens_nested_args(caplog):
"""Nested iterables are flattened, None dropped, and elements stringified."""
with caplog.at_level(logging.DEBUG):
result = run_cli((sys.executable, None, ("-c", ("print('ok')",))))
assert result.args == (sys.executable, "-c", "print('ok')")
assert result.stdout == "ok\n"
[docs]
def test_run_cli_discloses_command_at_info(caplog):
"""The invocation is logged up front at INFO, with its forced env vars, and
the output stays out of the INFO records."""
with caplog.at_level(logging.INFO):
run_cli(
(sys.executable, "-c", "print('sesame')"),
extra_env={"MY_VAR": "value"},
)
prompts = [
record
for record in caplog.records
if strip_ansi(record.getMessage()).startswith(PROMPT)
]
assert len(prompts) == 1
assert prompts[0].levelno == logging.INFO
message = strip_ansi(prompts[0].getMessage())
assert "MY_VAR=value " in message
assert sys.executable in message
# The child's output is a DEBUG concern, absent at INFO. The prompt record is
# excluded: the command line itself carries the print('sesame') code.
assert not any(
"sesame" in strip_ansi(record.getMessage())
for record in caplog.records
if record not in prompts
)
[docs]
def test_run_cli_command_level_override(caplog):
"""A caller can lower the disclosure line to DEBUG for internal probes."""
with caplog.at_level(logging.DEBUG):
run_cli((sys.executable, "-c", "pass"), command_level=logging.DEBUG)
prompts = [
record
for record in caplog.records
if strip_ansi(record.getMessage()).startswith(PROMPT)
]
assert len(prompts) == 1
assert prompts[0].levelno == logging.DEBUG
[docs]
def test_run_cli_streams_output_at_debug_with_label(caplog):
"""Every output line is forwarded to the logger, tagged with the label.
The tag rides the record's ``label`` attribute, not the message text: the
default :class:`click_extra.logging.Formatter` renders it glued to the level
name (``debug:probe: line1``). The prompt line carries it too, so the
command and its output read as one call when several children interleave.
"""
code = "import sys; print('line1'); print('line2'); print('boom', file=sys.stderr)"
with caplog.at_level(logging.DEBUG):
run_cli((sys.executable, "-c", code), label="probe")
streamed = [
strip_ansi(record.getMessage())
for record in caplog.records
if getattr(record, "label", None) == "probe"
]
assert "line1" in streamed
assert "line2" in streamed
assert "boom" in streamed
labeled = [r for r in caplog.records if getattr(r, "label", None) == "probe"]
prompts = [r for r in labeled if strip_ansi(r.getMessage()).startswith(PROMPT)]
# One prompt line at the command level, every other tagged line at the output
# level.
assert [r.levelno for r in prompts] == [logging.INFO]
assert all(r.levelno == logging.DEBUG for r in labeled if r not in prompts)
[docs]
def test_highlight_bin_name():
"""Only the binary's own name is styled; its directory stays plain, whichever
separator convention the path uses."""
theme = get_current_theme()
styled = theme.invoked_command
assert highlight_bin_name("/opt/homebrew/bin/mas") == (
f"/opt/homebrew/bin/{styled('mas')}"
)
assert highlight_bin_name("C:\\Tools\\mas.exe") == f"C:\\Tools\\{styled('mas.exe')}"
# A bare name (no separator) is styled whole.
assert highlight_bin_name("mas") == styled("mas")
[docs]
def test_run_cli_merged_streams():
"""merge_stderr interleaves stderr into stdout and nulls the stderr field."""
code = dedent("""\
import sys
print("to out")
sys.stdout.flush()
print("to err", file=sys.stderr)
""")
result = run_cli((sys.executable, "-c", code), merge_stderr=True)
# run_cli() types both streams as str; merging nulls stderr at runtime.
stderr: str | None = result.stderr
assert stderr is None
assert "to out" in result.stdout
assert "to err" in result.stdout
[docs]
def test_run_cli_timeout_kills_child_and_attaches_partial_output():
"""An overrun raises TimeoutExpired carrying what was captured so far, and
leaves no zombie in the live registry."""
code = "print('partial', flush=True); import time; time.sleep(30)"
start = monotonic()
with pytest.raises(subprocess.TimeoutExpired) as excinfo:
run_cli((sys.executable, "-c", code), timeout=2)
# The child was killed at the deadline, not waited out.
assert monotonic() - start < 15
assert "partial" in (excinfo.value.output or "")
assert not _LIVE_PROCESSES
[docs]
def test_run_cli_handles_an_interrupt_tripped_as_the_wait_began():
"""A Ctrl+C the interpreter noted as the main thread blocked on the child is
handled within the poll period, not when the child exits.
`interrupt_main()` leaves the interrupt pending without waking the main
thread: the state a `SIGINT` leaves when it lands after the interpreter last
checked for one and before the thread blocked in the C library. The child
never receives it, like a child in a session of its own, or one with no
console on Windows.
"""
def trip():
deadline = monotonic() + 10
while not _LIVE_PROCESSES and monotonic() < deadline:
sleep(0.01)
# Leaves the main thread time to block on the child.
sleep(0.3)
_thread.interrupt_main()
tripper = threading.Thread(target=trip)
tripper.start()
start = monotonic()
try:
with pytest.raises(KeyboardInterrupt):
run_cli((sys.executable, "-c", "import time; time.sleep(30)"))
assert monotonic() - start < 10
finally:
tripper.join()
assert not _LIVE_PROCESSES
[docs]
@skip_windows
# _KILL_SIGNAL, not signal.SIGKILL: these values are read as the module is
# collected, and Windows, where the test skips, has no SIGKILL.
@pytest.mark.parametrize(
("code", "stop_signals"),
(
pytest.param(
"import time; time.sleep(30)",
(signal.SIGTERM, _KILL_SIGNAL),
id="exits-on-sigterm",
),
pytest.param(
"import signal, time;"
" signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(30)",
(_KILL_SIGNAL,),
id="ignores-sigterm",
),
),
)
def test_run_cli_kill_survives_a_second_interrupt(monkeypatch, code, stop_signals):
"""A Ctrl+C landing while the timeout reads the process table waits for the
signals to land, so the child never outlives the interrupt.
The timeout sends `SIGTERM` first, which a child may die of. A child ignoring
it dies of the `SIGKILL` that the interrupt brings forward.
"""
spawned: list[subprocess.Popen] = []
class RecordingPopen(subprocess.Popen):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
spawned.append(self)
def interrupted_table():
# The second Ctrl+C arrives between the table read and the signals.
signal.raise_signal(signal.SIGINT)
return _posix_process_table()
monkeypatch.setattr(subprocess, "Popen", RecordingPopen)
monkeypatch.setattr("click_extra.execution._posix_process_table", interrupted_table)
try:
with pytest.raises(KeyboardInterrupt):
# Long enough for the child to set its handler before the timeout.
run_cli((sys.executable, "-c", code), timeout=2)
# The child, spawned first, was stopped despite the interrupt: it ends
# at once, where a surviving one would sleep through the wait.
assert spawned
assert -spawned[0].wait(timeout=5) in stop_signals
finally:
for process in spawned:
if process.poll() is None:
process.kill()
process.wait()
[docs]
@skip_windows
def test_run_cli_default_shares_process_group():
"""By default the child stays in the caller's process group: it keeps the
controlling terminal (an interactive ``sudo`` raised from inside the child
must be able to prompt on ``/dev/tty``) and receives the terminal's signals
with the rest of the foreground group."""
result = run_cli((sys.executable, "-c", "import os; print(os.getpgid(0))"))
assert int(result.stdout) == os.getpgid(0)
[docs]
@skip_windows
def test_run_cli_new_session_makes_child_group_leader():
"""With start_new_session the child leads its own session and process group,
whose ID is its own PID: the property every group-kill path relies on."""
result = run_cli(
(sys.executable, "-c", "import os; print(os.getpid(), os.getpgid(0))"),
start_new_session=True,
)
pid, pgid = (int(field) for field in result.stdout.split())
assert pid == pgid
assert pgid != os.getpgid(0)
def _is_unreaped_zombie(pid: int) -> bool:
"""Return whether ``pid`` is a zombie: killed but not yet reaped.
A build sandbox with no init to reap orphans (like an Alpine ``abuild``
container) leaves a SIGKILL'd process lingering as a zombie, which
``os.kill(pid, 0)`` still reports as alive. Only Linux exposes the state
through ``/proc``; elsewhere a real reaper collects the process, so the
plain existence check is enough.
"""
try:
with open(f"/proc/{pid}/stat", encoding="ascii") as stat_file:
# The state letter follows the ")" that closes the comm field,
# which may itself contain spaces or parentheses.
return stat_file.read().rpartition(")")[2].split()[0] == "Z"
except OSError:
return False
def _assert_process_dies(pid: int, deadline_seconds: float = 5.0) -> None:
"""Poll until ``pid`` is gone, killing it and failing if it survives.
Signal delivery and the reparenting of orphans to the reaper are
asynchronous, so the check retries briefly instead of asserting at once. A
zombie counts as gone: it has already been killed, only its reaping is
pending (which never comes in a reaper-less sandbox).
"""
deadline = monotonic() + deadline_seconds
while monotonic() < deadline:
try:
os.kill(pid, 0)
except ProcessLookupError:
return
except PermissionError:
# Very unlikely PID reuse by another user: consider it gone.
return
if _is_unreaped_zombie(pid):
return
sleep(0.05)
os.kill(pid, signal.SIGKILL) # Don't leak the sleeper past the test.
pytest.fail(f"PID {pid} survived the process-group kill.")
GRANDCHILD_SESSIONS = pytest.mark.parametrize(
"grandchild_session",
(
pytest.param(False, id="same-group"),
# A grandchild leading a session of its own has left the child's process
# group, like the `make test` supervisor CPAN.pm forks and detaches with
# setsid().
pytest.param(True, id="own-session"),
),
)
"""Run a group-kill test on a grandchild inside the child's group and outside it."""
def _assert_timeout_reaps_grandchild(
grandchild_session: bool,
start_new_session: bool,
) -> None:
"""Time out a child that spawned a grandchild, then assert neither survives.
The grandchild inherited the stdout pipe, so a surviving orphan would have
stalled the drain for the full kill grace: a prompt return doubles as
evidence the whole tree died.
"""
code = dedent(f"""\
import subprocess, sys, time
grandchild = subprocess.Popen(
(sys.executable, "-c", "import time; time.sleep(30)"),
start_new_session={grandchild_session},
)
print(f"grandchild={{grandchild.pid}}", flush=True)
time.sleep(30)
""")
start = monotonic()
with pytest.raises(subprocess.TimeoutExpired) as excinfo:
run_cli(
(sys.executable, "-c", code),
timeout=2,
start_new_session=start_new_session,
)
assert monotonic() - start < 15
match = re.search(r"grandchild=(\d+)", excinfo.value.output or "")
assert match, "the child never reported its grandchild's PID"
_assert_process_dies(int(match.group(1)))
assert not _LIVE_PROCESSES
[docs]
@skip_windows
@GRANDCHILD_SESSIONS
@pytest.mark.usefixtures("requires_process_table")
def test_run_cli_timeout_new_session_kills_grandchildren(grandchild_session):
"""A timed-out start_new_session child takes its whole process tree down:
the grandchild is reaped along with it instead of surviving as an orphan
holding the inherited output pipe open."""
_assert_timeout_reaps_grandchild(grandchild_session, start_new_session=True)
[docs]
def test_run_cli_registers_live_process_then_discards_it():
"""run_cli tracks its subprocess while it runs, and drops it once done.
A background call parks in a real subprocess. Once it is registered,
terminate_live_processes() unblocks it, and run_cli's ``finally`` clears the
registry: this is the exact path the SIGINT handler drives on Ctrl+C.
"""
def call():
run_cli((sys.executable, "-c", "import time; time.sleep(30)"))
worker = threading.Thread(target=call)
worker.start()
try:
deadline = monotonic() + 5
while not _LIVE_PROCESSES and monotonic() < deadline:
sleep(0.01)
assert _LIVE_PROCESSES, "run_cli() should register its live subprocess"
# Terminating the child unblocks the parked run_cli() call.
terminate_live_processes()
worker.join(timeout=5)
assert not worker.is_alive()
finally:
terminate_live_processes()
worker.join(timeout=5)
# run_cli()'s finally discarded the child once it was reaped.
assert not _LIVE_PROCESSES
[docs]
@skip_windows
@GRANDCHILD_SESSIONS
@pytest.mark.usefixtures("requires_process_table")
def test_terminate_live_processes_signals_whole_group(tmp_path, grandchild_session):
"""Interrupting a start_new_session child reaps its grandchild too: the
group never received the terminal's SIGINT (it left the foreground group),
so terminate_live_processes() is its only kill path and must cover the
descendants.
The child reports its grandchild's PID through a sentinel file, polled
before tearing down, so the group is never signalled mid-spawn. The sentinel
is published by rename: a bare ``open(..., "w")`` creates the file before the
write is flushed, so polling for existence could tear the group down between
creation and flush, leaving an empty file behind for ``int()`` to choke on.
"""
pid_file = tmp_path / "grandchild.pid"
code = dedent(f"""\
import os, subprocess, sys, time
grandchild = subprocess.Popen(
(sys.executable, "-c", "import time; time.sleep(30)"),
start_new_session={grandchild_session},
)
pid_file = {str(pid_file)!r}
with open(pid_file + ".tmp", "w", encoding="UTF-8") as f:
f.write(str(grandchild.pid))
os.replace(pid_file + ".tmp", pid_file)
time.sleep(30)
""")
def call():
run_cli((sys.executable, "-c", code), start_new_session=True)
worker = threading.Thread(target=call)
worker.start()
try:
deadline = monotonic() + 10
while not pid_file.exists() and monotonic() < deadline:
sleep(0.01)
assert pid_file.exists(), "the child never reported its grandchild's PID"
assert _GROUP_LEADERS, "the isolated child should be flagged group leader"
terminate_live_processes()
worker.join(timeout=10)
assert not worker.is_alive()
finally:
terminate_live_processes()
worker.join(timeout=5)
assert not _LIVE_PROCESSES
assert not _GROUP_LEADERS
_assert_process_dies(int(pid_file.read_text(encoding="UTF-8")))
[docs]
@pytest.mark.usefixtures("requires_process_table")
def test_kill_signal_reaches_a_grandchild_holding_the_output(tmp_path):
"""The kill signal ends what a child started too, on every platform.
The grandchild holds the pipe `run_cli` reads, so the call returns only once
the grandchild is gone: a prompt return is the evidence.
"""
ready = tmp_path / "ready"
code = dedent(f"""\
import subprocess, sys, time
subprocess.Popen(
(sys.executable, "-c", "import time; time.sleep(30)"),
stdout=sys.stdout,
)
open({str(ready)!r}, "w", encoding="UTF-8").close()
time.sleep(30)
""")
def call():
run_cli((sys.executable, "-c", code), timeout=30)
# A daemon, so a grandchild that survives cannot hold the test session.
worker = threading.Thread(target=call, daemon=True)
worker.start()
try:
deadline = monotonic() + 10
while not ready.exists() and monotonic() < deadline:
sleep(0.01)
assert ready.exists(), "the child never started its grandchild"
terminate_live_processes(_KILL_SIGNAL)
worker.join(timeout=10)
assert not worker.is_alive()
finally:
terminate_live_processes(_KILL_SIGNAL)
worker.join(timeout=5)
assert not _LIVE_PROCESSES
[docs]
@pytest.mark.parametrize(
("stat", "expected"),
(
("4242 (python3) S 1 4242 4242 0 -1 4194304", (1, 4242)),
# The command name may hold spaces and parentheses of its own.
("4242 (a (b) c) R 17 99 99 0 -1", (17, 99)),
("4242 (python3)", None),
("not a stat line", None),
),
)
def test_parse_proc_stat(stat, expected):
assert _parse_proc_stat(stat) == expected
[docs]
@skip_windows
@pytest.mark.usefixtures("requires_process_table")
def test_posix_process_table_lists_this_process():
table = _posix_process_table()
assert table[os.getpid()] == (os.getppid(), os.getpgid(0))
[docs]
@skip_windows
@GRANDCHILD_SESSIONS
@pytest.mark.usefixtures("requires_process_table")
def test_run_cli_timeout_kills_grandchildren_without_a_session(grandchild_session):
"""A timed-out child sharing the caller's process group still takes its
grandchild down, killed one PID at a time since no group can be signalled
without hitting the test process too.
This is the shape of a shell script running commands of its own. An
escalated call has it too, but the command `sudo` runs as root is out of
reach of the per-PID kill: see `test_run_cli_timeout_sends_sigterm_first`.
"""
_assert_timeout_reaps_grandchild(grandchild_session, start_new_session=False)
[docs]
@skip_windows
def test_run_cli_timeout_sends_sigterm_first(tmp_path):
"""A timed-out child gets `SIGTERM` before any `SIGKILL`, so it can clean up.
This is what stops an escalated command: `sudo` forwards a `SIGTERM` to the
command it runs as root, which a `SIGKILL` from the caller cannot reach.
"""
marker = tmp_path / "terminated"
code = dedent("""\
import signal, sys, time
def stop(signum, frame):
open(sys.argv[1], "w").close()
sys.exit(0)
signal.signal(signal.SIGTERM, stop)
time.sleep(30)
""")
start = monotonic()
with pytest.raises(subprocess.TimeoutExpired):
run_cli((sys.executable, "-c", code, str(marker)), timeout=3)
assert marker.exists(), "the child got no SIGTERM before it died"
# It exited on the SIGTERM, before the grace ran out.
assert monotonic() - start < 3 + _TERMINATE_GRACE
assert not _LIVE_PROCESSES
[docs]
@skip_windows
def test_run_cli_timeout_kills_a_child_ignoring_sigterm():
"""A timed-out child ignoring `SIGTERM` is killed once the grace runs out."""
code = dedent("""\
import os, signal, time
signal.signal(signal.SIGTERM, signal.SIG_IGN)
print(f"pid={os.getpid()}", flush=True)
time.sleep(30)
""")
start = monotonic()
with pytest.raises(subprocess.TimeoutExpired) as excinfo:
run_cli((sys.executable, "-c", code), timeout=2)
elapsed = monotonic() - start
assert 2 + _TERMINATE_GRACE <= elapsed < 15
match = re.search(r"pid=(\d+)", excinfo.value.output or "")
assert match, "the child never reported its PID"
_assert_process_dies(int(match.group(1)))
assert not _LIVE_PROCESSES
[docs]
def test_descendant_pids_walks_the_whole_tree_deepest_first():
"""Every descendant is found whatever group it moved to, and a parent never
precedes its own child.
The IDs sit above the largest PID any platform allocates, so none of them
can be a process the test runner owns.
"""
leader = 10_000_000
table = {
leader: (1, leader),
leader + 1: (leader, leader),
# A grandchild that called setsid(), and its own child.
leader + 2: (leader, leader + 2),
leader + 3: (leader + 2, leader + 2),
# A great-grandchild of the member that stayed in the group.
leader + 4: (leader + 1, leader + 4),
# Not a descendant.
leader + 5: (1, leader + 5),
}
walked = _descendant_pids(leader, table)
assert set(walked) == {leader + 1, leader + 2, leader + 3, leader + 4}
assert walked.index(leader + 3) < walked.index(leader + 2)
assert walked.index(leader + 4) < walked.index(leader + 1)
[docs]
@skip_windows
def test_escaped_process_groups():
"""Only the groups of descendants outside the leader's group are returned.
The IDs sit above the largest PID any platform allocates, so none of them
can be the group of the running test process.
"""
leader = 10_000_000
table = {
leader: (1, leader),
leader + 1: (leader, leader),
# A grandchild that called setsid(), and its own child.
leader + 2: (leader, leader + 2),
leader + 3: (leader + 2, leader + 2),
# A great-grandchild that left the group below a member that stayed.
leader + 4: (leader + 1, leader + 4),
# Not a descendant.
leader + 5: (1, leader + 5),
# Groups killpg() must never be handed.
leader + 6: (leader, 1),
leader + 7: (leader, os.getpgid(0)),
}
assert _escaped_process_groups(leader, table) == {leader + 2, leader + 4}
[docs]
def test_terminate_live_processes_ignores_already_reaped():
"""A process gone between snapshot and signal is skipped, not raised on."""
proc = subprocess.Popen(
(sys.executable, "-c", "pass"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding="UTF-8",
)
proc.wait(timeout=5) # Already exited before we signal it.
with _LIVE_PROCESSES_LOCK:
_LIVE_PROCESSES.add(proc)
try:
terminate_live_processes() # Must not raise on the dead process.
finally:
with _LIVE_PROCESSES_LOCK:
_LIVE_PROCESSES.discard(proc)
[docs]
def test_install_interrupt_handler_terminates_children_and_reraises():
"""The installed SIGINT handler SIGTERMs live children, then raises to abort."""
ctx = click.Context(click.Command("cli"))
original = signal.getsignal(signal.SIGINT)
install_interrupt_handler(ctx)
handler = signal.getsignal(signal.SIGINT)
assert callable(handler)
assert handler is not original
proc = subprocess.Popen(
(sys.executable, "-c", "import time; time.sleep(30)"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding="UTF-8",
)
with _LIVE_PROCESSES_LOCK:
_LIVE_PROCESSES.add(proc)
try:
# Simulate signal delivery: the handler kills the child, then re-raises.
with pytest.raises(KeyboardInterrupt):
handler(signal.SIGINT, None)
assert proc.wait(timeout=5) != 0
finally:
with _LIVE_PROCESSES_LOCK:
_LIVE_PROCESSES.discard(proc)
if proc.poll() is None:
proc.kill()
ctx.close() # Restores the previous handler via call_on_close.
assert signal.getsignal(signal.SIGINT) is original
[docs]
def test_install_interrupt_handler_restored_on_context_close():
"""Closing the context restores the handler in place before the install."""
original = signal.getsignal(signal.SIGINT)
ctx = click.Context(click.Command("cli"))
try:
install_interrupt_handler(ctx)
assert signal.getsignal(signal.SIGINT) is not original
finally:
ctx.close()
assert signal.getsignal(signal.SIGINT) is original
[docs]
def test_install_interrupt_handler_skips_off_main_thread():
"""signal.signal() only works in the main thread: off-thread install is a no-op."""
original = signal.getsignal(signal.SIGINT)
ctx = click.Context(click.Command("cli"))
errors: list[BaseException] = []
def off_main():
try:
install_interrupt_handler(ctx)
except BaseException as exc: # noqa: BLE001
errors.append(exc)
worker = threading.Thread(target=off_main)
worker.start()
worker.join()
assert not errors
assert signal.getsignal(signal.SIGINT) is original