diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index 1750abd..29f9462 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/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: diff --git a/tests/conftest.py b/tests/conftest.py index 4e7bad1..ee7c847 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 91b2d7e..06bbc83 100644 --- a/tests/test_cached_py_info.py +++ b/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")