diff --git a/.gitignore b/.gitignore index 670257ea..289e11e9 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ __pycache__/ src/__pycache__/ *.egg-info/ .coverage +.coverage.* coverage.* .duct/ .idea diff --git a/pyproject.toml b/pyproject.toml index 16377161..a1303933 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -117,6 +117,7 @@ src_paths = ["src"] [tool.coverage.run] branch = true parallel = true +patch = ["subprocess", "execv"] [tool.coverage.paths] source = [ diff --git a/src/con_duct/_duct_main.py b/src/con_duct/_duct_main.py index e2954927..89d3d8f2 100644 --- a/src/con_duct/_duct_main.py +++ b/src/con_duct/_duct_main.py @@ -9,7 +9,7 @@ from typing import IO, TextIO from con_duct._models import LogPaths, Outputs, RecordTypes, SessionMode from con_duct._output import TailPipe, prepare_outputs, remove_files, safe_close_files -from con_duct._signals import SigIntHandler +from con_duct._signals import SigIntHandler, start_thread from con_duct._tracker import Report, monitor_process __version__ = version("con-duct") @@ -65,6 +65,11 @@ def execute( sample_interval, ) + # Install before any command exists so no SIGINT can kill duct and orphan + # the command; until attach() the handler only counts them. + sigint_handler = SigIntHandler() + signal.signal(signal.SIGINT, sigint_handler) + log_paths = LogPaths.create(output_prefix, pid=os.getpid()) try: log_paths.prepare_paths(clobber, capture_outputs) @@ -99,6 +104,12 @@ def execute( ) files_to_close.append(report.usage_file) + if sigint_handler.sigcount: + safe_close_files(files_to_close) + remove_files(log_paths, assert_empty=True) + lgr.info("Received SIGINT before starting %r, not starting it", command) + return 128 + signal.SIGINT + report.start_time = time.time() try: report.process = process = subprocess.Popen( @@ -118,7 +129,7 @@ def execute( lgr.error("%s: command not found", command) return 127 # seems what zsh and bash return then - signal.signal(signal.SIGINT, SigIntHandler(process.pid)) + sigint_handler.attach(process.pid) lgr.info("duct %s is executing %r...", __version__, full_command) lgr.info("Log files will be written to %s", log_paths.prefix) try: @@ -145,15 +156,15 @@ def execute( monitoring_thread = threading.Thread( target=monitor_process, args=monitoring_args ) - monitoring_thread.start() + start_thread(monitoring_thread) else: monitoring_thread = None if record_types.has_system_summary(): env_thread = threading.Thread(target=report.collect_environment) - env_thread.start() + start_thread(env_thread) sys_info_thread = threading.Thread(target=report.get_system_info) - sys_info_thread.start() + start_thread(sys_info_thread) else: env_thread, sys_info_thread = None, None diff --git a/src/con_duct/_output.py b/src/con_duct/_output.py index d82bd658..ed90b30f 100644 --- a/src/con_duct/_output.py +++ b/src/con_duct/_output.py @@ -9,6 +9,7 @@ import time from typing import IO, Any, TextIO from con_duct._models import LogPaths, Outputs +from con_duct._signals import start_thread class TailPipe: @@ -27,7 +28,7 @@ def start(self) -> None: self.stop_event = threading.Event() self.infile = open(self.file_path, "rb") self.thread = threading.Thread(target=self._tail, daemon=True) - self.thread.start() + start_thread(self.thread) def fileno(self) -> int: assert self.infile is not None diff --git a/src/con_duct/_signals.py b/src/con_duct/_signals.py index 48d66c51..2a7ccbc8 100644 --- a/src/con_duct/_signals.py +++ b/src/con_duct/_signals.py @@ -1,41 +1,138 @@ """Signal handlers for con-duct.""" from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager import logging import os import signal +import threading from types import FrameType from typing import Optional lgr = logging.getLogger("con-duct") +@contextmanager +def _sigint_blocked() -> Iterator[None]: + """Block SIGINT in the calling thread, restoring its mask on exit. + + A SIGINT that arrives meanwhile stays pending and is handled once the + mask is restored. + """ + old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT}) + try: + yield + finally: + signal.pthread_sigmask(signal.SIG_SETMASK, old_mask) + + +def start_thread(thread: threading.Thread) -> None: + """Start a thread that the kernel will never hand a SIGINT to. + + A signal sent to the process is delivered to any one thread that does + not block it, but Python runs the handler only in the main thread. If + another thread takes a SIGINT, the main thread stays asleep in its wait + for the command and the Ctrl-C is not forwarded until the command exits. + A new thread inherits its creator's signal mask, so block SIGINT just + while starting it. Every thread duct starts must be started this way. + + Parameters + ---------- + thread : threading.Thread + The thread to start + """ + with _sigint_blocked(): + thread.start() + + class SigIntHandler: """ Handler of SIGINT signals received by the process running duct. """ - def __init__(self, pid: int) -> None: - """ + def __init__(self) -> None: + # Installed before the command starts; SIGINTs are only counted + # until attach() says which process to forward them to. + self.pid: Optional[int] = None + self.sigcount: int = 0 + + def attach(self, pid: int) -> None: + """Start forwarding SIGINTs to the command. + Parameters ---------- pid : int - The PID of the process monitored by duct + The PID of the process monitored by duct. A SIGINT that arrived + while it was being started is acted on now. """ - self.pid: int = pid - self.sigcount: int = 0 + # A SIGINT after publishing the PID must not be forwarded both by + # __call__ and by the pending-signal check below. + with _sigint_blocked(): + self.pid = pid + if self.sigcount: + self._act() def __call__(self, _sig: int, _frame: Optional[FrameType]) -> None: self.sigcount += 1 + if self.pid is not None: + self._act() + + def _act(self) -> None: + try: + self._forward() + except ProcessLookupError: + # The command has already exited. Python runs this handler in + # the main thread at its next bytecode, which can be only after + # the wait for the command returns. + _log( + logging.WARNING, + "Received SIGINT, but the command has already exited", + ) + + def _forward(self) -> None: + assert self.pid is not None + # Act before logging: the handler can interrupt duct mid-write to + # stderr, and a failed log write must not stop the signal. if self.sigcount == 1: - lgr.info("Received SIGINT, passing to command") os.kill(self.pid, signal.SIGINT) + _log(logging.INFO, "Received SIGINT, passing to command") elif self.sigcount == 2: - lgr.info("Received second SIGINT, again passing to command") os.kill(self.pid, signal.SIGINT) + _log(logging.INFO, "Received second SIGINT, again passing to command") elif self.sigcount == 3: - lgr.warning("Received third SIGINT, forcefully killing command process") os.kill(self.pid, signal.SIGKILL) + _log( + logging.WARNING, + "Received third SIGINT, forcefully killing command process", + ) elif self.sigcount >= 4: - lgr.critical("Exiting duct, skipping cleanup") + _log(logging.CRITICAL, "Exiting duct, skipping cleanup") os._exit(1) + + +def _log(level: int, msg: str) -> None: + """Log from a signal handler without ever raising. + + If the signal arrived while duct was writing to stderr, logging raises + ``RuntimeError: reentrant call`` (even its error report does): the + buffered writer is held by the write the handler interrupted, which + cannot finish until the handler returns. Write the message straight to + file descriptor 2 instead, if the log level lets the message through. + It does not go through the log format, and it may land in the middle of + the log line that was being written, so it starts on a new line and says + why it looks different. + """ + try: + lgr.log(level, msg) + except RuntimeError: + if not lgr.isEnabledFor(level): + return + try: + line = ( + f"\ncon-duct [{logging.getLevelName(level)}] (written directly: " + f"the signal interrupted a log write): {msg}\n" + ) + os.write(2, line.encode()) + except OSError: + pass diff --git a/test/data/signal_ignorer.py b/test/data/signal_ignorer.py index 0f6fe066..3d1191fa 100755 --- a/test/data/signal_ignorer.py +++ b/test/data/signal_ignorer.py @@ -1,12 +1,25 @@ #!/usr/bin/env python3 +"""Ignore SIGINT, reporting progress through the file given as argv[1]. + +Writes "ready" once the handler is installed, then one "SIGINT" line per +SIGINT received, so a test can wait for each step instead of sleeping. +""" + import signal +import sys import time from types import FrameType from typing import Optional +def report(line: str) -> None: + with open(sys.argv[1], "a") as f: + f.write(line + "\n") + + def handle_signal(sig: int, _frame: Optional[FrameType]) -> None: print(f"Received {sig}") + report("SIGINT") if __name__ == "__main__": @@ -14,6 +27,7 @@ def handle_signal(sig: int, _frame: Optional[FrameType]) -> None: signal.siginterrupt( signal.SIGINT, False ) # Restart interrupted system calls so we can test multiple SIGINTS + report("ready") t0 = time.time() - while time.time() - t0 < 10: + while time.time() - t0 < 60: time.sleep(0.01) diff --git a/test/duct_main/test_e2e.py b/test/duct_main/test_e2e.py index 62aefc39..8e27a311 100644 --- a/test/duct_main/test_e2e.py +++ b/test/duct_main/test_e2e.py @@ -3,13 +3,16 @@ import os from pathlib import Path import platform +import signal import subprocess import time import pytest +from utils import assert_files, rest_of_stderr, start_duct from con_duct._constants import SUFFIXES SYSTEM = platform.system() TEST_SCRIPT_DIR = Path(__file__).parent.parent / "data" +LOG_FILES = [SUFFIXES[k] for k in ("stdout", "stderr", "info", "usage")] # Allow overriding the duct executable for testing external builds (e.g., PyInstaller) _DUCT_EXECUTABLES = [ exe.strip() @@ -99,17 +102,15 @@ def test_session_modes(temp_output_dir: str, duct_cmd: str, session_mode: str) - def test_session_mode_behavior_difference(temp_output_dir: str, duct_cmd: str) -> None: """Test that new-session and current-session modes behave differently.""" - # Start a unique background process in the current session + # Start a unique background process in the current session. It must outlive + # both duct runs however slow they are (the finally block stops it). background_process = subprocess.Popen( - ["python", "-c", "print('DUCT_TEST_MARKER'); import time; time.sleep(10)"], + ["python", "-c", "print('DUCT_TEST_MARKER'); import time; time.sleep(600)"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: - # Give background process time to start - time.sleep(0.1) - new_session_prefix = f"{temp_output_dir}new_" current_session_prefix = f"{temp_output_dir}current_" @@ -212,3 +213,66 @@ def test_logging_levels(temp_output_dir: str, duct_cmd: str) -> None: assert ( result_none.stderr == "" ), f"Expected empty stderr, got: {result_none.stderr!r}" + + +# The --fail-time values whose effect does not depend on how long the run +# took: 0 always keeps a failed command's logs, a negative value never does +FAIL_TIMES = [0, -1] + + +# @pytest.mark.flaky(reruns=5) # disabled: start_duct waits for duct instead of sleeping +@pytest.mark.parametrize("fail_time", FAIL_TIMES) +def test_signal_int(temp_output_dir: str, duct_cmd: str, fail_time: int) -> None: + args = ["-p", temp_output_dir, f"--fail-time={fail_time}"] + proc = start_duct(duct_cmd, [*args, "sleep", "60"]) + os.kill(proc.pid, signal.SIGINT) + proc.communicate() + + # duct forwards SIGINT to the command and exits with the command's code + assert proc.returncode == 128 + signal.SIGINT + + if fail_time < 0: + assert_files(temp_output_dir, LOG_FILES, exists=False) + else: + with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info: + info_data = json.loads(info.read()) + assert info_data["execution_summary"]["exit_code"] == 128 + signal.SIGINT + + +def _wait_for_lines(path: Path, n: int, proc: subprocess.Popen[str]) -> None: + """Wait until *path* has *n* lines, failing if duct exits first.""" + while not path.exists() or len(path.read_text().splitlines()) < n: + assert proc.poll() is None, ( + f"duct exited with {proc.returncode} before {path.name} had {n} lines; " + f"its stderr:\n{rest_of_stderr(proc)}" + ) + time.sleep(0.01) + + +# @pytest.mark.flaky(reruns=5) # disabled: the test waits for each step instead of sleeping +@pytest.mark.parametrize("fail_time", FAIL_TIMES) +def test_signal_kill( + temp_output_dir: str, tmp_path: Path, duct_cmd: str, fail_time: int +) -> None: + progress = tmp_path / "signal_ignorer.progress" + args = ["-p", temp_output_dir, f"--fail-time={fail_time}"] + script = TEST_SCRIPT_DIR / "signal_ignorer.py" + proc = start_duct(duct_cmd, [*args, str(script), str(progress)]) + + _wait_for_lines(progress, 1, proc) # "ready": the command ignores SIGINT now + # duct forwards the first two; each must arrive before the next is sent, + # since pending SIGINTs merge into one + for received in (2, 3): + os.kill(proc.pid, signal.SIGINT) + _wait_for_lines(progress, received, proc) + os.kill(proc.pid, signal.SIGINT) # third: duct kills the command + proc.communicate() + + assert proc.returncode == 128 + signal.SIGKILL + + if fail_time < 0: + assert_files(temp_output_dir, LOG_FILES, exists=False) + else: + with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info: + info_data = json.loads(info.read()) + assert info_data["execution_summary"]["exit_code"] == 128 + signal.SIGKILL diff --git a/test/duct_main/test_execution.py b/test/duct_main/test_execution.py index 72745c97..47e5a734 100644 --- a/test/duct_main/test_execution.py +++ b/test/duct_main/test_execution.py @@ -1,13 +1,11 @@ from __future__ import annotations import json import logging -import multiprocessing import os from pathlib import Path -import signal import subprocess import sys -from time import sleep, time +from time import time import pytest from utils import assert_files, run_duct_command from con_duct import _duct_main @@ -122,6 +120,28 @@ def test_sanity_red( assert_expected_files(temp_output_dir) +@pytest.mark.parametrize( + "fail_time,command,logs_kept", + [ + # Failed sooner than the threshold: logs are removed + (3, "exit 1", False), # 3 is the default --fail-time + (60, "exit 1", False), + # Failed after running longer than the threshold: logs are kept + (1, "sleep 1.2; exit 1", True), + ], +) +def test_fail_time_threshold( + temp_output_dir: str, fail_time: int, command: str, logs_kept: bool +) -> None: + assert ( + run_duct_command( + ["sh", "-c", command], output_prefix=temp_output_dir, fail_time=fail_time + ) + == 1 + ) + assert_expected_files(temp_output_dir, exists=logs_kept) + + def test_outputs_full(temp_output_dir: str) -> None: script_path = str(TEST_SCRIPT_DIR / "test_script.py") assert ( @@ -228,91 +248,6 @@ def test_execute_unknown_command( assert_expected_files(temp_output_dir, exists=False) -def _runner_for_signal_int(temp_output_dir: str, fail_time: float | None) -> int: - kws = {} - if fail_time is not None: - kws["fail_time"] = fail_time - return run_duct_command( - ["sleep", "60.74016230000801"], output_prefix=temp_output_dir, **kws - ) - - -@pytest.mark.flaky(reruns=5) -@pytest.mark.parametrize("fail_time", [None, 0, 10, -1, -3.14]) -def test_signal_int( - request: pytest.FixtureRequest, temp_output_dir: str, fail_time: float | None -) -> None: - # Scale wait time on retries to handle slow CI runners (PyPy, Mac) - attempt = getattr(request.node, "execution_count", 1) - wait_time = 0.2 * attempt - proc = multiprocessing.Process( - target=_runner_for_signal_int, args=(temp_output_dir, fail_time) - ) - proc.start() - sleep(wait_time) - assert proc.pid is not None, "Process PID should not be None" # for mypy - os.kill(proc.pid, signal.SIGINT) - proc.join() - - # Once the command has been killed, duct should exit gracefully with exit code 0 - assert proc.exitcode == 0 - - if fail_time is None or fail_time != 0: - assert_expected_files(temp_output_dir, exists=False) - else: - # proc exit code should Cannot retrieve the exit code from the thread, it is written to the file - with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info: - info_data = json.loads(info.read()) - - command_exit_code = info_data["execution_summary"]["exit_code"] - # SIGINT - assert command_exit_code == 128 + 2 - - -def _runner_for_signal_kill(temp_output_dir: str, fail_time: float | None) -> int: - script_path = str(TEST_SCRIPT_DIR / "signal_ignorer.py") - kws = {} - if fail_time is not None: - kws["fail_time"] = fail_time - return run_duct_command([script_path], output_prefix=temp_output_dir, **kws) - - -@pytest.mark.flaky(reruns=5) -@pytest.mark.parametrize("fail_time", [None, 0, 10, -1, -3.14]) -def test_signal_kill( - request: pytest.FixtureRequest, temp_output_dir: str, fail_time: float | None -) -> None: - # Scale wait time on retries to handle slow CI runners (PyPy, Mac) - attempt = getattr(request.node, "execution_count", 1) - wait_time = 0.2 * attempt - proc = multiprocessing.Process( - target=_runner_for_signal_kill, args=(temp_output_dir, fail_time) - ) - proc.start() - sleep(wait_time) - assert proc.pid is not None, "Process PID should not be None" # for mypy - os.kill(proc.pid, signal.SIGINT) - sleep(wait_time) - os.kill(proc.pid, signal.SIGINT) - sleep(wait_time) - os.kill(proc.pid, signal.SIGINT) - proc.join() - - # Once the command has been killed, duct should exit gracefully with exit code 0 - assert proc.exitcode == 0 - - if fail_time is None or fail_time != 0: - assert_expected_files(temp_output_dir, exists=False) - else: - # Cannot retrieve the command exit code from the thread, get from duct log - with open(os.path.join(temp_output_dir, SUFFIXES["info"])) as info: - info_data = json.loads(info.read()) - - command_exit_code = info_data["execution_summary"]["exit_code"] - # SIGKILL - assert command_exit_code == 128 + 9 - - def test_duct_as_executable(temp_output_dir: str) -> None: ps_command = [ sys.executable, diff --git a/test/duct_main/test_signals.py b/test/duct_main/test_signals.py new file mode 100644 index 00000000..bdc84a11 --- /dev/null +++ b/test/duct_main/test_signals.py @@ -0,0 +1,209 @@ +from __future__ import annotations +from collections.abc import Generator +import logging +import os +from pathlib import Path +import signal +import threading +from typing import Any +import pytest +from utils import run_duct_command +from con_duct import _signals +from con_duct._signals import SigIntHandler, start_thread +from con_duct._tracker import Report + + +@pytest.fixture +def sent(monkeypatch: pytest.MonkeyPatch) -> list[tuple[int, int]]: + """Record os.kill calls made by the handler instead of sending them.""" + calls: list[tuple[int, int]] = [] + monkeypatch.setattr( + "con_duct._signals.os.kill", lambda pid, sig: calls.append((pid, sig)) + ) + return calls + + +@pytest.fixture +def restore_sigint() -> Generator[None, None, None]: + """execute() installs its own SIGINT handler; put pytest's back.""" + orig = signal.getsignal(signal.SIGINT) + yield + signal.signal(signal.SIGINT, orig) + + +@pytest.fixture +def stderr_locked(monkeypatch: pytest.MonkeyPatch) -> None: + """Make logging fail as it does when the handler interrupts a stderr write.""" + + def reentrant(*_args: Any) -> None: + raise RuntimeError("reentrant call inside <_io.BufferedWriter name=''>") + + monkeypatch.setattr(_signals.lgr, "log", reentrant) + + +@pytest.mark.parametrize( + "presses,expected", + [ + (1, [signal.SIGINT]), + (2, [signal.SIGINT, signal.SIGINT]), + (3, [signal.SIGINT, signal.SIGINT, signal.SIGKILL]), + ], +) +def test_forwards_after_attach( + sent: list[tuple[int, int]], presses: int, expected: list[int] +) -> None: + handler = SigIntHandler() + handler.attach(1234) + for _ in range(presses): + handler(signal.SIGINT, None) + assert sent == [(1234, sig) for sig in expected] + + +def test_counts_but_does_not_forward_before_attach( + sent: list[tuple[int, int]], +) -> None: + handler = SigIntHandler() + handler(signal.SIGINT, None) + assert handler.sigcount == 1 + assert sent == [] + + +@pytest.mark.parametrize( + "presses,expected", [(1, signal.SIGINT), (2, signal.SIGINT), (3, signal.SIGKILL)] +) +def test_attach_acts_on_earlier_sigints( + sent: list[tuple[int, int]], presses: int, expected: int +) -> None: + handler = SigIntHandler() + for _ in range(presses): + handler(signal.SIGINT, None) + handler.attach(1234) + # Only the latest escalation level is applied, once + assert sent == [(1234, expected)] + + +@pytest.mark.usefixtures("restore_sigint") +def test_sigint_during_attach_is_forwarded_once(sent: list[tuple[int, int]]) -> None: + class InterruptOnAttach(SigIntHandler): + def __setattr__(self, name: str, value: Any) -> None: + super().__setattr__(name, value) + if name == "pid" and value is not None: + # Deliver a real SIGINT immediately after publishing the PID. + signal.raise_signal(signal.SIGINT) + + handler = InterruptOnAttach() + signal.signal(signal.SIGINT, handler) + handler.attach(1234) + + assert handler.sigcount == 1 + assert sent == [(1234, signal.SIGINT)] + + +@pytest.mark.usefixtures("stderr_locked") +def test_forwards_even_if_logging_fails(sent: list[tuple[int, int]]) -> None: + handler = SigIntHandler() + handler.attach(1234) + handler(signal.SIGINT, None) + assert sent == [(1234, signal.SIGINT)] + + +@pytest.mark.usefixtures("stderr_locked") +def test_log_falls_back_to_raw_stderr_write( + capfd: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture +) -> None: + caplog.set_level(logging.INFO, logger="con-duct") + _signals._log(logging.INFO, "Received SIGINT, passing to command") + assert capfd.readouterr().err == ( + "\ncon-duct [INFO] (written directly: the signal interrupted a log write): " + "Received SIGINT, passing to command\n" + ) + + +@pytest.mark.usefixtures("stderr_locked") +def test_log_fallback_respects_log_level( + capfd: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture +) -> None: + caplog.set_level(logging.WARNING, logger="con-duct") + _signals._log(logging.INFO, "Received SIGINT, passing to command") + assert capfd.readouterr().err == "" + + +@pytest.mark.usefixtures("stderr_locked") +def test_log_fallback_never_raises( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + def closed(_fd: int, _data: bytes) -> int: + raise OSError(9, "Bad file descriptor") + + caplog.set_level(logging.INFO, logger="con-duct") + monkeypatch.setattr("con_duct._signals.os.write", closed) + _signals._log(logging.INFO, "Received SIGINT, passing to command") # must not raise + + +def test_command_already_exited( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + def no_such_process(_pid: int, _sig: int) -> None: + raise ProcessLookupError(3, "No such process") + + monkeypatch.setattr("con_duct._signals.os.kill", no_such_process) + handler = SigIntHandler() + handler.attach(1234) + handler(signal.SIGINT, None) # must not raise + assert [(r.levelname, r.message) for r in caplog.records] == [ + ("WARNING", "Received SIGINT, but the command has already exited") + ] + + +def _sigint_blocked() -> bool: + """Whether the calling thread blocks SIGINT.""" + return signal.SIGINT in signal.pthread_sigmask(signal.SIG_BLOCK, []) + + +def test_start_thread_blocks_sigint_only_in_the_new_thread() -> None: + blocked_in_thread: list[bool] = [] + thread = threading.Thread( + target=lambda: blocked_in_thread.append(_sigint_blocked()) + ) + start_thread(thread) + thread.join() + assert blocked_in_thread == [True] + assert not _sigint_blocked() + + +@pytest.mark.usefixtures("restore_sigint") +def test_every_duct_thread_starts_with_sigint_blocked( + temp_output_dir: str, monkeypatch: pytest.MonkeyPatch +) -> None: + # A thread inherits the mask its creator has at the moment it is started + blocked_at_start: list[bool] = [] + real_start = threading.Thread.start + + def recording_start(self: threading.Thread) -> None: + blocked_at_start.append(_sigint_blocked()) + real_start(self) + + monkeypatch.setattr(threading.Thread, "start", recording_start) + assert run_duct_command(["true"], output_prefix=temp_output_dir) == 0 + assert blocked_at_start and all(blocked_at_start) + + +@pytest.mark.usefixtures("restore_sigint") +def test_sigint_before_command_starts_does_not_start_it( + temp_output_dir: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_report = Report + + def report_then_sigint(*args: Any, **kwargs: Any) -> Any: + # Report is built after duct's handler is installed and before Popen + report = real_report(*args, **kwargs) + os.kill(os.getpid(), signal.SIGINT) + return report + + monkeypatch.setattr("con_duct._duct_main.Report", report_then_sigint) + marker = tmp_path / "command-ran" + rc = run_duct_command(["touch", str(marker)], output_prefix=temp_output_dir) + + assert rc == 128 + signal.SIGINT + assert not marker.exists() + assert list(Path(temp_output_dir).iterdir()) == [] diff --git a/test/utils.py b/test/utils.py index d5ec9bef..ad928512 100644 --- a/test/utils.py +++ b/test/utils.py @@ -1,8 +1,16 @@ from __future__ import annotations from io import BytesIO from pathlib import Path +import re +import shlex +import subprocess from typing import Any +# duct logs this right after attaching its SIGINT handler to the command +# (_duct_main.execute). +# Anchored on the log format so the wrapped command's echoed stderr can't match. +DUCT_READY = re.compile(r"\] con-duct: duct \S+ is executing ") + def run_duct_command(cli_args: list[str], **kwargs: Any) -> int: """Helper to run duct with test-friendly defaults. @@ -40,6 +48,56 @@ def run_duct_command(cli_args: list[str], **kwargs: Any) -> int: return duct_execute(command=command, command_args=command_args, **defaults) # type: ignore[arg-type] +def start_duct(duct_cmd: str, args: list[str]) -> subprocess.Popen[str]: + """Start duct as a subprocess and return once it is ready for signals. + + Reads duct's stderr until the log line emitted just after the SIGINT + handler is attached to the command. The caller should finish with ``communicate()``, + which drains the rest of stderr and waits for duct to exit. + + That line is logged at INFO, so duct is started with ``--log-level=INFO`` + whatever DUCT_LOG_LEVEL or a .env file says; ``args`` must not lower it. + + Args: + duct_cmd: How to invoke duct, e.g. "duct" or "con-duct run" + args: Arguments for duct, including the command to run + + Returns: + The running duct process, with its SIGINT handler installed + """ + proc = subprocess.Popen( + [*shlex.split(duct_cmd), "--log-level=INFO", *args], + stderr=subprocess.PIPE, + text=True, + ) + assert proc.stderr is not None # for mypy + seen = [] + for line in proc.stderr: + if DUCT_READY.search(line): + return proc + seen.append(line) + raise RuntimeError( + f"duct exited with {proc.wait()} before logging that it is executing; " + f"its stderr:\n{''.join(seen)}" + ) + + +def rest_of_stderr(proc: subprocess.Popen[str]) -> str: + """What duct wrote to stderr since ``start_duct`` returned, for failure messages. + + Reads to end of file, so only call it once duct has exited. + + Args: + proc: A duct process from ``start_duct`` + + Returns: + The unread remainder of duct's stderr + """ + assert proc.stderr is not None # for mypy + rest: str = proc.stderr.read() + return rest + + class MockStream: """Mocks stderr or stdout""" diff --git a/tox.ini b/tox.ini index e6710d69..2e16b17d 100644 --- a/tox.ini +++ b/tox.ini @@ -8,6 +8,7 @@ minversion = 3.3.0 deps = pytest pytest-cov + coverage>=7.10 # for [tool.coverage.run] patch in pyproject.toml pytest-rerunfailures -e .[all] commands = @@ -37,6 +38,7 @@ description = Test with minimal dependencies (no optional extras) deps = pytest pytest-cov + coverage>=7.10 # for [tool.coverage.run] patch in pyproject.toml pytest-rerunfailures -e . commands = @@ -49,6 +51,7 @@ allowlist_externals = uv deps = pytest pytest-cov + coverage>=7.10 # for [tool.coverage.run] patch in pyproject.toml pytest-rerunfailures commands = # numpy<2 required because matplotlib 3.5 was compiled against numpy 1.x