CelerisLab/tests/unit/test_validation_source_provenance.py

"""Explicit experimental repin and historical-source provenance contracts."""

import hashlib
import json
from pathlib import Path
import subprocess
import sys

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "tests/validation"))
import pytest
from source_provenance import (HistoryUnavailable, checkout_provenance, historical_sha256,
                               historical_tree, history_unavailable, verify_manifest)
import unittest
import re100_helpers
import run_jet_circle_cv

ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"


def test_manifests_match_their_recorded_source_and_preserve_origin():
    """Historical evidence integrity: each manifest describes the tree of its own commit."""
    manifests = []
    for group, hashes_key, revision_key in (
        ("dfg_re100", "production_sha256", "commit"),
        ("jet_circle_cv", "files", "revision"),
    ):
        manifest = json.loads(
            (
                ROOT / "tests/validation/data" / group / "source_manifest.json"
            ).read_text()
        )
        # Structural checks need no history and always run.
        assert manifest[revision_key] == ORIGIN
        assert manifest["source_origin_commits"][-1] == ORIGIN
        assert (
            manifest["historical_checkpoints"]["production"]
            == "d79e3f6f0a424f5bcc3a1a6d610313e49daf97a2"
        )
        assert "not inherited from d79" in manifest["experiment"]
        manifests.append(manifest[hashes_key])
    try:
        names = historical_tree(ROOT, ORIGIN)
        for digests in manifests:
            assert set(digests) == set(names)
            verify_manifest(ROOT, ORIGIN, digests)
    except HistoryUnavailable as error:
        history_unavailable(unittest.TestCase(), error)
    assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN


def test_runtime_checkout_identity_is_separate_from_source_origin(tmp_path):
    subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
    (tmp_path / "test.txt").write_text("tracked\n")
    subprocess.run(["git", "-C", str(tmp_path), "add", "test.txt"], check=True)
    subprocess.run(
        [
            "git",
            "-C",
            str(tmp_path),
            "-c",
            "user.name=Test",
            "-c",
            "user.email=test@example.invalid",
            "commit",
            "-qm",
            "fixture",
        ],
        check=True,
    )
    expected = subprocess.check_output(
        ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], text=True
    ).strip()
    assert checkout_provenance(tmp_path) == dict(
        checkout_revision=expected, checkout_dirty=False
    )
    (tmp_path / "test.txt").write_text("changed\n")
    assert checkout_provenance(tmp_path) == dict(
        checkout_revision=expected, checkout_dirty=True
    )


def test_historical_check_rejects_tampering_and_reports_missing_history(tmp_path):
    with pytest.raises(HistoryUnavailable):
        historical_sha256(tmp_path, ORIGIN, ["src/x.py"])  # not a repository
    with pytest.raises(HistoryUnavailable):
        historical_sha256(ROOT, "0" * 40, ["src/x.py"])  # commit absent
    manifest = json.loads(
        (ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text())
    digests = dict(manifest["production_sha256"])
    name = sorted(digests)[0]
    tampered = dict(digests, **{name: ("0" if digests[name][0] != "0" else "1") + digests[name][1:]})
    try:
        historical_tree(ROOT, ORIGIN)
    except HistoryUnavailable as error:
        history_unavailable(unittest.TestCase(), error)
    with pytest.raises(RuntimeError, match="Historical manifest mismatch"):
        verify_manifest(ROOT, ORIGIN, tampered)  # one flipped digest is caught
    with pytest.raises(ValueError, match="not part of"):
        historical_sha256(ROOT, ORIGIN, ["src/CelerisLab/does_not_exist.py"])


def test_historical_files_reports_unreadable_blob_as_history_unavailable(tmp_path):
    """A blob missing from the clone is HistoryUnavailable, not a raw CalledProcessError."""
    import source_provenance
    from unittest.mock import patch

    missing = subprocess.CompletedProcess(["git", "show"], 128, b"", b"fatal: missing blob")
    with patch.object(source_provenance, "historical_sha256", return_value={"src/a.py": "0" * 64}), \
         patch.object(source_provenance.subprocess, "run", return_value=missing):
        with pytest.raises(HistoryUnavailable, match="blob not in this clone"):
            source_provenance.historical_files(ROOT, ORIGIN, ["src/a.py"], tmp_path)
    assert not (tmp_path / "src/a.py").exists()


def test_final_source_check_unchanged_changed_and_failed_recheck():
    from source_provenance import final_source_check

    recorded = dict(revision="a" * 40, dirty=False)
    assert final_source_check(recorded, lambda: dict(recorded)) == dict(source_recheck="UNCHANGED")
    changed = final_source_check(recorded, lambda: dict(recorded, dirty=True))
    assert changed == dict(source_recheck="CHANGED", source_after=dict(recorded, dirty=True),
                           source_recheck_error=None)

    def fails():
        raise RuntimeError("Source hash mismatch: src/x.py")

    failed = final_source_check(recorded, fails)
    assert (failed["source_recheck"], failed["source_after"]) == ("RECHECK_FAILED", None)
    assert "Source hash mismatch" in failed["source_recheck_error"]


def test_reject_output_inside_checkout(tmp_path):
    from source_provenance import reject_output_inside

    for output in (ROOT, ROOT / "out", ROOT / "src" / "x.json"):
        with pytest.raises(SystemExit, match="outside the source checkout"):
            reject_output_inside(output, ROOT)
    reject_output_inside(tmp_path / "out", ROOT)