Terminate ruff when it times out

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]):

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'):