diff --git a/docs/changelog/148.bugfix.rst b/docs/changelog/148.bugfix.rst new file mode 100644 index 0000000..9df1cc6 --- /dev/null +++ b/docs/changelog/148.bugfix.rst @@ -0,0 +1,2 @@ +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/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index 9d2fbc1..0a28cf3 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -30,15 +30,16 @@ from ._cache import ContentStore, PyInfoCache class CacheEntryMeta(TypedDict): - """Identity of a cached interrogation result: the queried binary and the script that produced it.""" + """Identity of a cached interrogation result: the queried binary, its arguments and the script.""" st_mtime: float path: str hash: str | None + arguments: 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,34 @@ 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] + 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) - result = _CACHE[exe_path] = py_info - if isinstance(result, PythonInfo): + 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 -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: @@ -107,17 +112,22 @@ def _get_via_file_cache( 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} + 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) + 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) + 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)}) @@ -214,16 +224,20 @@ 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] + 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", LogCmd(cmd)) + _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, @@ -248,7 +262,8 @@ def _run_subprocess( 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, " 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..7f20e1c 100644 --- a/src/python_discovery/_windows/_pep514.py +++ b/src/python_discovery/_windows/_pep514.py @@ -125,7 +125,10 @@ 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") + 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: msg(f"{key_path}/InstallPath", "missing") 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_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/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..bb9250e 100644 --- a/tests/windows/test_windows_pep514.py +++ b/tests/windows/test_windows_pep514.py @@ -1,10 +1,22 @@ from __future__ import annotations +import json +import os +import shutil +import subprocess import sys import textwrap +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import MagicMock import pytest +if TYPE_CHECKING: + from collections.abc import Callable + + from pytest_mock import MockerFixture + @pytest.mark.usefixtures("_mock_registry") def test_pep514_discovers_interpreters() -> None: @@ -154,3 +166,66 @@ 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.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", "expected"), + [ + 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( + load_exe_with_args: Callable[[object], tuple[str, str | None] | None], registered: str | None, expected: str | None +) -> None: + assert load_exe_with_args(registered) == (r"C:\py.exe", expected) + + +@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" 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") + assert launcher is not None + script = ( + "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")] + ) + + info = next( + _propose.propose_interpreters(PythonSpec.from_string_spec("python3.11"), DiskCache(tmp_path), os.environ) + ) + + assert Path(info.executable).samefile(selected)