Frank14f/CelerisLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: tighten historical-pin checks per review of 941eb28b

Browse Source 
- test_refined_dfg::test_source_pin runs its mismatch negative controls
  first (now including window mode) so missing history cannot skip them.
- test_re100: renamed to test_manifest_matches_its_pinned_commit, plus an
  explicit record that today's src differs from the pin.
- Structural manifest checks run before any history-dependent check.
- Real tamper test: a flipped digest raises via verify_manifest, and a
  path absent at the commit raises ValueError.
- A blobless clone (tree present, blob missing) is reported as unavailable
  history, not as a missing path.
- All history skips go through history_unavailable(): fatal when
  CELERIS_REQUIRE_HISTORY=1.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-10-20 18:28:09 +00:00
co-authored by Claude Opus 5.5
parent f02355a2b7
commit 8064834c37
6 changed files with 72 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_jet_circle_cv.py
+2 -2
View File
@@ -168,11 +168,11 @@ class Host(unittest.TestCase):
def test_source_pin(self):
"""Historical integrity: the manifest matches its pinned commit, read with git."""
from run_jet_circle_cv import verify_historical
from source_provenance import HistoryUnavailable
from source_provenance import HistoryUnavailable, history_unavailable
try:
verify_historical(REPO)
except HistoryUnavailable as error:
self.skipTest(f"historical integrity NOT checked: {error}")
history_unavailable(self, error)
def test_pinned_mode_still_rejects_changed_source(self):
"""Negative control: the frozen-source guard refuses a tree that differs from PIN."""
tests/unit/test_re100.py
+20 -6
View File
@@ -62,17 +62,31 @@ def test_periodic_synthetic():
assert not p.assess(rows()[:100])["periodic"]
def test_production_unmodified():
def test_manifest_matches_its_pinned_commit():
"""Historical integrity only: the manifest describes the h.PIN tree (read with git)."""
m = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
)
assert m["commit"] == h.PIN
from source_provenance import HistoryUnavailable, historical_sha256
try: # the manifest certifies its own commit, not today's working tree
actual = historical_sha256(ROOT, m["commit"], m["production_sha256"])
import unittest
from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
try:
verify_manifest(ROOT, m["commit"], m["production_sha256"])
except HistoryUnavailable as error:
pytest.skip(f"historical integrity NOT checked: {error}")
assert actual == m["production_sha256"]
history_unavailable(unittest.TestCase(), error)
def test_current_src_is_not_the_pinned_qualification():
"""Explicit record: today's src differs from h.PIN, so its qualification does not carry over."""
m = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
)
changed = sorted(n for n, d in m["production_sha256"].items()
if not (ROOT / n).exists()
or hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != d)
if not changed:
pytest.skip("current src equals the pinned tree")
assert changed # e.g. src/CelerisLab/lbm/channel.py since 47b05bc
def test_reference():
tests/unit/test_refined_dfg.py
+8 -9
View File
@@ -28,18 +28,17 @@ def test_source_pin():
mrt_history = historical_manifest("MRT")
assert mrt_history["commit"].startswith("3ea131e")
assert historical_manifest("MRT", "window")["commit"].startswith("a233c90")
from source_provenance import HistoryUnavailable, historical_sha256
# Negative controls first, so missing history can never hide them: every pinned
# mode refuses a working tree that differs from its pin.
for collision, mode in (("MRT", "last_base_step"), ("SRT", "last_base_step"), ("MRT", "window")):
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, collision, mode)
from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
for manifest in (historical, mrt_history, historical_manifest("MRT", "window")):
try: # historical integrity: each manifest matches the tree of its own commit
actual = historical_sha256(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"])
verify_manifest(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"])
except HistoryUnavailable as error:
unittest.TestCase().skipTest(f"historical integrity NOT checked: {error}")
assert actual == manifest["sha256"], manifest["commit"]
# Negative controls: the pinned modes refuse a working tree that differs from their pin.
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "MRT")
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "SRT")
history_unavailable(unittest.TestCase(), error)
def test_matched_units_and_planes():
tests/unit/test_validation_source_provenance.py
+21 -11
View File
@@ -10,7 +10,8 @@ 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)
historical_tree, history_unavailable, verify_manifest)
import unittest
import re100_helpers
import run_jet_circle_cv
@@ -19,10 +20,7 @@ ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"
def test_manifests_match_their_recorded_source_and_preserve_origin():
"""Historical evidence integrity: each manifest describes the tree of its own commit."""
try:
names = historical_tree(ROOT, ORIGIN)
except HistoryUnavailable as error:
pytest.skip(f"historical integrity NOT checked: {error}")
manifests = []
for group, hashes_key, revision_key in (
("dfg_re100", "production_sha256", "commit"),
("jet_circle_cv", "files", "revision"),
@@ -32,15 +30,22 @@ def test_manifests_match_their_recorded_source_and_preserve_origin():
ROOT / "tests/validation/data" / group / "source_manifest.json"
).read_text()
)
# Structural checks need no history and always run.
assert manifest[revision_key] == ORIGIN
assert set(manifest[hashes_key]) == set(names)
assert manifest["source_origin_commits"][-1] == ORIGIN
assert (
manifest["historical_checkpoints"]["production"]
== "d79e3f6f0a424f5bcc3a1a6d610313e49daf97a2"
)
assert "not inherited from d79" in manifest["experiment"]
assert historical_sha256(ROOT, ORIGIN, manifest[hashes_key]) == manifest[hashes_key]
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
@@ -82,9 +87,14 @@ def test_historical_check_rejects_tampering_and_reports_missing_history(tmp_path
historical_sha256(ROOT, "0" * 40, ["src/x.py"]) # commit absent
manifest = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text())
name = sorted(manifest["production_sha256"])[0]
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:
actual = historical_sha256(ROOT, ORIGIN, [name])
historical_tree(ROOT, ORIGIN)
except HistoryUnavailable as error:
pytest.skip(f"historical integrity NOT checked: {error}")
assert actual != {name: "0" * 64} # a tampered digest would not match
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"])
tests/validation/run_jet_circle_cv.py
+2 -3
View File
@@ -223,12 +223,11 @@ def verify(repo):
def verify_historical(repo):
"""The pinned manifest describes the PIN tree (read with git, not the working tree)."""
from source_provenance import historical_sha256
from source_provenance import verify_manifest
manifest = json.loads((HERE / "data/jet_circle_cv/source_manifest.json").read_text())
if manifest["revision"] != PIN:
raise RuntimeError("Wrong pinned revision")
if historical_sha256(repo, PIN, manifest["files"]) != manifest["files"]:
raise RuntimeError("Historical manifest does not match its pinned revision")
verify_manifest(repo, PIN, manifest["files"])
return manifest
tests/validation/source_provenance.py
+19 -6
View File
@@ -34,15 +34,28 @@ def historical_sha256(repo, commit, paths):
raise HistoryUnavailable(f"commit {commit} is not available in {repo}") from error
digests = {}
for path in paths:
try:
blob = subprocess.run([*git, "show", f"{commit}:{path}"], check=True,
capture_output=True).stdout
except subprocess.CalledProcessError as error:
raise ValueError(f"{path} is not part of {commit}") from error
digests[path] = hashlib.sha256(blob).hexdigest()
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, [])