From 52677f76db2186c6097c1af8a31b3438ad9e7ed9 Mon Sep 17 00:00:00 2001 From: blouirts123 <201396345+blouirts123@users.noreply.github.com> Date: Fri, 16 Oct 2026 05:21:36 +0530 Subject: [PATCH 1/4] fix: honor PEP 514 executable arguments during discovery --- src/python_discovery/_cache.py | 6 + src/python_discovery/_cached_py_info.py | 157 ++++++++++++++++------ src/python_discovery/_py_info.py | 33 ++++- src/python_discovery/_windows/_pep514.py | 15 ++- src/python_discovery/_windows/_propose.py | 10 +- tests/test_cache.py | 12 ++ tests/test_cached_py_info.py | 45 ++++++- tests/windows/conftest.py | 2 +- tests/windows/test_windows_pep514.py | 88 ++++++++++++ 9 files changed, 320 insertions(+), 48 deletions(-) diff --git a/src/python_discovery/_cache.py b/src/python_discovery/_cache.py index a728723..cba67b7 100644 --- a/src/python_discovery/_cache.py +++ b/src/python_discovery/_cache.py @@ -135,6 +135,12 @@ def py_info(self, path: Path) -> DiskContentStore: key = sha256(str(path).encode("utf-8")).hexdigest() return DiskContentStore(self._py_info_dir, key) + def _py_info_with_args(self, path: Path, arguments: str) -> DiskContentStore: + """Return a distinct store when one executable is queried with launcher arguments.""" + identity = f"{path}\0{arguments}" + key = sha256(identity.encode("utf-8")).hexdigest() + return DiskContentStore(self._py_info_dir, key) + def py_info_clear(self) -> None: """Remove all cached interpreter information.""" folder = self._py_info_dir diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index 9d2fbc1..81e6867 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -17,7 +17,7 @@ from pathlib import Path from shlex import quote from subprocess import Popen, TimeoutExpired # ruff:ignore[suspicious-subprocess-import] -from typing import TYPE_CHECKING, Final +from typing import TYPE_CHECKING, Any, Final, cast from ._cache import NoOpCache from ._py_info import PythonInfo @@ -29,16 +29,17 @@ from ._cache import ContentStore, PyInfoCache - class CacheEntryMeta(TypedDict): - """Identity of a cached interrogation result: the queried binary and the script that produced it.""" + class CacheEntryMeta(TypedDict, total=False): + """Identity of a cached interrogation result: the queried binary, arguments, and script.""" st_mtime: float path: str hash: str | None + arguments_hash: str | None -_CACHE: OrderedDict[Path, PythonInfo | Exception] = OrderedDict() -_CACHE[Path(sys.executable)] = PythonInfo() +_CACHE: OrderedDict[tuple[Path, str | None], PythonInfo | Exception] = OrderedDict() +_CACHE[Path(sys.executable), None] = PythonInfo() _LOGGER: Final[logging.Logger] = logging.getLogger(__name__) _PY_INFO_SCRIPT: Final[Path] = Path(__file__).resolve().parent / "_py_info_collect.py" _DEFAULT_QUERY_TIMEOUT: Final[float] = 15.0 @@ -58,9 +59,10 @@ def from_exe( # ruff:ignore[too-many-arguments] *, raise_on_error: bool = True, ignore_cache: bool = False, + executable_args: str | None = None, ) -> PythonInfo | None: env = os.environ if env is None else env - result = _get_from_cache(cls, cache, exe, env, ignore_cache=ignore_cache) + result = _get_from_cache(cls, cache, exe, env, ignore_cache=ignore_cache, executable_args=executable_args) if isinstance(result, Exception): if raise_on_error: raise result @@ -69,31 +71,35 @@ def from_exe( # ruff:ignore[too-many-arguments] return result -def _get_from_cache( +def _get_from_cache( # ruff:ignore[too-many-arguments] cls: type[PythonInfo], cache: PyInfoCache | None, exe: str, env: Mapping[str, str], *, ignore_cache: bool = True, + executable_args: str | None = None, ) -> PythonInfo | Exception: exe_path = Path(exe) - if not ignore_cache and exe_path in _CACHE: - result = _CACHE[exe_path] + cache_key = (exe_path, executable_args) + if not ignore_cache and cache_key in _CACHE: + result = _CACHE[cache_key] else: - py_info = _get_via_file_cache(cls, cache, exe_path, exe, env) - result = _CACHE[exe_path] = py_info - if isinstance(result, PythonInfo): + py_info = _get_via_file_cache(cls, cache, exe_path, exe, env, executable_args=executable_args) + result = _CACHE[cache_key] = py_info + if isinstance(result, PythonInfo) and executable_args is None: result.executable = exe return result -def _get_via_file_cache( +def _get_via_file_cache( # ruff:ignore[too-many-arguments] cls: type[PythonInfo], cache: PyInfoCache | None, path: Path, exe: str, env: Mapping[str, str], + *, + executable_args: str | None = None, ) -> PythonInfo | Exception: path_text = str(path) try: @@ -106,18 +112,18 @@ def _get_via_file_cache( py_info_hash = None resolved_cache = cache if cache is not None else NoOpCache() - py_info_store = resolved_cache.py_info(path) - entry_meta: CacheEntryMeta = {"st_mtime": path_modified, "path": path_text, "hash": py_info_hash} + py_info_store = _cache_store(resolved_cache, path, executable_args) + entry_meta = _cache_entry_meta(path_modified, path_text, py_info_hash, executable_args) with py_info_store.locked(): cached = _read_cache_entry(cls, py_info_store, entry_meta) if isinstance(cached, _UnsupportedInterpreterError): return cached py_info = cached if py_info is None: - failure, py_info = _run_subprocess(cls, exe, env) + failure, py_info = _run_subprocess(cls, exe, env, executable_args=executable_args) if failure is not None and not isinstance(failure, _UnsupportedInterpreterError): _LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure) - failure, py_info = _run_subprocess(cls, exe, env) + failure, py_info = _run_subprocess(cls, exe, env, executable_args=executable_args) if isinstance(failure, _UnsupportedInterpreterError): _LOGGER.warning("%s", failure) # the verdict is permanent, warn once and remember it py_info_store.write({**entry_meta, "unsupported": str(failure)}) @@ -132,6 +138,25 @@ def _get_via_file_cache( return py_info +def _cache_store(cache: PyInfoCache, path: Path, executable_args: str | None) -> ContentStore: + if executable_args is not None and (store_for_args := getattr(cache, "_py_info_with_args", None)) is not None: + return store_for_args(path, executable_args) + return cache.py_info(path) + + +def _arguments_hash(executable_args: str) -> str: + return hashlib.sha256(executable_args.encode("utf-8")).hexdigest() + + +def _cache_entry_meta( + path_modified: float, path_text: str, py_info_hash: str | None, executable_args: str | None +) -> CacheEntryMeta: + entry_meta: CacheEntryMeta = {"st_mtime": path_modified, "path": path_text, "hash": py_info_hash} + if executable_args is not None: + entry_meta["arguments_hash"] = _arguments_hash(executable_args) + return entry_meta + + def _read_cache_entry( cls: type[PythonInfo], py_info_store: ContentStore, @@ -214,41 +239,24 @@ def _run_subprocess( cls: type[PythonInfo], exe: str, env: Mapping[str, str], + *, + executable_args: str | None = None, ) -> tuple[Exception | None, PythonInfo | None]: start_cookie = gen_cookie() end_cookie = gen_cookie() timeout = _query_timeout(env.get("PY_DISCOVERY_TIMEOUT")) with _resolve_py_info_script() as py_info_script: - cmd = [exe, str(py_info_script), start_cookie, end_cookie] - env = dict(env) - env.pop("__PYVENV_LAUNCHER__", None) - env["PYTHONUTF8"] = "1" - _LOGGER.debug("get interpreter info via cmd: %s", LogCmd(cmd)) - try: - process = Popen( # ruff:ignore[subprocess-without-shell-equals-true] - cmd, - universal_newlines=True, - stdin=subprocess.PIPE, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - env=env, - encoding="utf-8", - errors="backslashreplace", - ) - out, err = process.communicate(timeout=timeout) - code = process.returncode - except TimeoutExpired: - process.kill() - process.communicate() - out, err, code = "", "timed out", -1 - except OSError as os_error: - out, err, code = "", os_error.strerror, os_error.errno + if executable_args is not None and os.name != "nt": + msg = "PEP 514 ExecutableArguments can only be used on Windows" + return RuntimeError(msg), None + out, err, code = _launch_query(exe, py_info_script, start_cookie, end_cookie, env, timeout, executable_args) if code != 0: return _query_failure(exe, out, err, code), None out, raw_out, out_starts, out_ends = _extract_between_cookies(out, start_cookie, end_cookie) try: result = cls.from_json(out) - result.executable = exe + if executable_args is None: + result.executable = exe except json.JSONDecodeError as exc: _LOGGER.warning( "subprocess %s returned invalid JSON; raw stdout %d chars, start cookie %s, end cookie %s, " @@ -267,6 +275,71 @@ def _run_subprocess( return None, result +def _launch_query( # ruff:ignore[too-many-arguments, too-many-positional-arguments] + exe: str, + py_info_script: Path, + start_cookie: str, + end_cookie: str, + env: Mapping[str, str], + timeout: float | None, + executable_args: str | None, +) -> tuple[str, str | None, int | None]: + env = dict(env) + env.pop("__PYVENV_LAUNCHER__", None) + env["PYTHONUTF8"] = "1" + tail = [str(py_info_script), start_cookie, end_cookie] + if executable_args is None: + raw_command = None + logged_command: LogCmd | str = LogCmd([exe, *tail]) + else: + raw_command = _windows_command_line(exe, executable_args, tail) + logged_command = ( + f"{subprocess.list2cmdline([exe])} {subprocess.list2cmdline(tail)}" + ) + _LOGGER.debug("get interpreter info via cmd: %s", logged_command) + try: # ruff:ignore[too-many-statements-in-try-clause] + if raw_command is None: + process: Popen[str] = Popen( # ruff:ignore[subprocess-without-shell-equals-true] + [exe, *tail], + universal_newlines=True, + stdin=subprocess.PIPE, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + env=env, + encoding="utf-8", + errors="backslashreplace", + ) + else: + # Popen supports raw command-line strings on Windows, which typeshed omits here. + process = Popen( # ruff:ignore[subprocess-without-shell-equals-true] + cast(Any, raw_command), # ruff:ignore[runtime-cast-value] + universal_newlines=True, + stdin=subprocess.PIPE, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + env=env, + encoding="utf-8", + errors="backslashreplace", + executable=exe, + ) + out, err = process.communicate(timeout=timeout) + code = process.returncode + except TimeoutExpired: + process.kill() + process.communicate() + out, err, code = "", "timed out", -1 + except OSError as os_error: + out, err, code = "", os_error.strerror, os_error.errno + return out, err, code + + +def _windows_command_line(exe: str, executable_args: str, trailing_args: list[str]) -> str: + """Build a Windows command line, keeping PEP 514's leading argument text intact.""" + prefix = subprocess.list2cmdline([exe]) + suffix = subprocess.list2cmdline(trailing_args) + return f"{prefix} {executable_args} {suffix}" + + @lru_cache(maxsize=128) def _query_timeout(raw: str | None) -> float | None: if raw is None: diff --git a/src/python_discovery/_py_info.py b/src/python_discovery/_py_info.py index cfc2b94..15bbdc6 100644 --- a/src/python_discovery/_py_info.py +++ b/src/python_discovery/_py_info.py @@ -378,10 +378,41 @@ def from_exe( # ruff:ignore[too-many-arguments] :param resolve_to_host: resolve through virtualenv layers to the system interpreter. :param env: environment mapping; defaults to :data:`os.environ`. """ + return cls._from_exe_with_args( + exe, + cache, + executable_args=None, + raise_on_error=raise_on_error, + ignore_cache=ignore_cache, + resolve_to_host=resolve_to_host, + env=env, + ) + + @classmethod + def _from_exe_with_args( # ruff:ignore[too-many-arguments] + cls, + exe: str, + cache: PyInfoCache | None = None, + *, + executable_args: str | None, + raise_on_error: bool = True, + ignore_cache: bool = False, + resolve_to_host: bool = True, + env: Mapping[str, str] | None = None, + ) -> PythonInfo | None: + """Interrogate an executable with optional platform-specific leading arguments.""" from ._cached_py_info import from_exe # ruff:ignore[import-outside-top-level] env = os.environ if env is None else env - proposed = from_exe(cls, cache, exe, env=env, raise_on_error=raise_on_error, ignore_cache=ignore_cache) + proposed = from_exe( + cls, + cache, + exe, + env=env, + raise_on_error=raise_on_error, + ignore_cache=ignore_cache, + executable_args=executable_args, + ) if isinstance(proposed, PythonInfo) and resolve_to_host: try: diff --git a/src/python_discovery/_windows/_pep514.py b/src/python_discovery/_windows/_pep514.py index 51c13da..372d03a 100644 --- a/src/python_discovery/_windows/_pep514.py +++ b/src/python_discovery/_windows/_pep514.py @@ -125,13 +125,26 @@ def load_exe(hive_name: str, company: str, company_key: Any, tag: str) -> tuple[ try: with winreg.OpenKeyEx(company_key, rf"{tag}\InstallPath") as ip_key, ip_key: # ty: ignore[unresolved-attribute] if (exe := _resolve_exe(ip_key, key_path)) is not None and os.path.exists(exe): - return exe, get_value(ip_key, "ExecutableArguments") + valid, executable_args = _load_executable_arguments(ip_key, key_path) + if valid: + return exe, executable_args + return None msg(key_path, f"could not load exe with value {exe}") except OSError: msg(f"{key_path}/InstallPath", "missing") return None +def _load_executable_arguments(ip_key: Any, key_path: str) -> tuple[bool, str | None]: # ruff:ignore[any-type] + executable_args = get_value(ip_key, "ExecutableArguments") + if executable_args is not None and not isinstance(executable_args, str): + msg(f"{key_path}/ExecutableArguments", f"expected a string, got {type(executable_args).__name__}") + return False, None + if executable_args is not None and not executable_args.strip(): + executable_args = None + return True, executable_args + + def _resolve_exe(ip_key: Any, key_path: str) -> str | None: # ruff:ignore[any-type] if (exe := get_value(ip_key, "ExecutablePath")) is not None: return exe diff --git a/src/python_discovery/_windows/_propose.py b/src/python_discovery/_windows/_propose.py index 6b9ff33..57846cb 100644 --- a/src/python_discovery/_windows/_propose.py +++ b/src/python_discovery/_windows/_propose.py @@ -36,13 +36,19 @@ def propose_interpreters( reverse=True, ) - for name, major, minor, arch, threaded, exe, _ in existing: + for name, major, minor, arch, threaded, exe, executable_args in existing: implementation = _IMPLEMENTATION_BY_ORG.get(name, name) skip_pre_filter = implementation.lower() != "cpython" registry_spec = PythonSpec("", implementation, major, minor, None, arch, exe, free_threaded=threaded) if skip_pre_filter or registry_spec.satisfies(spec): - interpreter = Pep514PythonInfo.from_exe(exe, cache, env=env, raise_on_error=False) + interpreter = Pep514PythonInfo._from_exe_with_args( # ruff:ignore[private-member-access] + exe, + cache, + env=env, + executable_args=executable_args, + raise_on_error=False, + ) if interpreter is not None and interpreter.satisfies(spec, impl_must_match=True): yield interpreter diff --git a/tests/test_cache.py b/tests/test_cache.py index f868b11..7e656ef 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -58,6 +58,18 @@ def test_disk_cache_py_info(tmp_path: Path) -> None: assert isinstance(store, DiskContentStore) +def test_disk_cache_py_info_with_args_uses_a_distinct_store(tmp_path: Path) -> None: + cache = DiskCache(tmp_path) + path = Path("/some/path") + no_args = cache.py_info(path) + with_args = cache._py_info_with_args(path, "-3.11") + another_args = cache._py_info_with_args(path, "-3.12") + + assert with_args._file != no_args._file + assert with_args._file != another_args._file + assert with_args._file == cache._py_info_with_args(path, "-3.11")._file + + def test_disk_cache_py_info_clear(tmp_path: Path) -> None: cache = DiskCache(tmp_path) store = cache.py_info(Path("/some/path")) diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 6bd41c0..2f12c0c 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -5,7 +5,7 @@ import os import sys from pathlib import Path -from subprocess import TimeoutExpired +from subprocess import TimeoutExpired, list2cmdline from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch @@ -14,11 +14,13 @@ from python_discovery import DiskCache, PythonInfo from python_discovery._cached_py_info import ( LogCmd, + _get_from_cache, _get_via_file_cache, _load_cached_py_info, _resolve_py_info_script, _run_subprocess, _script_hash, + _windows_command_line, gen_cookie, ) @@ -45,6 +47,47 @@ def test_log_cmd_repr_with_env() -> None: assert "FOO" in result +def test_windows_command_line_keeps_pep514_arguments_before_quoted_tail() -> None: + exe = r"C:\Program Files\Python Launcher\py.exe" + executable_args = '-3.11 --config "settings folder\\python.json"' + trailing_args = [r"C:\temporary files\collector.py", "start-cookie", "end-cookie"] + + command = _windows_command_line(exe, executable_args, trailing_args) + + assert command.startswith(f"{list2cmdline([exe])} ") + assert f" {executable_args} " in command + assert command.endswith(list2cmdline(trailing_args)) + + +def test_in_memory_cache_identity_includes_executable_arguments(tmp_path: Path, mocker: MockerFixture) -> None: + from collections import OrderedDict + + path = tmp_path / "launcher.exe" + cache = DiskCache(tmp_path / "cache") + first = PythonInfo() + second = PythonInfo() + mocker.patch("python_discovery._cached_py_info._CACHE", OrderedDict()) + query = mocker.patch( + "python_discovery._cached_py_info._get_via_file_cache", + side_effect=[first, second], + ) + + result_311 = _get_from_cache( + PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.11" + ) + result_312 = _get_from_cache( + PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.12" + ) + cached_311 = _get_from_cache( + PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.11" + ) + + assert result_311 is first + assert result_312 is second + assert cached_311 is first + assert query.call_count == 2 + + def test_resolve_py_info_script_file_exists() -> None: with _resolve_py_info_script() as script: assert script.exists() diff --git a/tests/windows/conftest.py b/tests/windows/conftest.py index 3df8148..ebce8b6 100644 --- a/tests/windows/conftest.py +++ b/tests/windows/conftest.py @@ -172,4 +172,4 @@ def _populate_pyinfo_cache(monkeypatch: pytest.MonkeyPatch) -> None: ] for _, major, minor, arch, threaded, exe in interpreters: info = _mock_pyinfo(major, minor, arch, exe, threaded) - monkeypatch.setitem(_CACHE, Path(info.executable), info) + monkeypatch.setitem(_CACHE, (Path(info.executable), None), info) diff --git a/tests/windows/test_windows_pep514.py b/tests/windows/test_windows_pep514.py index 1dedfb6..595d2d5 100644 --- a/tests/windows/test_windows_pep514.py +++ b/tests/windows/test_windows_pep514.py @@ -1,9 +1,15 @@ from __future__ import annotations +import json +import os +import shutil +import subprocess import sys import textwrap +from pathlib import Path import pytest +from typing_extensions import Self @pytest.mark.usefixtures("_mock_registry") @@ -154,3 +160,85 @@ def test_pep514_run(capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptur f"{prefix}HKEY_CURRENT_USER/PythonCore/3.X error: invalid format 3.X", ] assert caplog.messages == expected_logs + + +@pytest.mark.usefixtures("_mock_registry") +@pytest.mark.parametrize( + ("registered_args", "expected_args"), + [ + ('-3.11 --config "settings folder\\python.json"', '-3.11 --config "settings folder\\python.json"'), + ("", None), + (" ", None), + (42, False), + ], +) +def test_pep514_loads_executable_arguments( + monkeypatch: pytest.MonkeyPatch, registered_args: object, expected_args: str | bool | None +) -> None: + from python_discovery._windows import _pep514 as pep514 + + class Key: + def __enter__(self) -> Self: + return self + + def __exit__(self, *_args: object) -> None: + return None + + monkeypatch.setattr(pep514.winreg, "OpenKeyEx", lambda *_args: Key()) + monkeypatch.setattr(pep514, "_resolve_exe", lambda *_args: r"C:\PythonLauncher\py.exe") + monkeypatch.setattr(pep514.os.path, "exists", lambda _path: True) + monkeypatch.setattr( + pep514, + "get_value", + lambda _key, name: registered_args if name == "ExecutableArguments" else None, + ) + + result = pep514.load_exe("HKCU", "PythonCore", object(), "3.11") + + if expected_args is False: + assert result is None + else: + assert result == (r"C:\PythonLauncher\py.exe", expected_args) + + +@pytest.mark.skipif(sys.platform != "win32", reason="requires the Windows Python launcher") +def test_launcher_registry_arguments_return_an_executable_for_selected_python( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + from python_discovery import DiskCache, PythonSpec + from python_discovery._windows import _propose + + launcher = shutil.which("py") + if launcher is None: + # pytest accepts a reason, but ty's bundled Python 3.9 pytest signature is stale. + pytest.skip("the Windows Python launcher is unavailable") # ty: ignore[too-many-positional-arguments] + selector = "-3.11" + script = ( + "import json, struct, sys; print(json.dumps((sys.executable, *sys.version_info[:2], struct.calcsize('P') * 8)))" + ) + try: + selected = subprocess.run( + [launcher, selector, "-c", script], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + except subprocess.CalledProcessError: + # pytest accepts a reason, but ty's bundled Python 3.9 pytest signature is stale. + pytest.skip(f"the Windows launcher cannot select {selector}") # ty: ignore[too-many-positional-arguments] + selected_exe, major, minor, arch = json.loads(selected) + registry_row = ("PythonCore", major, minor, arch, False, launcher, selector) + monkeypatch.setattr(_propose, "discover_pythons", lambda: iter([registry_row])) + + spec = PythonSpec.from_string_spec(f"python{major}.{minor}") + info = next(_propose.propose_interpreters(spec, DiskCache(tmp_path), dict(os.environ))) + assert info.version_info[:2] == (major, minor) + assert Path(info.executable).samefile(selected_exe) + + actual = subprocess.run( + [info.executable, "-c", "import sys; print(*sys.version_info[:2])"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + assert actual == f"{major} {minor}" From aa2c1a464f9675b297fa0d48728de71c559539f9 Mon Sep 17 00:00:00 2001 From: blouirts123 <201396345+blouirts123@users.noreply.github.com> Date: Fri, 16 Oct 2026 05:22:15 +0530 Subject: [PATCH 2/4] docs: add PEP 514 discovery changelog entry --- docs/changelog/148.bugfix.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 docs/changelog/148.bugfix.rst diff --git a/docs/changelog/148.bugfix.rst b/docs/changelog/148.bugfix.rst new file mode 100644 index 0000000..913ed82 --- /dev/null +++ b/docs/changelog/148.bugfix.rst @@ -0,0 +1,2 @@ +Honor PEP 514 ExecutableArguments when interrogating registered Windows interpreters and +distinguish launcher arguments in cached results. From 2a8f32384c5136c4f59bcc7ddc6e5790bd9a5859 Mon Sep 17 00:00:00 2001 From: blouirts123 <201396345+blouirts123@users.noreply.github.com> Date: Fri, 16 Oct 2026 05:54:02 +0530 Subject: [PATCH 3/4] Cover launcher argument query and cache paths across platforms --- tests/test_cached_py_info.py | 115 +++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 2f12c0c..6fccd2f 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -1,11 +1,13 @@ from __future__ import annotations +import hashlib import json import logging import os import sys from pathlib import Path from subprocess import TimeoutExpired, list2cmdline +from types import SimpleNamespace from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch @@ -14,8 +16,10 @@ from python_discovery import DiskCache, PythonInfo from python_discovery._cached_py_info import ( LogCmd, + _cache_store, _get_from_cache, _get_via_file_cache, + _launch_query, _load_cached_py_info, _resolve_py_info_script, _run_subprocess, @@ -27,6 +31,8 @@ if TYPE_CHECKING: from pytest_mock import MockerFixture + from python_discovery._cache import DiskContentStore + def test_gen_cookie_length() -> None: cookie = gen_cookie() @@ -59,6 +65,115 @@ def test_windows_command_line_keeps_pep514_arguments_before_quoted_tail() -> Non assert command.endswith(list2cmdline(trailing_args)) +def test_cache_store_uses_argument_aware_disk_store(tmp_path: Path, mocker: MockerFixture) -> None: + cache = DiskCache(tmp_path) + path = Path(sys.executable) + executable_args = "-3.11" + result = PythonInfo() + run = mocker.patch( + "python_discovery._cached_py_info._run_subprocess", + return_value=(None, result), + ) + + cached = _get_via_file_cache( + PythonInfo, cache, path, sys.executable, dict(os.environ), executable_args=executable_args + ) + + args_store = cache._py_info_with_args(path, executable_args) + data = args_store.read() + assert cached is result + assert data is not None + assert data["arguments_hash"] == hashlib.sha256(executable_args.encode("utf-8")).hexdigest() + assert cache.py_info(path).read() is None + run.assert_called_once_with(PythonInfo, sys.executable, dict(os.environ), executable_args=executable_args) + + +def test_cache_store_falls_back_for_legacy_cache_with_args(tmp_path: Path) -> None: + disk_cache = DiskCache(tmp_path) + + class LegacyCache: + def __init__(self, cache: DiskCache) -> None: + self._cache = cache + + def py_info(self, path: Path) -> DiskContentStore: + return self._cache.py_info(path) + + def py_info_clear(self) -> None: + self._cache.py_info_clear() + + cache = LegacyCache(disk_cache) + path = Path(sys.executable) + + store = _cache_store(cache, path, "-3.11") + + assert store._file == cache.py_info(path)._file + + +def test_run_subprocess_rejects_executable_args_off_windows( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture +) -> None: + monkeypatch.setattr( + "python_discovery._cached_py_info.os", + SimpleNamespace(name="posix", write=os.write, close=os.close), + ) + launch = mocker.patch("python_discovery._cached_py_info._launch_query") + + failure, result = _run_subprocess(PythonInfo, "py", {}, executable_args="-3.11") + + assert isinstance(failure, RuntimeError) + assert "only be used on Windows" in str(failure) + assert result is None + launch.assert_not_called() + + +def test_run_subprocess_with_args_preserves_selected_executable( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture +) -> None: + launcher = r"C:\Python Launcher\py.exe" + selected_python = r"C:\Python311\python.exe" + start_cookie, end_cookie = "a" * 32, "b" * 32 + payload = PythonInfo().to_dict() + payload["executable"] = selected_python + output = f"pre{start_cookie[::-1]}{json.dumps(payload)}{end_cookie[::-1]}post" + monkeypatch.setattr( + "python_discovery._cached_py_info.os", + SimpleNamespace(name="nt", write=os.write, close=os.close), + ) + mocker.patch("python_discovery._cached_py_info.gen_cookie", side_effect=[start_cookie, end_cookie]) + launch = mocker.patch( + "python_discovery._cached_py_info._launch_query", + return_value=(output, None, 0), + ) + + failure, result = _run_subprocess(PythonInfo, launcher, {}, executable_args="-3.11") + + assert failure is None + assert result is not None + assert result.executable == selected_python + assert launch.call_args.args[0] == launcher + assert launch.call_args.args[-1] == "-3.11" + + +def test_launch_query_passes_raw_pep514_command_to_popen(mocker: MockerFixture) -> None: + exe = r"C:\Program Files\Python Launcher\py.exe" + executable_args = '-3.11 --config "settings folder\\python.json"' + script = Path(r"C:\temporary files\collector.py") + start_cookie, end_cookie = "a" * 32, "b" * 32 + env = {"__PYVENV_LAUNCHER__": "ignored", "PATH": "kept"} + process = MagicMock() + process.communicate.return_value = ("output", "") + process.returncode = 0 + popen = mocker.patch("python_discovery._cached_py_info.Popen", return_value=process) + + result = _launch_query(exe, script, start_cookie, end_cookie, env, 12, executable_args) + + expected_command = _windows_command_line(exe, executable_args, [str(script), start_cookie, end_cookie]) + assert popen.call_args.args == (expected_command,) + assert popen.call_args.kwargs["executable"] == exe + assert popen.call_args.kwargs["env"] == {"PATH": "kept", "PYTHONUTF8": "1"} + assert result == ("output", "", 0) + + def test_in_memory_cache_identity_includes_executable_arguments(tmp_path: Path, mocker: MockerFixture) -> None: from collections import OrderedDict From 68149d6c83e58a44210e42744fc1e52717daf800 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Thu, 15 Oct 2026 17:14:49 -0700 Subject: [PATCH 4/4] refactor: reduce PEP 514 argument handling Pass ExecutableArguments through the existing query path instead of a parallel one: one Popen call that takes a raw command line when arguments are present, and the arguments stored in the disk cache entry metadata rather than a separate DiskCache store reached through getattr. The metadata key is always written, so a plain query of a launcher no longer reads an entry written for the same launcher with arguments, which the previous optional key allowed for caches without the private method. Drop the unreachable non-Windows guard and the registry helper that returned a validity flag, and move the argument tests to their own file. --- docs/changelog/148.bugfix.rst | 4 +- src/python_discovery/_cache.py | 6 - src/python_discovery/_cached_py_info.py | 148 ++++++----------- src/python_discovery/_windows/_pep514.py | 16 +- tests/test_cache.py | 12 -- tests/test_cached_py_info.py | 160 +------------------ tests/test_cached_py_info_executable_args.py | 79 +++++++++ tests/windows/test_windows_pep514.py | 115 ++++++------- 8 files changed, 181 insertions(+), 359 deletions(-) create mode 100644 tests/test_cached_py_info_executable_args.py diff --git a/docs/changelog/148.bugfix.rst b/docs/changelog/148.bugfix.rst index 913ed82..9df1cc6 100644 --- a/docs/changelog/148.bugfix.rst +++ b/docs/changelog/148.bugfix.rst @@ -1,2 +1,2 @@ -Honor PEP 514 ExecutableArguments when interrogating registered Windows interpreters and -distinguish launcher arguments in cached results. +Pass PEP 514 ``ExecutableArguments`` to registered Windows interpreters when querying them, so a registration that +points at a launcher such as ``py.exe -3.11`` reports the interpreter it selects - by :user:`blouirts123`. diff --git a/src/python_discovery/_cache.py b/src/python_discovery/_cache.py index cba67b7..a728723 100644 --- a/src/python_discovery/_cache.py +++ b/src/python_discovery/_cache.py @@ -135,12 +135,6 @@ def py_info(self, path: Path) -> DiskContentStore: key = sha256(str(path).encode("utf-8")).hexdigest() return DiskContentStore(self._py_info_dir, key) - def _py_info_with_args(self, path: Path, arguments: str) -> DiskContentStore: - """Return a distinct store when one executable is queried with launcher arguments.""" - identity = f"{path}\0{arguments}" - key = sha256(identity.encode("utf-8")).hexdigest() - return DiskContentStore(self._py_info_dir, key) - def py_info_clear(self) -> None: """Remove all cached interpreter information.""" folder = self._py_info_dir diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index 81e6867..0a28cf3 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -17,7 +17,7 @@ from pathlib import Path from shlex import quote from subprocess import Popen, TimeoutExpired # ruff:ignore[suspicious-subprocess-import] -from typing import TYPE_CHECKING, Any, Final, cast +from typing import TYPE_CHECKING, Final from ._cache import NoOpCache from ._py_info import PythonInfo @@ -29,13 +29,13 @@ from ._cache import ContentStore, PyInfoCache - class CacheEntryMeta(TypedDict, total=False): - """Identity of a cached interrogation result: the queried binary, arguments, and script.""" + class CacheEntryMeta(TypedDict): + """Identity of a cached interrogation result: the queried binary, its arguments and the script.""" st_mtime: float path: str hash: str | None - arguments_hash: str | None + arguments: str | None _CACHE: OrderedDict[tuple[Path, str | None], PythonInfo | Exception] = OrderedDict() @@ -80,13 +80,12 @@ def _get_from_cache( # ruff:ignore[too-many-arguments] ignore_cache: bool = True, executable_args: str | None = None, ) -> PythonInfo | Exception: - exe_path = Path(exe) - cache_key = (exe_path, executable_args) - if not ignore_cache and cache_key in _CACHE: - result = _CACHE[cache_key] + key = Path(exe), executable_args + if not ignore_cache and key in _CACHE: + result = _CACHE[key] else: - py_info = _get_via_file_cache(cls, cache, exe_path, exe, env, executable_args=executable_args) - result = _CACHE[cache_key] = py_info + result = _CACHE[key] = _get_via_file_cache(cls, cache, key[0], exe, env, executable_args=executable_args) + # with launcher arguments the queried binary is not the interpreter, so keep the one the query reported if isinstance(result, PythonInfo) and executable_args is None: result.executable = exe return result @@ -112,18 +111,23 @@ def _get_via_file_cache( # ruff:ignore[too-many-arguments] py_info_hash = None resolved_cache = cache if cache is not None else NoOpCache() - py_info_store = _cache_store(resolved_cache, path, executable_args) - entry_meta = _cache_entry_meta(path_modified, path_text, py_info_hash, executable_args) + py_info_store = resolved_cache.py_info(path) + entry_meta: CacheEntryMeta = { + "st_mtime": path_modified, + "path": path_text, + "hash": py_info_hash, + "arguments": executable_args, + } with py_info_store.locked(): cached = _read_cache_entry(cls, py_info_store, entry_meta) if isinstance(cached, _UnsupportedInterpreterError): return cached py_info = cached if py_info is None: - failure, py_info = _run_subprocess(cls, exe, env, executable_args=executable_args) + failure, py_info = _run_subprocess(cls, exe, env, executable_args) if failure is not None and not isinstance(failure, _UnsupportedInterpreterError): _LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure) - failure, py_info = _run_subprocess(cls, exe, env, executable_args=executable_args) + failure, py_info = _run_subprocess(cls, exe, env, executable_args) if isinstance(failure, _UnsupportedInterpreterError): _LOGGER.warning("%s", failure) # the verdict is permanent, warn once and remember it py_info_store.write({**entry_meta, "unsupported": str(failure)}) @@ -138,25 +142,6 @@ def _get_via_file_cache( # ruff:ignore[too-many-arguments] return py_info -def _cache_store(cache: PyInfoCache, path: Path, executable_args: str | None) -> ContentStore: - if executable_args is not None and (store_for_args := getattr(cache, "_py_info_with_args", None)) is not None: - return store_for_args(path, executable_args) - return cache.py_info(path) - - -def _arguments_hash(executable_args: str) -> str: - return hashlib.sha256(executable_args.encode("utf-8")).hexdigest() - - -def _cache_entry_meta( - path_modified: float, path_text: str, py_info_hash: str | None, executable_args: str | None -) -> CacheEntryMeta: - entry_meta: CacheEntryMeta = {"st_mtime": path_modified, "path": path_text, "hash": py_info_hash} - if executable_args is not None: - entry_meta["arguments_hash"] = _arguments_hash(executable_args) - return entry_meta - - def _read_cache_entry( cls: type[PythonInfo], py_info_store: ContentStore, @@ -239,17 +224,39 @@ def _run_subprocess( cls: type[PythonInfo], exe: str, env: Mapping[str, str], - *, executable_args: str | None = None, ) -> tuple[Exception | None, PythonInfo | None]: start_cookie = gen_cookie() end_cookie = gen_cookie() timeout = _query_timeout(env.get("PY_DISCOVERY_TIMEOUT")) with _resolve_py_info_script() as py_info_script: - if executable_args is not None and os.name != "nt": - msg = "PEP 514 ExecutableArguments can only be used on Windows" - return RuntimeError(msg), None - out, err, code = _launch_query(exe, py_info_script, start_cookie, end_cookie, env, timeout, executable_args) + cmd: list[str] | str = [exe, str(py_info_script), start_cookie, end_cookie] + if executable_args is not None: + # PEP 514 ExecutableArguments is raw command line text, placed before the script per the PEP + cmd = f"{subprocess.list2cmdline(cmd[:1])} {executable_args} {subprocess.list2cmdline(cmd[1:])}" + env = dict(env) + env.pop("__PYVENV_LAUNCHER__", None) + env["PYTHONUTF8"] = "1" + _LOGGER.debug("get interpreter info via cmd: %s", cmd if isinstance(cmd, str) else LogCmd(cmd)) + try: + process = Popen( # ruff:ignore[subprocess-without-shell-equals-true] + cmd, + universal_newlines=True, + stdin=subprocess.PIPE, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + env=env, + encoding="utf-8", + errors="backslashreplace", + ) + out, err = process.communicate(timeout=timeout) + code = process.returncode + except TimeoutExpired: + process.kill() + process.communicate() + out, err, code = "", "timed out", -1 + except OSError as os_error: + out, err, code = "", os_error.strerror, os_error.errno if code != 0: return _query_failure(exe, out, err, code), None out, raw_out, out_starts, out_ends = _extract_between_cookies(out, start_cookie, end_cookie) @@ -275,71 +282,6 @@ def _run_subprocess( return None, result -def _launch_query( # ruff:ignore[too-many-arguments, too-many-positional-arguments] - exe: str, - py_info_script: Path, - start_cookie: str, - end_cookie: str, - env: Mapping[str, str], - timeout: float | None, - executable_args: str | None, -) -> tuple[str, str | None, int | None]: - env = dict(env) - env.pop("__PYVENV_LAUNCHER__", None) - env["PYTHONUTF8"] = "1" - tail = [str(py_info_script), start_cookie, end_cookie] - if executable_args is None: - raw_command = None - logged_command: LogCmd | str = LogCmd([exe, *tail]) - else: - raw_command = _windows_command_line(exe, executable_args, tail) - logged_command = ( - f"{subprocess.list2cmdline([exe])} {subprocess.list2cmdline(tail)}" - ) - _LOGGER.debug("get interpreter info via cmd: %s", logged_command) - try: # ruff:ignore[too-many-statements-in-try-clause] - if raw_command is None: - process: Popen[str] = Popen( # ruff:ignore[subprocess-without-shell-equals-true] - [exe, *tail], - universal_newlines=True, - stdin=subprocess.PIPE, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - env=env, - encoding="utf-8", - errors="backslashreplace", - ) - else: - # Popen supports raw command-line strings on Windows, which typeshed omits here. - process = Popen( # ruff:ignore[subprocess-without-shell-equals-true] - cast(Any, raw_command), # ruff:ignore[runtime-cast-value] - universal_newlines=True, - stdin=subprocess.PIPE, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - env=env, - encoding="utf-8", - errors="backslashreplace", - executable=exe, - ) - out, err = process.communicate(timeout=timeout) - code = process.returncode - except TimeoutExpired: - process.kill() - process.communicate() - out, err, code = "", "timed out", -1 - except OSError as os_error: - out, err, code = "", os_error.strerror, os_error.errno - return out, err, code - - -def _windows_command_line(exe: str, executable_args: str, trailing_args: list[str]) -> str: - """Build a Windows command line, keeping PEP 514's leading argument text intact.""" - prefix = subprocess.list2cmdline([exe]) - suffix = subprocess.list2cmdline(trailing_args) - return f"{prefix} {executable_args} {suffix}" - - @lru_cache(maxsize=128) def _query_timeout(raw: str | None) -> float | None: if raw is None: diff --git a/src/python_discovery/_windows/_pep514.py b/src/python_discovery/_windows/_pep514.py index 372d03a..7f20e1c 100644 --- a/src/python_discovery/_windows/_pep514.py +++ b/src/python_discovery/_windows/_pep514.py @@ -125,9 +125,9 @@ def load_exe(hive_name: str, company: str, company_key: Any, tag: str) -> tuple[ try: with winreg.OpenKeyEx(company_key, rf"{tag}\InstallPath") as ip_key, ip_key: # ty: ignore[unresolved-attribute] if (exe := _resolve_exe(ip_key, key_path)) is not None and os.path.exists(exe): - valid, executable_args = _load_executable_arguments(ip_key, key_path) - if valid: - return exe, executable_args + if (args := get_value(ip_key, "ExecutableArguments")) is None or isinstance(args, str): + return exe, args if args and args.strip() else None + msg(f"{key_path}/ExecutableArguments", f"expected a string, got {type(args).__name__}") return None msg(key_path, f"could not load exe with value {exe}") except OSError: @@ -135,16 +135,6 @@ def load_exe(hive_name: str, company: str, company_key: Any, tag: str) -> tuple[ return None -def _load_executable_arguments(ip_key: Any, key_path: str) -> tuple[bool, str | None]: # ruff:ignore[any-type] - executable_args = get_value(ip_key, "ExecutableArguments") - if executable_args is not None and not isinstance(executable_args, str): - msg(f"{key_path}/ExecutableArguments", f"expected a string, got {type(executable_args).__name__}") - return False, None - if executable_args is not None and not executable_args.strip(): - executable_args = None - return True, executable_args - - def _resolve_exe(ip_key: Any, key_path: str) -> str | None: # ruff:ignore[any-type] if (exe := get_value(ip_key, "ExecutablePath")) is not None: return exe diff --git a/tests/test_cache.py b/tests/test_cache.py index 7e656ef..f868b11 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -58,18 +58,6 @@ def test_disk_cache_py_info(tmp_path: Path) -> None: assert isinstance(store, DiskContentStore) -def test_disk_cache_py_info_with_args_uses_a_distinct_store(tmp_path: Path) -> None: - cache = DiskCache(tmp_path) - path = Path("/some/path") - no_args = cache.py_info(path) - with_args = cache._py_info_with_args(path, "-3.11") - another_args = cache._py_info_with_args(path, "-3.12") - - assert with_args._file != no_args._file - assert with_args._file != another_args._file - assert with_args._file == cache._py_info_with_args(path, "-3.11")._file - - def test_disk_cache_py_info_clear(tmp_path: Path) -> None: cache = DiskCache(tmp_path) store = cache.py_info(Path("/some/path")) diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 6fccd2f..6bd41c0 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -1,13 +1,11 @@ from __future__ import annotations -import hashlib import json import logging import os import sys from pathlib import Path -from subprocess import TimeoutExpired, list2cmdline -from types import SimpleNamespace +from subprocess import TimeoutExpired from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch @@ -16,23 +14,17 @@ from python_discovery import DiskCache, PythonInfo from python_discovery._cached_py_info import ( LogCmd, - _cache_store, - _get_from_cache, _get_via_file_cache, - _launch_query, _load_cached_py_info, _resolve_py_info_script, _run_subprocess, _script_hash, - _windows_command_line, gen_cookie, ) if TYPE_CHECKING: from pytest_mock import MockerFixture - from python_discovery._cache import DiskContentStore - def test_gen_cookie_length() -> None: cookie = gen_cookie() @@ -53,156 +45,6 @@ def test_log_cmd_repr_with_env() -> None: assert "FOO" in result -def test_windows_command_line_keeps_pep514_arguments_before_quoted_tail() -> None: - exe = r"C:\Program Files\Python Launcher\py.exe" - executable_args = '-3.11 --config "settings folder\\python.json"' - trailing_args = [r"C:\temporary files\collector.py", "start-cookie", "end-cookie"] - - command = _windows_command_line(exe, executable_args, trailing_args) - - assert command.startswith(f"{list2cmdline([exe])} ") - assert f" {executable_args} " in command - assert command.endswith(list2cmdline(trailing_args)) - - -def test_cache_store_uses_argument_aware_disk_store(tmp_path: Path, mocker: MockerFixture) -> None: - cache = DiskCache(tmp_path) - path = Path(sys.executable) - executable_args = "-3.11" - result = PythonInfo() - run = mocker.patch( - "python_discovery._cached_py_info._run_subprocess", - return_value=(None, result), - ) - - cached = _get_via_file_cache( - PythonInfo, cache, path, sys.executable, dict(os.environ), executable_args=executable_args - ) - - args_store = cache._py_info_with_args(path, executable_args) - data = args_store.read() - assert cached is result - assert data is not None - assert data["arguments_hash"] == hashlib.sha256(executable_args.encode("utf-8")).hexdigest() - assert cache.py_info(path).read() is None - run.assert_called_once_with(PythonInfo, sys.executable, dict(os.environ), executable_args=executable_args) - - -def test_cache_store_falls_back_for_legacy_cache_with_args(tmp_path: Path) -> None: - disk_cache = DiskCache(tmp_path) - - class LegacyCache: - def __init__(self, cache: DiskCache) -> None: - self._cache = cache - - def py_info(self, path: Path) -> DiskContentStore: - return self._cache.py_info(path) - - def py_info_clear(self) -> None: - self._cache.py_info_clear() - - cache = LegacyCache(disk_cache) - path = Path(sys.executable) - - store = _cache_store(cache, path, "-3.11") - - assert store._file == cache.py_info(path)._file - - -def test_run_subprocess_rejects_executable_args_off_windows( - monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture -) -> None: - monkeypatch.setattr( - "python_discovery._cached_py_info.os", - SimpleNamespace(name="posix", write=os.write, close=os.close), - ) - launch = mocker.patch("python_discovery._cached_py_info._launch_query") - - failure, result = _run_subprocess(PythonInfo, "py", {}, executable_args="-3.11") - - assert isinstance(failure, RuntimeError) - assert "only be used on Windows" in str(failure) - assert result is None - launch.assert_not_called() - - -def test_run_subprocess_with_args_preserves_selected_executable( - monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture -) -> None: - launcher = r"C:\Python Launcher\py.exe" - selected_python = r"C:\Python311\python.exe" - start_cookie, end_cookie = "a" * 32, "b" * 32 - payload = PythonInfo().to_dict() - payload["executable"] = selected_python - output = f"pre{start_cookie[::-1]}{json.dumps(payload)}{end_cookie[::-1]}post" - monkeypatch.setattr( - "python_discovery._cached_py_info.os", - SimpleNamespace(name="nt", write=os.write, close=os.close), - ) - mocker.patch("python_discovery._cached_py_info.gen_cookie", side_effect=[start_cookie, end_cookie]) - launch = mocker.patch( - "python_discovery._cached_py_info._launch_query", - return_value=(output, None, 0), - ) - - failure, result = _run_subprocess(PythonInfo, launcher, {}, executable_args="-3.11") - - assert failure is None - assert result is not None - assert result.executable == selected_python - assert launch.call_args.args[0] == launcher - assert launch.call_args.args[-1] == "-3.11" - - -def test_launch_query_passes_raw_pep514_command_to_popen(mocker: MockerFixture) -> None: - exe = r"C:\Program Files\Python Launcher\py.exe" - executable_args = '-3.11 --config "settings folder\\python.json"' - script = Path(r"C:\temporary files\collector.py") - start_cookie, end_cookie = "a" * 32, "b" * 32 - env = {"__PYVENV_LAUNCHER__": "ignored", "PATH": "kept"} - process = MagicMock() - process.communicate.return_value = ("output", "") - process.returncode = 0 - popen = mocker.patch("python_discovery._cached_py_info.Popen", return_value=process) - - result = _launch_query(exe, script, start_cookie, end_cookie, env, 12, executable_args) - - expected_command = _windows_command_line(exe, executable_args, [str(script), start_cookie, end_cookie]) - assert popen.call_args.args == (expected_command,) - assert popen.call_args.kwargs["executable"] == exe - assert popen.call_args.kwargs["env"] == {"PATH": "kept", "PYTHONUTF8": "1"} - assert result == ("output", "", 0) - - -def test_in_memory_cache_identity_includes_executable_arguments(tmp_path: Path, mocker: MockerFixture) -> None: - from collections import OrderedDict - - path = tmp_path / "launcher.exe" - cache = DiskCache(tmp_path / "cache") - first = PythonInfo() - second = PythonInfo() - mocker.patch("python_discovery._cached_py_info._CACHE", OrderedDict()) - query = mocker.patch( - "python_discovery._cached_py_info._get_via_file_cache", - side_effect=[first, second], - ) - - result_311 = _get_from_cache( - PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.11" - ) - result_312 = _get_from_cache( - PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.12" - ) - cached_311 = _get_from_cache( - PythonInfo, cache, str(path), dict(os.environ), ignore_cache=False, executable_args="-3.11" - ) - - assert result_311 is first - assert result_312 is second - assert cached_311 is first - assert query.call_count == 2 - - def test_resolve_py_info_script_file_exists() -> None: with _resolve_py_info_script() as script: assert script.exists() diff --git a/tests/test_cached_py_info_executable_args.py b/tests/test_cached_py_info_executable_args.py new file mode 100644 index 0000000..78254b3 --- /dev/null +++ b/tests/test_cached_py_info_executable_args.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import os +import sys +from pathlib import Path +from subprocess import list2cmdline +from typing import TYPE_CHECKING, Final +from unittest.mock import MagicMock + +import pytest + +from python_discovery import DiskCache, PythonInfo +from python_discovery._cached_py_info import _get_from_cache, _get_via_file_cache, _run_subprocess + +if TYPE_CHECKING: + from pytest_mock import MockerFixture + +LAUNCHER: Final[str] = r"C:\Program Files\Python Launcher\py.exe" +ARGS: Final[str] = '-3.11 --config "settings folder\\python.json"' +SELECTED: Final[str] = r"C:\Python311\python.exe" + + +@pytest.fixture +def popen(mocker: MockerFixture) -> MagicMock: + start, end = "a" * 32, "b" * 32 + mocker.patch("python_discovery._cached_py_info.gen_cookie", side_effect=[start, end]) + info = PythonInfo() + info.executable = SELECTED + out = f"{start[::-1]}{info.to_json()}{end[::-1]}" + process = MagicMock(**{"communicate.return_value": (out, ""), "returncode": 0}) + return mocker.patch("python_discovery._cached_py_info.Popen", return_value=process) + + +def test_executable_args_precede_query_script(popen: MagicMock) -> None: + _run_subprocess(PythonInfo, LAUNCHER, dict(os.environ), ARGS) + + cmd = popen.call_args.args[0] + assert cmd.startswith(f"{list2cmdline([LAUNCHER])} {ARGS} ") + assert cmd.endswith(f" {'a' * 32} {'b' * 32}") + + +@pytest.mark.usefixtures("popen") +def test_executable_args_report_the_selected_interpreter() -> None: + failure, result = _run_subprocess(PythonInfo, LAUNCHER, dict(os.environ), ARGS) + + assert failure is None + assert result is not None + assert result.executable == SELECTED + + +def test_executable_args_disk_entry_not_reused_without_args(tmp_path: Path, mocker: MockerFixture) -> None: + with_args, without_args = PythonInfo(), PythonInfo() + with_args.executable = "selected" + query = mocker.patch( + "python_discovery._cached_py_info._run_subprocess", + side_effect=[(None, with_args), (None, without_args)], + ) + cache, path, env = DiskCache(tmp_path), Path(sys.executable), dict(os.environ) + _get_via_file_cache(PythonInfo, cache, path, sys.executable, env, executable_args=ARGS) + + result = _get_via_file_cache(PythonInfo, cache, path, sys.executable, env) + + assert result is without_args + assert query.call_count == 2 + + +def test_executable_args_memory_cache_keyed_by_args(tmp_path: Path, mocker: MockerFixture) -> None: + mocker.patch.dict("python_discovery._cached_py_info._CACHE", clear=True) + first, second = PythonInfo(), PythonInfo() + query = mocker.patch("python_discovery._cached_py_info._get_via_file_cache", side_effect=[first, second]) + cache, env = DiskCache(tmp_path), dict(os.environ) + + results = [ + _get_from_cache(PythonInfo, cache, LAUNCHER, env, ignore_cache=False, executable_args=args) + for args in ("-3.11", "-3.12", "-3.11") + ] + + assert results == [first, second, first] + assert query.call_count == 2 diff --git a/tests/windows/test_windows_pep514.py b/tests/windows/test_windows_pep514.py index 595d2d5..bb9250e 100644 --- a/tests/windows/test_windows_pep514.py +++ b/tests/windows/test_windows_pep514.py @@ -7,9 +7,15 @@ import sys import textwrap from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import MagicMock import pytest -from typing_extensions import Self + +if TYPE_CHECKING: + from collections.abc import Callable + + from pytest_mock import MockerFixture @pytest.mark.usefixtures("_mock_registry") @@ -162,83 +168,64 @@ def test_pep514_run(capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptur assert caplog.messages == expected_logs +@pytest.fixture +def load_exe_with_args(mocker: MockerFixture) -> Callable[[object], tuple[str, str | None] | None]: + from python_discovery._windows import _pep514 as pep514 + + mocker.patch.object(pep514.winreg, "OpenKeyEx", return_value=MagicMock()) + mocker.patch.object(pep514, "_resolve_exe", return_value=r"C:\py.exe") + mocker.patch.object(pep514.os.path, "exists", return_value=True) + + def load(registered: object) -> tuple[str, str | None] | None: + mocker.patch.object(pep514, "get_value", return_value=registered) + return pep514.load_exe("HKCU", "PythonCore", MagicMock(), "3.11") + + return load + + @pytest.mark.usefixtures("_mock_registry") @pytest.mark.parametrize( - ("registered_args", "expected_args"), + ("registered", "expected"), [ - ('-3.11 --config "settings folder\\python.json"', '-3.11 --config "settings folder\\python.json"'), - ("", None), - (" ", None), - (42, False), + pytest.param('-3.11 --config "a b"', '-3.11 --config "a b"', id="kept"), + pytest.param(None, None, id="missing"), + pytest.param("", None, id="empty"), + pytest.param(" ", None, id="blank"), ], ) def test_pep514_loads_executable_arguments( - monkeypatch: pytest.MonkeyPatch, registered_args: object, expected_args: str | bool | None + load_exe_with_args: Callable[[object], tuple[str, str | None] | None], registered: str | None, expected: str | None ) -> None: - from python_discovery._windows import _pep514 as pep514 - - class Key: - def __enter__(self) -> Self: - return self + assert load_exe_with_args(registered) == (r"C:\py.exe", expected) - def __exit__(self, *_args: object) -> None: - return None - monkeypatch.setattr(pep514.winreg, "OpenKeyEx", lambda *_args: Key()) - monkeypatch.setattr(pep514, "_resolve_exe", lambda *_args: r"C:\PythonLauncher\py.exe") - monkeypatch.setattr(pep514.os.path, "exists", lambda _path: True) - monkeypatch.setattr( - pep514, - "get_value", - lambda _key, name: registered_args if name == "ExecutableArguments" else None, - ) - - result = pep514.load_exe("HKCU", "PythonCore", object(), "3.11") - - if expected_args is False: - assert result is None - else: - assert result == (r"C:\PythonLauncher\py.exe", expected_args) +@pytest.mark.usefixtures("_mock_registry") +def test_pep514_rejects_non_string_executable_arguments( + load_exe_with_args: Callable[[object], tuple[str, str | None] | None], +) -> None: + assert load_exe_with_args(42) is None -@pytest.mark.skipif(sys.platform != "win32", reason="requires the Windows Python launcher") -def test_launcher_registry_arguments_return_an_executable_for_selected_python( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: +@pytest.mark.skipif(sys.platform != "win32" or shutil.which("py") is None, reason="requires the Windows launcher") +def test_pep514_launcher_arguments_select_interpreter(mocker: MockerFixture, tmp_path: Path) -> None: from python_discovery import DiskCache, PythonSpec from python_discovery._windows import _propose launcher = shutil.which("py") - if launcher is None: - # pytest accepts a reason, but ty's bundled Python 3.9 pytest signature is stale. - pytest.skip("the Windows Python launcher is unavailable") # ty: ignore[too-many-positional-arguments] - selector = "-3.11" + assert launcher is not None script = ( - "import json, struct, sys; print(json.dumps((sys.executable, *sys.version_info[:2], struct.calcsize('P') * 8)))" + "import json, struct, sys; print(json.dumps([sys.executable, *sys.version_info[:2], struct.calcsize('P') * 8]))" + ) + probe = subprocess.run([launcher, "-3.11", "-c", script], capture_output=True, text=True, check=False) + if probe.returncode: + pytest.skip("the launcher cannot select 3.11") # ty: ignore[too-many-positional-arguments] # ty ships a stale pytest stub for 3.9 + selected, major, minor, arch = json.loads(probe.stdout) + mocker.patch.object( + _propose, "discover_pythons", return_value=[("PythonCore", major, minor, arch, False, launcher, "-3.11")] ) - try: - selected = subprocess.run( - [launcher, selector, "-c", script], - check=True, - capture_output=True, - text=True, - ).stdout.strip() - except subprocess.CalledProcessError: - # pytest accepts a reason, but ty's bundled Python 3.9 pytest signature is stale. - pytest.skip(f"the Windows launcher cannot select {selector}") # ty: ignore[too-many-positional-arguments] - selected_exe, major, minor, arch = json.loads(selected) - registry_row = ("PythonCore", major, minor, arch, False, launcher, selector) - monkeypatch.setattr(_propose, "discover_pythons", lambda: iter([registry_row])) - - spec = PythonSpec.from_string_spec(f"python{major}.{minor}") - info = next(_propose.propose_interpreters(spec, DiskCache(tmp_path), dict(os.environ))) - assert info.version_info[:2] == (major, minor) - assert Path(info.executable).samefile(selected_exe) - - actual = subprocess.run( - [info.executable, "-c", "import sys; print(*sys.version_info[:2])"], - check=True, - capture_output=True, - text=True, - ).stdout.strip() - assert actual == f"{major} {minor}" + + info = next( + _propose.propose_interpreters(PythonSpec.from_string_spec("python3.11"), DiskCache(tmp_path), os.environ) + ) + + assert Path(info.executable).samefile(selected)