test: run history-independent identity checks before the history lookup

This commit is contained in:
2026-10-21 16:49:41 +08:00
parent d437faa501
commit fce24a64e8
4 changed files with 22 additions and 15 deletions
+2 -2
View File
@@ -171,10 +171,10 @@ class Host(unittest.TestCase):
from tempfile import TemporaryDirectory
from historical_tree import historical_tree
with TemporaryDirectory() as directory:
verify(historical_tree(PIN, directory))
with self.assertRaisesRegex(RuntimeError, "Source hash mismatch"):
verify(REPO)
with TemporaryDirectory() as directory:
verify(historical_tree(PIN, directory))
if __name__ == "__main__":
+5 -2
View File
@@ -48,8 +48,11 @@ class GuardTests(unittest.TestCase):
def test_current_corrected_pins(self):
with TemporaryDirectory() as directory:
root = historical_tree(A135, directory)
with patch("run_uniform_les_flow.subprocess.check_output", return_value=A135 + "\n"):
self.assertEqual(source_guard(root, "corrected", "a135252")["base_commit"], A135)
with patch("run_uniform_les_flow.subprocess.check_output", return_value="historical-tree\n"):
identity = source_guard(root, "corrected", "a135252")
# base_commit comes from the pins file; the mock only feeds checkout_commit.
self.assertEqual(identity["base_commit"], A135)
self.assertEqual(identity["checkout_commit"], "historical-tree")
def test_corrected_and_tamper_guard(self):
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
+6 -6
View File
@@ -20,6 +20,12 @@ def test_source_pin():
assert historical["commit"].startswith("cd3a657")
mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
assert mrt_history["commit"].startswith("3ea131e")
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
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")
with TemporaryDirectory() as directory:
for collision, mode, prefix in (
("SRT", "last_base_step", "cd3a657"),
@@ -32,12 +38,6 @@ def test_source_pin():
assert commit.startswith(prefix)
root = historical_tree(commit, Path(directory) / commit)
assert r.verify_source(root, collision, mode)["commit"] == commit
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
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")
def test_matched_units_and_planes():
@@ -19,8 +19,8 @@ ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"
def test_manifests_cover_current_source_and_preserve_origin(tmp_path):
names = tracked(ORIGIN)
origin = historical_tree(ORIGIN, tmp_path)
assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
manifests = {}
for group, hashes_key, revision_key in (
("dfg_re100", "production_sha256", "commit"),
("jet_circle_cv", "files", "revision"),
@@ -31,18 +31,22 @@ def test_manifests_cover_current_source_and_preserve_origin(tmp_path):
).read_text()
)
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"]
for name, expected in manifest[hashes_key].items():
manifests[group] = manifest[hashes_key]
# History-dependent checks last, so a missing commit masks nothing above.
names = tracked(ORIGIN)
origin = historical_tree(ORIGIN, tmp_path)
for group, hashes in manifests.items():
assert set(hashes) == set(names), group
for name, expected in hashes.items():
assert (
hashlib.sha256((origin / name).read_bytes()).hexdigest() == expected
), name
assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
def test_runtime_checkout_identity_is_separate_from_source_origin(tmp_path):