diff --git a/pm/build_operations.py b/pm/build_operations.py index 8e2bfa933c..38d258b774 100644 --- a/pm/build_operations.py +++ b/pm/build_operations.py @@ -15,7 +15,10 @@ def check_project_lock( ) -> None: """Reject a missing or stale lock without rewriting source or creating a venv.""" from pm.environment import managed_environment + from pm.operations import _require_install_allowed + if not offline: + _require_install_allowed(explicit) source = Path(source).absolute() if not (source / "pyproject.toml").is_file(): raise InstallError("venv", f"project manifest is missing: {source}") @@ -33,7 +36,9 @@ def export_requirements( ) -> None: """Export locked runtime requirements, preserving markers and direct URL pins.""" from pm.environment import managed_environment + from pm.operations import _require_install_allowed + _require_install_allowed(explicit) source, out = Path(source).absolute(), Path(out).absolute() if not (source / "pyproject.toml").is_file(): raise InstallError("venv", f"project manifest is missing: {source}") @@ -60,8 +65,9 @@ def build_requirements_environment( this invocation's exclusively claimed output, not prior builds. """ from pm.environment import managed_environment, prune_site_pth - from pm.operations import _requirements + from pm.operations import _require_install_allowed, _requirements + _require_install_allowed(explicit) requirements = _requirements(requirements) if requirements or isinstance(requirements, str) else [] out = Path(out).absolute() wheelhouse = Path(wheelhouse).absolute() if wheelhouse is not None else None @@ -92,5 +98,5 @@ def prune_cache(cache: Path, *, ci: bool = False) -> None: from pm.environment import managed_environment cache = Path(cache).absolute() - environment = managed_environment(cache, cache=cache, explicit=True, output=sys.stderr) + environment = managed_environment(cache, cache=cache, realize=False, output=sys.stderr) environment.prune_cache(ci=ci) diff --git a/pm/environment.py b/pm/environment.py index 67791cefde..01c7fa98cf 100644 --- a/pm/environment.py +++ b/pm/environment.py @@ -108,11 +108,11 @@ def _base_environment(env: Mapping[str, str] | None = None) -> dict[str, str]: def managed_environment(destination: Path, *, python: Path | None = None, cache: Path | None = None, env: Mapping[str, str] | None = None, offline: bool = False, explicit: bool = False, - output: TextIO | None = None) -> PythonEnvironment: + output: TextIO | None = None, realize: bool = True) -> PythonEnvironment: from pm._uv import _toolchain from pm.packages import uv_cache_dir - tools = _toolchain(explicit=explicit) + tools = _toolchain(explicit=explicit, realize=realize) if tools is None: raise InstallError("venv", "PM's pinned toolchain is unavailable") uv, pinned_python = tools @@ -221,7 +221,8 @@ class PythonEnvironment: if result.returncode: raise classify_uv_failure("sync", result.returncode, result.stderr or result.stdout) - def export_requirements(self, source: Path, out: Path, *, extras: Sequence[str] = ()) -> None: + def export_requirements(self, source: Path, out: Path, *, extras: Sequence[str] = (), + timeout: int = 1800) -> None: from pm.workspace import classify_uv_failure command = ["export", "--frozen", "--python", str(self.python), "--no-default-groups", @@ -229,44 +230,37 @@ class PythonEnvironment: "--format", "requirements-txt", "--output-file", str(out)] for extra in sorted(set(extras)): command += ["--extra", extra] - result = self._run(command, cwd=source, timeout=1800) + result = self._run(command, cwd=source, timeout=timeout) if result.returncode: raise classify_uv_failure("export", result.returncode, result.stderr or result.stdout) def install_requirements(self, requirements: Sequence[str], *, wheelhouse: Path | None = None) -> None: - from pm.workspace import classify_uv_failure - if not requirements: return # A file avoids command-line length limits and shell/marker quoting. with tempfile.TemporaryDirectory(prefix="pm-requirements-") as temporary: requirements_file = Path(temporary) / "requirements.txt" requirements_file.write_text("\n".join(requirements) + "\n", encoding="utf-8") - command = ["pip", "install", "--no-config", "--python", str(self.executable), - "--requirements", str(requirements_file)] - if wheelhouse is not None: - command += ["--no-index", "--only-binary", ":all:", - "--find-links", str(wheelhouse)] - result = self._run(command, cwd=self.destination.parent, timeout=1800) + self._install_requirements_file(requirements_file, wheelhouse=wheelhouse) + + def _install_requirements_file(self, requirements: Path, *, wheelhouse: Path | None = None, + timeout: int = 1800) -> None: + from pm.workspace import classify_uv_failure + + command = ["pip", "install", "--no-config", "--python", str(self.executable), + "--requirements", str(requirements)] + if wheelhouse is not None: + command += ["--no-index", "--only-binary", ":all:", + "--find-links", str(wheelhouse.absolute())] + result = self._run(command, cwd=requirements.parent, timeout=timeout) if result.returncode: raise classify_uv_failure("pip", result.returncode, result.stderr or result.stdout) def install_wheelhouse(self, source: Path, wheelhouse: Path, *, timeout: int = 1800) -> None: """Install rebuilt wheels whose hashes the bundle manifest owns, not uv.lock.""" - from pm.workspace import classify_uv_failure - requirements = source / "requirements.txt" - commands = [ - ["export", "--frozen", "--python", str(self.python), "--no-default-groups", - "--no-emit-project", "--no-hashes", "--output-file", str(requirements)], - ["pip", "install", "--python", str(self.executable), "--no-index", - "--only-binary", ":all:", "--find-links", str(wheelhouse.absolute()), - "-r", str(requirements)], - ] - for command in commands: - result = self._run(command, cwd=source, timeout=timeout) - if result.returncode: - raise classify_uv_failure(command[0], result.returncode, result.stderr or result.stdout) + self.export_requirements(source, requirements, timeout=timeout) + self._install_requirements_file(requirements, wheelhouse=wheelhouse, timeout=timeout) self.check() def prune_cache(self, *, ci: bool = False) -> None: diff --git a/tests/pm/test_build_operations.py b/tests/pm/test_build_operations.py index e52fc13cd4..dc54ad95e9 100644 --- a/tests/pm/test_build_operations.py +++ b/tests/pm/test_build_operations.py @@ -50,7 +50,7 @@ def locked_source(tmp_path, build_tools): f'find-links=[{json.dumps(wheels.as_posix())}]\n', encoding="utf-8", ) lock_project(source, python=Path(sys.executable), cache=tmp_path / "cache", - env=build_tools, offline=True) + env=build_tools, offline=True, explicit=True) return source @@ -67,11 +67,11 @@ def test_check_lock_never_changes_source(locked_source, tmp_path, build_tools, l before = {p.relative_to(source): p.read_bytes() for p in source.rglob("*") if p.is_file()} if lock_state == "current": check_project_lock(source, python=Path(sys.executable), cache=tmp_path / "cache", - env=build_tools, offline=True) + env=build_tools, offline=True, explicit=True) else: with pytest.raises(InstallError): check_project_lock(source, python=Path(sys.executable), cache=tmp_path / "cache", - env=build_tools, offline=True) + env=build_tools, offline=True, explicit=True) assert {p.relative_to(source): p.read_bytes() for p in source.rglob("*") if p.is_file()} == before assert not (source / ".venv").exists() @@ -87,7 +87,7 @@ def test_frozen_export_keeps_markers_and_excludes_build_metadata(locked_source, before = {p.name: p.read_bytes() for p in source.iterdir() if p.is_file()} out = tmp_path / "export directory" / "requirements.txt" export_requirements(source, out, extras=["chosen", "chosen"], python=Path(sys.executable), - cache=tmp_path / "cache", env=build_tools) + cache=tmp_path / "cache", env=build_tools, explicit=True) text = out.read_text(encoding="utf-8") requirements = {r.name: r for r in map(Requirement, text.splitlines())} assert set(requirements) == {"base-dep", "chosen-dep"} @@ -121,10 +121,10 @@ def test_frozen_export_preserves_git_commit_pin(locked_source, tmp_path, build_t manifest = locked_source / "pyproject.toml" manifest.write_text(manifest.read_text().replace('"base-dep==1.0"', json.dumps(f"git-dep @ {url}")), encoding="utf-8") - lock_project(locked_source, python=Path(sys.executable), cache=tmp_path / "cache", env=build_tools) + lock_project(locked_source, python=Path(sys.executable), cache=tmp_path / "cache", env=build_tools, explicit=True) before = (locked_source / "uv.lock").read_bytes() out = tmp_path / "git-requirements.txt" - export_requirements(locked_source, out, cache=tmp_path / "cache", env=build_tools) + export_requirements(locked_source, out, cache=tmp_path / "cache", env=build_tools, explicit=True) requirement = Requirement(out.read_text(encoding="utf-8").strip()) assert requirement.name == "git-dep" assert requirement.url == url @@ -149,8 +149,7 @@ def test_requirements_build_installs_offline_markers_and_seals_only_build_pth(tm executable = build_requirements_environment( ["app-dep==1.0; python_version >= '3'", "missing-dep==1.0; python_version < '2'"], out=out, python=Path(sys.executable), wheelhouse=wheels, cache=tmp_path / "cache", - env=env, offline=True, sealed=sealed, - ) + env=env, offline=True, sealed=sealed, explicit=True) result = json.loads(_run( [str(executable), "-I", "-c", "import sys, json, app_dep, leaf_dep, importlib.util; " "print(json.dumps([app_dep.__version__, leaf_dep.__version__, sys.base_prefix, " @@ -174,7 +173,7 @@ def test_failed_requirement_build_removes_only_its_candidate(tmp_path, build_too wheel = _wheel(wheels, "app_dep") previous = tmp_path / "previous" executable = build_requirements_environment(["app-dep==1.0"], out=previous, wheelhouse=wheels, - env=build_tools, cache=tmp_path / "cache", offline=True) + env=build_tools, cache=tmp_path / "cache", offline=True, explicit=True) previous_cfg = (previous / "pyvenv.cfg").read_bytes() python = Path(sys.executable) if failure == "create": @@ -191,11 +190,11 @@ def test_failed_requirement_build_removes_only_its_candidate(tmp_path, build_too out = tmp_path / "candidate" with pytest.raises(InstallError, match="dependency validation" if failure == "check" else None): build_requirements_environment(["app-dep==1.0"], out=out, python=python, wheelhouse=wheels, - env=build_tools, cache=tmp_path / "cold-cache", offline=True) + env=build_tools, cache=tmp_path / "cold-cache", offline=True, explicit=True) assert not out.exists() with pytest.raises(FileExistsError): build_requirements_environment(["app-dep==1.0"], out=previous, wheelhouse=wheels, - env=build_tools, cache=tmp_path / "cache", offline=True) + env=build_tools, cache=tmp_path / "cache", offline=True, explicit=True) assert (previous / "pyvenv.cfg").read_bytes() == previous_cfg assert _run([str(executable), "-I", "-c", "import app_dep; print(app_dep.__version__)"], cwd=tmp_path, env=build_tools) == "1.0" @@ -236,18 +235,56 @@ def test_wheelhouse_cannot_fall_back_to_index_or_build_source(tmp_path, build_to out = tmp_path / "candidate" with pytest.raises(InstallError): build_requirements_environment(["leaf-dep==1.0"], out=out, wheelhouse=wheels, - env=env, cache=tmp_path / "cache", offline=True) + env=env, cache=tmp_path / "cache", offline=True, explicit=True) assert not out.exists() # Both alternate sources really work when there is no wheelhouse restriction. for name, settings in (("index", env), ("source", dict(build_tools, UV_NO_INDEX="1", UV_FIND_LINKS=str(wheels)))): executable = build_requirements_environment( ["leaf-dep==1.0"], out=tmp_path / f"from-{name}", env=settings, - cache=tmp_path / f"{name}-cache", offline=True, - ) + cache=tmp_path / f"{name}-cache", offline=True, explicit=True) assert _run([str(executable), "-I", "-c", "import leaf_dep; print(leaf_dep.__version__)"], cwd=tmp_path, env=build_tools) == "1.0" +@pytest.mark.parametrize("operation", ["check", "export", "build"]) +def test_ready_tools_do_not_bypass_disabled_lazy_operations(locked_source, tmp_path, build_tools, monkeypatch, operation): + import importlib + from pm import build_requirements_environment, check_project_lock, export_requirements + + monkeypatch.setattr(importlib.import_module("pm.ensure"), "lazy_installs_allowed", lambda: False) + out = tmp_path / "blocked-output" + before = {p.name: p.read_bytes() for p in locked_source.iterdir() if p.is_file()} + actions = { + "check": lambda: check_project_lock(locked_source, env=build_tools, cache=tmp_path / "cache"), + "export": lambda: export_requirements(locked_source, out, env=build_tools, cache=tmp_path / "cache"), + "build": lambda: build_requirements_environment(["base-dep==1.0"], out=out, env=build_tools, + cache=tmp_path / "cache", offline=True), + } + with pytest.raises(InstallError, match="lazy installs are disabled"): + actions[operation]() + assert not out.exists() + assert {p.name: p.read_bytes() for p in locked_source.iterdir() if p.is_file()} == before + # Passive lock validation is safe with already-ready tools and no network. + check_project_lock(locked_source, env=build_tools, cache=tmp_path / "cache", offline=True) + + +def test_prune_cache_does_not_acquire_a_missing_toolchain(tmp_path, monkeypatch): + import importlib + import pm.paths + from pm import prune_cache + + monkeypatch.setattr("pm.client.is_runtime", lambda: True) + store = tmp_path / "empty-store" + monkeypatch.setattr(pm.paths, "store_root", lambda: store) + monkeypatch.setattr(pm.paths, "writable_store_root", lambda: store) + monkeypatch.setattr(pm.paths, "facts_path", lambda: store / "facts.json") + monkeypatch.setattr(importlib.import_module("pm.ensure"), "ensure", + lambda *args, **kwargs: pytest.fail("cache pruning must not acquire tools")) + with pytest.raises(InstallError, match="pinned toolchain is unavailable"): + prune_cache(tmp_path / "cache") + assert not store.exists() + + @pytest.mark.parametrize("ci", [False, True]) def test_prune_cache_preserves_downloaded_wheels_unless_ci(tmp_path, build_tools, ci): from functools import partial @@ -265,8 +302,7 @@ def test_prune_cache_preserves_downloaded_wheels_unless_ci(tmp_path, build_tools requirement = f"cached-dep @ http://127.0.0.1:{server.server_port}/{wheel.name}" try: executable = build_requirements_environment( - [requirement], out=tmp_path / "first", cache=cache, env=build_tools, - ) + [requirement], out=tmp_path / "first", cache=cache, env=build_tools, explicit=True) finally: server.shutdown() server.server_close() @@ -277,9 +313,9 @@ def test_prune_cache_preserves_downloaded_wheels_unless_ci(tmp_path, build_tools out = tmp_path / "second" if ci: with pytest.raises(InstallError): - build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True) + build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True, explicit=True) assert not out.exists() else: - second = build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True) + second = build_requirements_environment([requirement], out=out, cache=cache, env=build_tools, offline=True, explicit=True) assert _run([str(second), "-I", "-c", "import cached_dep; print(cached_dep.__version__)"], cwd=tmp_path, env=build_tools) == "1.0"