"""Aggregate verification tests: temporary files and an identity codec only. These test metadata/content binding, not zstd itself. The separate archive-tool suite includes its optional real-zstd integration test. No real project data, network, Git mutation, GPU, or persistent outputs are used here. """ import contextlib import gzip import io import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch import publish_archives as P import verify_publication as V def identity_codec(*args, **kwargs): return [sys.executable, "-c", "import sys,shutil;shutil.copyfileobj(sys.stdin.buffer,sys.stdout.buffer)"] class AggregateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.source = self.base / "source" self.data = self.source / "data" self.data.mkdir(parents=True) (self.data / "LICENSE").write_bytes(b"Synthetic license\n") (self.data / "one.feather").write_bytes(b"fake binary bytes\x00") self.pins = { p.relative_to(self.source).as_posix(): dict( path=p.relative_to(self.source).as_posix(), kind="file", **P.sha_file(p)) for p in sorted(self.data.iterdir()) } self.plan = { "schema": "lfs-publication-plan-v1", "source_root": str(self.source), "groups": [{ "id": "malecns-synaptic-partners", "paths": ["data"], "license": {"evidence_file": "LICENSE", "evidence_sha256": self.pins["data/LICENSE"]["sha256"]}, "provenance": {"registered_file_sha256": self.pins["data/one.feather"]["sha256"]}, }], "exclude_paths": [], } self.plan_file = self.base / "plan.json" self.root, self.work = self.base / "payload", self.base / "work" self.receipt = self.base / "receipt.json" self.history_file = self.base / "history.jsonl.gz" for item in (patch.object(P, "SOURCE_ROOT", str(self.source)), patch.object(P, "CHUNK_BYTES", 4096), patch.object(P, "zstd_command", identity_codec)): item.start() self.addCleanup(item.stop) def pack(self): self.plan_file.write_bytes(P.canonical(self.plan)) with contextlib.redirect_stdout(io.StringIO()): P.pack(self.plan_file, self.root, self.work) def verify(self, history=False): with contextlib.redirect_stdout(io.StringIO()): return V.verify_publication(self.root, self.receipt, self.history_file if history else None) def write_history(self, rows): with gzip.open(self.history_file, "wt", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row) + "\n") def test_full_without_history_and_exclusive_receipt(self): self.pack() result = self.verify() self.assertTrue(result["pass_all"]) self.assertFalse(result["fresh_download_proven"]) self.assertEqual(result["historical_status"], "NOT_REQUESTED") self.assertEqual(result["source_files"], 2) with self.assertRaises(ValueError): self.verify() def test_full_history_absolute_path_and_skipped_nonfile(self): self.pack() rows = [dict(row) for row in self.pins.values()] # Use the exact registered source_root prefix on either Windows/Linux. rows[0]["path"] = str(self.source) + "/" + rows[0]["path"] rows.append({"path": "outside/link", "kind": "symlink"}) self.write_history(rows) result = self.verify(history=True) report = json.loads(self.receipt.read_bytes()) self.assertEqual(result["historical_status"], "FULL_CURRENT_MEMBER_MATCH") self.assertEqual(report["historical_inventory"]["nonfile_records_skipped"], 1) def test_partial_history_is_not_full_historical_pass(self): self.pack() self.write_history([self.pins["data/LICENSE"], {"path": "data/one.feather", "kind": "symlink"}]) self.verify(history=True) history = json.loads(self.receipt.read_bytes())["historical_inventory"] self.assertEqual(history["status"], "PARTIAL_CURRENT_MEMBER_COVERAGE") self.assertFalse(history["all_current_members_historically_matched"]) self.assertEqual(history["current_members_without_regular_history"], 1) self.assertEqual(history["current_members_with_nonfile_history"], 1) def test_history_mismatch_writes_no_receipt(self): self.pack() self.write_history([dict(self.pins["data/LICENSE"], sha256="0" * 64)]) with self.assertRaisesRegex(ValueError, "Historical member mismatch"): self.verify(history=True) self.assertFalse(self.receipt.exists()) def test_registered_feather_sha_mismatch(self): self.plan["groups"][0]["provenance"]["registered_file_sha256"] = "0" * 64 self.pack() with self.assertRaisesRegex(ValueError, "Registered feather SHA"): self.verify() def test_multiple_feather_members_refused(self): (self.data / "two.feather").write_bytes(b"another fixture") self.pack() with self.assertRaisesRegex(ValueError, "exactly one feather"): self.verify() def test_license_evidence_sha_mismatch(self): self.plan["groups"][0]["license"]["evidence_sha256"] = "0" * 64 self.pack() with self.assertRaisesRegex(ValueError, "License evidence SHA"): self.verify() def test_extra_root_entry_refused(self): self.pack() (self.root / "unlisted").write_bytes(b"x") with self.assertRaisesRegex(ValueError, "extra/missing"): self.verify() def test_wrong_catalog_pin_refused(self): self.pack() summary = P.json_read(self.root / "pack-summary.json") summary["groups"][0]["catalog_sha256"] = "0" * 64 (self.root / "pack-summary.json").write_bytes(P.canonical(summary)) with self.assertRaisesRegex(ValueError, "Catalog SHA"): self.verify() def test_root_file_count_mismatch(self): self.pack() summary = P.json_read(self.root / "pack-summary.json") summary["source_files"] = 3 (self.root / "pack-summary.json").write_bytes(P.canonical(summary)) with self.assertRaisesRegex(ValueError, "Root source file count"): self.verify() def test_receipt_inside_payload_refused(self): self.pack() self.receipt = self.root / "receipt.json" with self.assertRaisesRegex(ValueError, "outside"): self.verify() def test_plan_raw_sha_mismatch(self): self.pack() path = self.root / "plan.json" path.write_bytes(path.read_bytes() + b"\n") with self.assertRaisesRegex(ValueError, "plan SHA"): self.verify() class GitMetadataTests(unittest.TestCase): def test_ascii_head_with_non_ascii_stderr(self): result = subprocess.CompletedProcess([], 0, stdout=b"a" * 40 + b"\n", stderr=bytes([255])) with patch.object(V.subprocess, "run", return_value=result): self.assertEqual(V.git_identity()["head"], "a" * 40) def test_failed_git_with_non_ascii_stderr_is_optional(self): result = subprocess.CompletedProcess([], 128, stdout=b"", stderr=bytes([255])) with patch.object(V.subprocess, "run", return_value=result): self.assertIsNone(V.git_identity()["head"]) if __name__ == "__main__": unittest.main()