🐛 fix(subprocess): add timeout to interpreter probing (#42)

Author: gaborbernat

src/python_discovery/_cached_py_info.py

@@ -15,7 +15,7 @@
 from contextlib import contextmanager
 from pathlib import Path
 from shlex import quote
-from subprocess import Popen  # noqa: S404
+from subprocess import Popen, TimeoutExpired  # noqa: S404
 from typing import TYPE_CHECKING, Final
 
 from ._cache import NoOpCache
@@ -206,8 +206,11 @@ def _run_subprocess(
                 encoding="utf-8",
                 errors="backslashreplace",
             )
-            out, err = process.communicate()
+            out, err = process.communicate(timeout=5)
             code = process.returncode
+        except TimeoutExpired:
+            process.kill()
+            out, err, code = "", "timed out", -1
         except OSError as os_error:
             out, err, code = "", os_error.strerror, os_error.errno
     if code != 0:

tests/conftest.py

@@ -26,4 +26,5 @@ def _ensure_py_info_cache_empty(session_cache: DiskCache) -> Generator[None]:
 def _skip_if_test_in_system(session_cache: DiskCache) -> None:
     current = PythonInfo.current(session_cache)
     if current.system_executable is not None:  # pragma: no cover
-        pytest.skip("test not valid if run under system")
+        msg = "test not valid if run under system"
+        raise pytest.skip.Exception(msg)

tests/test_cached_py_info.py

@@ -5,6 +5,7 @@
 import os
 import sys
 from pathlib import Path
+from subprocess import TimeoutExpired
 from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, patch
 
@@ -110,6 +111,17 @@ def test_run_subprocess_with_cookies(mocker: MockerFixture) -> None:
     assert mock_stdout.write.call_count == 2
 
 
+def test_run_subprocess_timeout(mocker: MockerFixture) -> None:
+    mock_process = MagicMock()
+    mock_process.communicate.side_effect = TimeoutExpired(cmd="python", timeout=30)
+    mocker.patch("python_discovery._cached_py_info.Popen", return_value=mock_process)
+    failure, result = _run_subprocess(PythonInfo, sys.executable, dict(os.environ))
+    assert failure is not None
+    assert "timed out" in str(failure)
+    assert result is None
+    mock_process.kill.assert_called_once()
+
+
 def test_run_subprocess_nonzero_exit(mocker: MockerFixture) -> None:
     mock_process = MagicMock()
     mock_process.communicate.return_value = ("some output", "some error")