diff --git a/pm/operations.py b/pm/operations.py index a78dcf7aef..8513028786 100644 --- a/pm/operations.py +++ b/pm/operations.py @@ -196,7 +196,7 @@ def ensure_environment( if executable is not None: _tool(Path("unused/python"), executable) # Validate before any write. - def build(generation: Path) -> Path: + def build(generation: Path, base_python: Path) -> Path: (generation / "pyproject.toml").write_text( '[project]\nname = "hermes-side-environment"\nversion = "0"\n' 'requires-python = ">=3.11"\ndependencies = ' @@ -207,8 +207,8 @@ def ensure_environment( seed = root / previous["generation"] / "uv.lock" if seed.is_file(): shutil.copyfile(seed, generation / "uv.lock") - return build_environment(source=generation, out=generation / "venv", frozen=False, - explicit=explicit, timeout=timeout) + return build_environment(source=generation, out=generation / "venv", python=base_python, + frozen=False, explicit=explicit, timeout=timeout) return _ensure_generation(name, root, {"requirements": requirements}, build, record={"requirements": requirements}, explicit=explicit, @@ -236,16 +236,17 @@ def ensure_project_environment( except FileNotFoundError as exc: raise InstallError("venv", f"locked project environment needs {project / manifest}") from exc - def build(generation: Path) -> Path: - return build_environment(source=project, out=generation / "venv", extras=extras, groups=groups, - no_install_project=True, frozen=True, explicit=explicit, timeout=timeout) + def build(generation: Path, base_python: Path) -> Path: + return build_environment(source=project, out=generation / "venv", python=base_python, + extras=extras, groups=groups, no_install_project=True, + frozen=True, explicit=explicit, timeout=timeout) return _ensure_generation(name, root, {"manifests": manifests, "extras": extras, "groups": groups}, build, record={"extras": extras, "groups": groups}, explicit=explicit) def _ensure_generation( - name: str, root: Path, inputs: dict, build: Callable[[Path], Path], *, record: dict, explicit: bool, + name: str, root: Path, inputs: dict, build: Callable[[Path, Path], Path], *, record: dict, explicit: bool, executable: str | None = None, ) -> Path: """Select the generation whose inputs match, building one only when none does. @@ -254,6 +255,7 @@ def _ensure_generation( The prior generation survives both successful replacement and failed builds. """ from hermes_cli.runtime_state import _lock + from pm._uv import _toolchain from pm.install import _refuse_lazy, lazy_installs_allowed from pm.lock import Lockfile, _write from pm import paths @@ -263,17 +265,28 @@ def _ensure_generation( target = current_target() inputs = {**inputs, "python": lock.version("python"), "target": target, "artifacts": [item["sha256"] for item in lock.artifacts("python", target)]} - identity = hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest() - def current() -> Path | None: + def selected_python() -> Path | None: + tools = _toolchain(realize=False) + return tools[1].resolve() if tools is not None else None + + def identity(base_python: Path) -> str: + # The same pinned artifact can live in different stores. A venv's + # pyvenv.cfg keeps the original interpreter path, not just its version. + return hashlib.sha256(json.dumps({**inputs, "interpreter": str(base_python)}, + sort_keys=True).encode()).hexdigest() + + def current(base_python: Path | None) -> Path | None: + if base_python is None: + return None selected = _selection(root) python = environment_python(name, root=root) - if (selected.get("inputs") == identity and python is not None + if (selected.get("inputs") == identity(base_python) and python is not None and (executable is None or _tool(python, executable) is not None)): return python return None - existing = current() + existing = current(selected_python()) if existing is not None: return existing if not explicit and not lazy_installs_allowed(): @@ -281,16 +294,23 @@ def _ensure_generation( root.mkdir(parents=True, exist_ok=True) with (root / ".install.lock").open("a+b") as mutex: _lock(mutex.fileno(), wait=True) - existing = current() + base_python = selected_python() + existing = current(base_python) if existing is not None: return existing + if base_python is None: + tools = _toolchain(explicit=explicit) + if tools is None: + raise InstallError(name, "PM's pinned toolchain is unavailable") + base_python = tools[1].resolve() generation = root / f"gen-{uuid.uuid4().hex}" generation.mkdir() try: - python = build(generation) + python = build(generation, base_python) if executable is not None and _tool(python, executable) is None: raise InstallError(name, f"installed requirements do not provide {executable!r}") - _write(root / "active.json", {"generation": generation.name, "inputs": identity, **record}) + _write(root / "active.json", {"generation": generation.name, + "inputs": identity(base_python), **record}) except BaseException: shutil.rmtree(generation, ignore_errors=True) raise diff --git a/tests/pm/test_uv_python.py b/tests/pm/test_uv_python.py index 93500b33e4..056423ac78 100644 --- a/tests/pm/test_uv_python.py +++ b/tests/pm/test_uv_python.py @@ -103,6 +103,61 @@ def test_all_uv_commands_keep_the_pm_interpreter(installed_uv, monkeypatch): assert dict(os.environ) == before +def test_project_environment_replaces_generation_when_pinned_python_moves(installed_uv, tmp_path, monkeypatch): + """An unchanged dependency pin cannot reuse a venv made by another tools store.""" + from pm import operations + import pm.registry as registry + from tests.pm._fixtures import _run, _wheel, stage_host_python + + root, uv, facts, target, digest = installed_uv + store = root / "store" + from pm.store import tree_digest + facts.record("uv", "test", uv.parent.name, {}, store, target=target, + artifacts=[digest], digest=tree_digest(uv.parent)) + + class FixturePython(Python): + binary_rel = {"win32": "Scripts/python.exe", "posix": "bin/python"} + + monkeypatch.setitem(registry._packages, "python", FixturePython()) + interpreters = [stage_host_python(store / name / "bin" / "python") + for name in ("python-a", "python-b")] + project = tmp_path / "project" + project.mkdir() + wheel = _wheel(tmp_path, "side_dep") + (project / "pyproject.toml").write_text( + '[project]\nname="pm-generation-proof"\nversion="1"\nrequires-python=">=3.11"\n' + 'dependencies=["side-dep==1.0"]\n[tool.uv]\npackage=false\nno-index=true\n' + f'find-links=["{wheel.parent.as_posix()}"]\n', encoding="utf-8", + ) + env = {key: value for key, value in os.environ.items() + if not key.startswith(("UV_", "PYTHON")) and key != "VIRTUAL_ENV"} + env.update(UV_CACHE_DIR=str(tmp_path / "cache"), UV_PYTHON_DOWNLOADS="never") + _run([str(uv), "lock", "--python", str(interpreters[0])], cwd=project, env=env) + locked = (project / "uv.lock").read_bytes() + selection_root = tmp_path / "selection" + chosen = [] + for interpreter in (*interpreters, interpreters[0]): + facts.record("python", "test", interpreter.parent.parent.name, {}, store, + target=target, artifacts=[digest], digest=tree_digest(interpreter.parent.parent)) + selected = operations.ensure_project_environment( + "test-environment", project, root=selection_root, explicit=True) + chosen.append(selected) + assert Path(_run([str(selected), "-I", "-c", + "import sys, side_dep; print(sys.base_prefix)"], cwd=project, env=env)) == interpreter.parent.parent + assert (project / "uv.lock").read_bytes() == locked + assert chosen[0] != chosen[1] != chosen[2] + + # A failed replacement must leave the previously selected environment intact. + facts.record("python", "test", interpreters[1].parent.parent.name, {}, store, + target=target, artifacts=[digest], digest=tree_digest(interpreters[1].parent.parent)) + active = (selection_root / "active.json").read_bytes() + monkeypatch.setattr(operations, "build_environment", lambda **kwargs: (_ for _ in ()).throw(RuntimeError("build failed"))) + with pytest.raises(RuntimeError, match="build failed"): + operations.ensure_project_environment("test-environment", project, root=selection_root, explicit=True) + assert (selection_root / "active.json").read_bytes() == active + assert operations.environment_python("test-environment", root=selection_root) == chosen[-1] + + def test_uv_refuses_discovery_when_pm_python_is_missing(installed_uv, monkeypatch): root, _, facts, target, digest = installed_uv ensure = importlib.import_module("pm.install")