fix: honor PEP 514 executable arguments during discovery

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`.

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, "

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:

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")

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
 

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

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)

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)