diff --git a/scripts/ci/materialize_base_python_requirements.py b/scripts/ci/materialize_base_python_requirements.py index 28ce5364f..8158372df 100644 --- a/scripts/ci/materialize_base_python_requirements.py +++ b/scripts/ci/materialize_base_python_requirements.py @@ -8,11 +8,14 @@ import json import pathlib import re +import shutil import subprocess import sys +import tempfile SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") +UV_EXPORT_TIMEOUT_SECONDS = 120 def _is_candidate_lock_name(name: str) -> bool: @@ -77,6 +80,84 @@ def _git(repo_root: pathlib.Path, *args: str) -> bytes: return completed.stdout +def _run_uv_export( + work_dir: pathlib.Path, + uv_path: str, + *, + timeout: float = UV_EXPORT_TIMEOUT_SECONDS, +) -> subprocess.CompletedProcess[bytes]: + """Run ``uv export`` for a reconstructed base project and return the result. + + ``--frozen`` forbids lock mutation and ``--offline`` forbids network access, + so the export is a pure function of the already-trusted base ``uv.lock`` and + ``pyproject.toml``; ``--no-emit-project``/``--no-editable`` drop the project + itself (installed via ``PYTHONPATH`` in the sandbox) and keep only its + hash-pinned dependency closure. + """ + return subprocess.run( + [ + uv_path, + "export", + "--frozen", + "--offline", + "--no-emit-project", + "--no-editable", + "--format", + "requirements-txt", + ], + cwd=str(work_dir), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + ) + + +def _export_uv_lock( + repo_root: pathlib.Path, base_sha: str, lock_path: str +) -> bytes | None: + """Export a base ``uv.lock`` to a hash-pinned requirements closure, or ``None``. + + ``uv.lock`` is not a pip-installable format, so a uv-managed repository + materializes no dependencies and its offline coverage run fails at import. + When ``uv`` is available, reconstruct the exact base ``uv.lock`` and its + sibling ``pyproject.toml`` in an isolated temporary directory and run + ``uv export --frozen`` to produce a fully hash-pinned closure the trusted + installer can consume like any other lock. Both inputs are read only from + the validated base commit, so no PR-mutable content reaches ``uv``. Return + ``None`` — degrading to the prior no-uv behavior — when ``uv`` is absent, + the sibling ``pyproject.toml`` is missing at the base commit, the export + fails, or its output is not fully hash-pinned, so this can never break an + otherwise-working build. + """ + uv_path = shutil.which("uv") + if uv_path is None: + return None + project_dir = pathlib.PurePosixPath(lock_path).parent + pyproject_path = ( + "pyproject.toml" + if str(project_dir) == "." + else f"{project_dir}/pyproject.toml" + ) + try: + lock_content = _git(repo_root, "show", f"{base_sha}:{lock_path}") + pyproject_content = _git(repo_root, "show", f"{base_sha}:{pyproject_path}") + except RuntimeError: + return None + with tempfile.TemporaryDirectory() as work_dir: + work_path = pathlib.Path(work_dir) + (work_path / "uv.lock").write_bytes(lock_content) + (work_path / "pyproject.toml").write_bytes(pyproject_content) + try: + completed = _run_uv_export(work_path, uv_path) + except (OSError, subprocess.TimeoutExpired): + return None + if completed.returncode != 0: + return None + exported = completed.stdout + return exported if _is_hash_pinned(exported) else None + + def base_hash_locks(repo_root: pathlib.Path, base_sha: str) -> list[tuple[str, bytes]]: """Return regular hash-lock blobs from the exact validated base commit.""" if not SHA_RE.fullmatch(base_sha): @@ -103,13 +184,16 @@ def base_hash_locks(repo_root: pathlib.Path, base_sha: str) -> list[tuple[str, b or not mode.startswith("100") or candidate.is_absolute() or ".." in candidate.parts - or not _is_candidate_lock_name(candidate.name) ): continue - content = _git(repo_root, "show", f"{base_sha}:{path}") - if not _is_hash_pinned(content): - continue - locks.append((path, content)) + if _is_candidate_lock_name(candidate.name): + content = _git(repo_root, "show", f"{base_sha}:{path}") + if _is_hash_pinned(content): + locks.append((path, content)) + elif candidate.name == "uv.lock": + exported = _export_uv_lock(repo_root, base_sha, path) + if exported is not None: + locks.append((path, exported)) return sorted(locks, key=lambda item: item[0]) diff --git a/tests/test_materialize_base_python_requirements.py b/tests/test_materialize_base_python_requirements.py index 21b984f4d..41b86b261 100644 --- a/tests/test_materialize_base_python_requirements.py +++ b/tests/test_materialize_base_python_requirements.py @@ -323,3 +323,170 @@ def test_script_entrypoint_exits_through_main( runpy.run_path(str(module_path), run_name="__main__") assert raised.value.code == 1 + + +def test_skips_non_blob_tree_entries( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Submodule/gitlink (non-blob) tree entries are skipped, never materialized.""" + blob = b"pinned==1 --hash=sha256:" + b"a" * 64 + b"\n" + tree = ( + b"160000 commit " + b"0" * 40 + b"\tvendored-submodule\0" + b"100644 blob " + b"1" * 40 + b"\trequirements.txt\0" + ) + + def fake_git(_repo_root: Path, *args: str) -> bytes: + if args[0] == "ls-tree": + return tree + if args[0] == "show": + return blob + raise AssertionError(args) + + monkeypatch.setattr(materializer, "_git", fake_git) + + assert materializer.base_hash_locks(tmp_path, "a" * 40) == [ + ("requirements.txt", blob) + ] + + +def _uv_repo(tmp_path: Path, *, with_pyproject: bool, lock_dir: str = "") -> tuple[Path, str]: + """Init a fixture repo with a uv.lock (and optional pyproject.toml) at lock_dir.""" + repo = tmp_path / "repo" + repo.mkdir() + git(repo, "init") + git(repo, "config", "user.name", "Test") + git(repo, "config", "user.email", "test@example.invalid") + base = repo / lock_dir if lock_dir else repo + base.mkdir(parents=True, exist_ok=True) + (base / "uv.lock").write_text("version = 1\n", encoding="utf-8") + if with_pyproject: + (base / "pyproject.toml").write_text( + "[project]\nname = 'demo'\nversion = '0'\n", encoding="utf-8" + ) + git(repo, "add", ".") + git(repo, "commit", "-m", "base") + return repo, git(repo, "rev-parse", "HEAD") + + +def _export(returncode: int, stdout: bytes) -> subprocess.CompletedProcess[bytes]: + """Build a fake ``uv export`` completed-process result.""" + return subprocess.CompletedProcess(["uv", "export"], returncode, stdout, b"") + + +def test_uv_lock_is_exported_to_a_hash_pinned_lock( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A base uv.lock is exported via uv into a materialized hash-pinned closure.""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=True) + monkeypatch.setattr(materializer.shutil, "which", lambda _name: "/usr/bin/uv") + hashed = b"demo-dep==1 --hash=sha256:" + b"a" * 64 + b"\n" + monkeypatch.setattr( + materializer, + "_run_uv_export", + lambda _work, _uv_path: _export(0, hashed), + ) + + output = tmp_path / "output" + manifest = materializer.materialize(repo, base_sha, output) + + assert manifest == [{"file": "requirements-000.txt", "source": "uv.lock"}] + assert (output / "requirements-000.txt").read_bytes() == hashed + + +def test_uv_lock_skipped_when_uv_is_unavailable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Without the uv exporter, a uv.lock-only repo materializes nothing (no regression).""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=True) + monkeypatch.setattr(materializer.shutil, "which", lambda _name: None) + + assert materializer.materialize(repo, base_sha, tmp_path / "output") == [] + + +def test_uv_lock_skipped_when_pyproject_is_absent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A uv.lock without a sibling pyproject.toml at base (in a subdir) cannot be exported.""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=False, lock_dir="service") + monkeypatch.setattr(materializer.shutil, "which", lambda _name: "/usr/bin/uv") + + assert materializer.materialize(repo, base_sha, tmp_path / "output") == [] + + +def test_uv_lock_skipped_when_export_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A non-zero uv export (e.g. a stale lock) is skipped, never materialized.""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=True) + monkeypatch.setattr(materializer.shutil, "which", lambda _name: "/usr/bin/uv") + monkeypatch.setattr( + materializer, + "_run_uv_export", + lambda _work, _uv_path: _export(1, b""), + ) + + assert materializer.materialize(repo, base_sha, tmp_path / "output") == [] + + +def test_uv_lock_skipped_when_export_is_not_hash_pinned( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A uv export that somehow lacks hashes is rejected by the hash-pin guard.""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=True) + monkeypatch.setattr(materializer.shutil, "which", lambda _name: "/usr/bin/uv") + monkeypatch.setattr( + materializer, + "_run_uv_export", + lambda _work, _uv_path: _export(0, b"unpinned==1\n"), + ) + + assert materializer.materialize(repo, base_sha, tmp_path / "output") == [] + + +def test_run_uv_export_invokes_uv_with_frozen_offline_flags( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The uv export helper runs uv with frozen, project-excluding, offline flags.""" + captured: dict[str, object] = {} + + def fake_run(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: + captured["argv"] = argv + captured["cwd"] = kwargs.get("cwd") + captured["timeout"] = kwargs.get("timeout") + return subprocess.CompletedProcess(argv, 0, b"out", b"") + + monkeypatch.setattr(materializer.subprocess, "run", fake_run) + + result = materializer._run_uv_export(tmp_path, "/usr/bin/uv") + + assert result.stdout == b"out" + assert captured["argv"][:3] == ["/usr/bin/uv", "export", "--frozen"] + assert "--offline" in captured["argv"] + assert "--no-emit-project" in captured["argv"] + assert "--no-editable" in captured["argv"] + assert captured["cwd"] == str(tmp_path) + assert captured["timeout"] == materializer.UV_EXPORT_TIMEOUT_SECONDS + + +@pytest.mark.parametrize( + "export_error", + [ + FileNotFoundError("uv disappeared"), + subprocess.TimeoutExpired(["/usr/bin/uv", "export"], timeout=120), + ], +) +def test_uv_export_process_failures_fall_back_to_no_lock( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + export_error: OSError | subprocess.TimeoutExpired, +) -> None: + """A missing or hung uv process preserves the documented best-effort fallback.""" + repo, base_sha = _uv_repo(tmp_path, with_pyproject=True) + monkeypatch.setattr(materializer.shutil, "which", lambda _name: "/usr/bin/uv") + + def fail_export(_work: Path, _uv_path: str) -> None: + raise export_error + + monkeypatch.setattr(materializer, "_run_uv_export", fail_export) + + assert materializer.materialize(repo, base_sha, tmp_path / "output") == []