test: check historical source manifests against their recorded commits
Manifests and receipts unchanged. Each of the 8 group-A identity tests now extracts its manifest's recorded commit and verifies hashes there instead of against the current checkout. Tamper assertions and runner source refusals are kept; missing recorded history is an explicit failure, not a skip or repin.
This commit is contained in:
1 parent
26e49451ec
commit
d437faa501
6 files changed
+109
-25
No files matched your search
@@ -15,6 +15,10 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
|
||||
from run_uniform_les_flow import source_guard, decode_macros, HERE, HEADER
|
||||
|
||||
ROOT = HERE.parents[1]
|
||||
sys.path.insert(0, str(ROOT / "tests/unit"))
|
||||
from historical_tree import historical_tree, tracked
|
||||
|
||||
A135 = "a135252ae78b47b57d7e77d19001700b3bedac8b"
|
||||
|
||||
|
||||
class GuardTests(unittest.TestCase):
|
||||
@@ -42,19 +46,20 @@ class GuardTests(unittest.TestCase):
|
||||
decode_macros(invalid, w, cx, cy)
|
||||
|
||||
def test_current_corrected_pins(self):
|
||||
self.assertEqual(
|
||||
source_guard(ROOT, "corrected", "a135252")["base_commit"],
|
||||
"a135252ae78b47b57d7e77d19001700b3bedac8b",
|
||||
)
|
||||
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)
|
||||
|
||||
def test_corrected_and_tamper_guard(self):
|
||||
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
|
||||
with TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
origin = historical_tree(A135, Path(directory) / "origin")
|
||||
root = Path(directory) / "fixture"
|
||||
for name in pins["sources"]:
|
||||
dest = root / name
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(ROOT / name, dest)
|
||||
shutil.copyfile(origin / name, dest)
|
||||
with patch(
|
||||
"run_uniform_les_flow.subprocess.check_output",
|
||||
return_value="isolated-fixture\n",
|
||||
@@ -87,23 +92,26 @@ class GuardTests(unittest.TestCase):
|
||||
|
||||
def test_current_manifest_pins_actual_corrected_tree(self):
|
||||
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
|
||||
tracked = subprocess.check_output(["git", "-C", str(ROOT), "ls-files", "src"], text=True)
|
||||
self.assertEqual(set(pins["sources"]), set(tracked.splitlines()))
|
||||
for name, expected in pins["sources"].items():
|
||||
self.assertEqual(hashlib.sha256((ROOT / name).read_bytes()).hexdigest(), expected)
|
||||
self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"])
|
||||
self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"])
|
||||
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
|
||||
source_guard(ROOT, "baseline", "a135252")
|
||||
self.assertEqual(pins["base_commit"], A135)
|
||||
self.assertEqual(set(pins["sources"]), tracked(A135))
|
||||
with TemporaryDirectory() as directory:
|
||||
root = historical_tree(A135, directory)
|
||||
for name, expected in pins["sources"].items():
|
||||
self.assertEqual(hashlib.sha256((root / name).read_bytes()).hexdigest(), expected)
|
||||
self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"])
|
||||
self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"])
|
||||
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
|
||||
source_guard(root, "baseline", "a135252")
|
||||
|
||||
def test_named_baseline_accepts_only_reversed_header(self):
|
||||
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
|
||||
with TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
origin = historical_tree(A135, Path(directory) / "origin")
|
||||
root = Path(directory) / "fixture"
|
||||
for name in pins["sources"]:
|
||||
dest = root / name
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(ROOT / name, dest)
|
||||
shutil.copyfile(origin / name, dest)
|
||||
subprocess.run(["git", "apply", "--reverse", str(HERE / "les/data/smagorinsky_normalization.patch")], cwd=root, check=True)
|
||||
with patch("run_uniform_les_flow.subprocess.check_output", return_value="baseline-fixture\n"):
|
||||
source_guard(root, "baseline", "a135252")
|
||||
|
||||
Reference in new issue
Block a user