CelerisLab/tests/validation/source_provenance.py

"""Record runtime checkout identity separately from the reviewed source origin."""

import subprocess


def checkout_provenance(repo):
    """Require a Git checkout; never label changed-source runs as historical evidence."""
    revision = subprocess.check_output(
        ["git", "-C", str(repo), "rev-parse", "HEAD"], text=True
    ).strip()
    dirty = subprocess.check_output(
        ["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=normal"],
        text=True,
    )
    return dict(checkout_revision=revision, checkout_dirty=bool(dirty))


def final_source_check(recorded, recheck):
    """Re-derive the source identity after a run and compare it with the recorded one.

    Returns receipt fields: source_recheck "UNCHANGED"; "CHANGED" with source_after; or
    "RECHECK_FAILED" (source_after None) with the error. Callers mark a receipt complete
    only for "UNCHANGED"; anything else invalidates the run.
    """
    try:
        after = recheck()
    except Exception as error:  # a failed recheck cannot confirm the source: invalid too
        return dict(source_recheck="RECHECK_FAILED", source_after=None,
                    source_recheck_error=f"{type(error).__name__}: {error}")
    if after == recorded:
        return dict(source_recheck="UNCHANGED")
    return dict(source_recheck="CHANGED", source_after=after, source_recheck_error=None)


def reject_output_inside(output, repo):
    """Refuse run outputs inside the checkout before any work: they would dirty it."""
    from pathlib import Path
    output, repo = Path(output).resolve(), Path(repo).resolve()
    if output == repo or repo in output.parents:
        raise SystemExit(f"--output must be outside the source checkout {repo}")


class HistoryUnavailable(RuntimeError):
    """The clone cannot show the recorded commit (no git, shallow clone, installed copy)."""


def historical_sha256(repo, commit, paths):
    """SHA-256 of each path as stored in `commit`, read with git, never from the working tree.

    Raises HistoryUnavailable when git or the commit object is missing, so callers can
    report an unchecked integrity record explicitly instead of passing or failing silently.
    """
    import hashlib
    git = ["git", "-C", str(repo)]
    try:
        subprocess.run([*git, "cat-file", "-e", f"{commit}^{{commit}}"], check=True,
                       capture_output=True)
    except (OSError, subprocess.CalledProcessError) as error:
        raise HistoryUnavailable(f"commit {commit} is not available in {repo}") from error
    digests = {}
    for path in paths:
        listed = subprocess.run([*git, "ls-tree", "--name-only", commit, "--", path],
                                capture_output=True, text=True)
        if listed.returncode:
            raise HistoryUnavailable(f"tree of {commit} is not available in {repo}")
        if listed.stdout.strip() != path:
            raise ValueError(f"{path} is not part of {commit}")
        shown = subprocess.run([*git, "show", f"{commit}:{path}"], capture_output=True)
        if shown.returncode:  # e.g. a blobless partial clone without network access
            raise HistoryUnavailable(f"content of {path} at {commit} is not available in {repo}")
        digests[path] = hashlib.sha256(shown.stdout).hexdigest()
    return digests


def verify_manifest(repo, commit, digests):
    """Raise RuntimeError naming the first path whose recorded digest differs at `commit`."""
    actual = historical_sha256(repo, commit, digests)
    for path, digest in digests.items():
        if actual[path] != digest:
            raise RuntimeError(f"Historical manifest mismatch at {commit}: {path}")
    return actual


def historical_tree(repo, commit, prefix="src"):
    """Tracked file names under `prefix` at `commit` (HistoryUnavailable if missing)."""
    historical_sha256(repo, commit, [])
    return subprocess.check_output(["git", "-C", str(repo), "ls-tree", "-r", "--name-only",
                                    commit, "--", prefix], text=True).splitlines()


def historical_files(repo, commit, paths, dest):
    """Write each path as stored in `commit` under `dest` (HistoryUnavailable if missing)."""
    from pathlib import Path
    historical_sha256(repo, commit, paths)  # HistoryUnavailable / ValueError before writing
    for path in paths:
        shown = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"],
                               capture_output=True)
        if shown.returncode:
            raise HistoryUnavailable(f"cannot read {commit}:{path} (blob not in this clone)")
        target = Path(dest) / path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(shown.stdout)


def history_unavailable(testcase, error):
    """Skip with the reason, or fail when CELERIS_REQUIRE_HISTORY=1 (release gating)."""
    import os
    message = f"historical integrity NOT checked: {error}"
    if os.environ.get("CELERIS_REQUIRE_HISTORY") == "1":
        raise AssertionError(message)
    testcase.skipTest(message)