diff --git a/pytest_examples/lint.py b/pytest_examples/lint.py index 07d22b0..471bd85 100644 --- a/pytest_examples/lint.py +++ b/pytest_examples/lint.py @@ -1,7 +1,7 @@ from __future__ import annotations as _annotations import re -from subprocess import PIPE, Popen +from subprocess import PIPE, Popen, TimeoutExpired from textwrap import indent from typing import TYPE_CHECKING @@ -52,7 +52,15 @@ def ruff_check( args = ruff, 'check', '-', *config.ruff_config(), *extra_ruff_args p = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, encoding='utf-8') - stdout, stderr = p.communicate(example.source, timeout=10) + try: + stdout, stderr = p.communicate(example.source, timeout=10) + # a timed out `communicate` leaves the child running with its pipes open, so reap it before giving up + # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.communicate + except TimeoutExpired: + p.kill() + p.communicate() + raise + if p.returncode == 1 and stdout: def replace_offset(m: re.Match[str]): diff --git a/tests/test_lint.py b/tests/test_lint.py index b735bb1..ec8a709 100644 --- a/tests/test_lint.py +++ b/tests/test_lint.py @@ -1,6 +1,9 @@ +from subprocess import Popen, TimeoutExpired +from typing import Any + import pytest -from pytest_examples import CodeExample +from pytest_examples import CodeExample, lint from pytest_examples.config import ExamplesConfig from pytest_examples.lint import FormatError, black_check, ruff_check @@ -32,6 +35,34 @@ def test_ruff_offset(): ruff_check(example, ExamplesConfig()) +def test_ruff_timeout_kills_the_process(monkeypatch: pytest.MonkeyPatch) -> None: + spawned: list[Popen[str]] = [] + + class TimingOutPopen(Popen[str]): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + spawned.append(self) + + def communicate(self, input: str | None = None, timeout: float | None = None) -> tuple[str, str]: + # `ruff_check` passes a timeout, the cleanup it runs afterwards does not + if timeout is not None: + raise TimeoutExpired(self.args, timeout) + return super().communicate() + + monkeypatch.setattr(lint, 'Popen', TimingOutPopen) + + example = CodeExample.create('x = 1\n') + with pytest.raises(TimeoutExpired): + ruff_check(example, ExamplesConfig()) + + ruff_process = spawned[0] + + assert ruff_process.poll() is not None + assert ruff_process.stdin.closed + assert ruff_process.stdout.closed + assert ruff_process.stderr.closed + + def test_black_line_length(): example = CodeExample.create(long_function, start_line=4) with pytest.raises(FormatError, match='^black failed:\n'):