import hashlib import json import os import re import subprocess import sys import tempfile import unittest ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(ROOT, "verifier")) sys.path.insert(0, os.path.join(ROOT, "fixtures")) import verify_checkpoint as vc # noqa: E402 import quarantine as q # noqa: E402 import make_fixtures as mf # noqa: E402 FX = os.path.join(ROOT, "fixtures") VERIFY = os.path.join(ROOT, "verifier", "verify_checkpoint.py") QUAR = os.path.join(ROOT, "verifier", "quarantine.py") def fx(name): return os.path.join(FX, name) def read(path): with open(path, "rb") as f: return f.read() def run(script, *args): p = subprocess.run([sys.executable, script, *args], capture_output=True, text=True, timeout=30) return p.returncode, p.stdout, p.stderr class VerifyTests(unittest.TestCase): def test_clean_passes(self): rc, out, _ = run(VERIFY, fx("clean_store.jsonl"), fx("clean_checkpoint.json")) self.assertEqual(rc, 0, out) def test_contaminated_chain_is_valid(self): rc, _, _ = run(VERIFY, fx("contaminated_store.jsonl"), fx("contaminated_checkpoint.json")) self.assertEqual(rc, 0) def test_rewrite_with_recomputed_chain_detected(self): rc, out, _ = run(VERIFY, fx("tampered_store.jsonl"), fx("clean_checkpoint.json")) self.assertEqual(rc, 1) self.assertIn("prefix", out) def test_naive_edit_detected(self): rc, out, _ = run(VERIFY, fx("tampered_naive_store.jsonl"), fx("clean_checkpoint.json")) self.assertEqual(rc, 1) self.assertIn("hash does not match", out) def test_truncation_detected(self): rc, out, _ = run(VERIFY, fx("truncated_store.jsonl"), fx("clean_checkpoint.json")) self.assertEqual(rc, 1) self.assertIn("truncated", out) def test_middle_deletion_detected(self): lines = read(fx("clean_store.jsonl")).splitlines(keepends=True) del lines[3] cp = vc.load_checkpoint(fx("clean_checkpoint.json")) self.assertTrue(vc.verify(b"".join(lines), cp)) def test_append_only_is_allowed(self): data = read(fx("clean_store.jsonl")) cp = vc.make_checkpoint(data, 5) self.assertEqual(vc.verify(data, cp), []) def test_reorder_detected(self): lines = read(fx("clean_store.jsonl")).splitlines(keepends=True) lines[1], lines[2] = lines[2], lines[1] self.assertTrue(vc.verify(b"".join(lines), vc.load_checkpoint(fx("clean_checkpoint.json")))) def test_usage_errors(self): self.assertEqual(run(VERIFY)[0], 2) self.assertEqual(run(VERIFY, "/nonexistent/x", fx("clean_checkpoint.json"))[0], 2) def test_deterministic_output(self): a = run(VERIFY, fx("clean_store.jsonl"), fx("clean_checkpoint.json")) b = run(VERIFY, fx("clean_store.jsonl"), fx("clean_checkpoint.json")) self.assertEqual(a, b) class QuarantineTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.out = os.path.join(self.tmp.name, "out") self.addCleanup(self.tmp.cleanup) self.store = fx("contaminated_store.jsonl") self.cp = fx("contaminated_checkpoint.json") def quarantine(self, *extra): return run(QUAR, self.store, self.cp, "--out-dir", self.out, *extra) def reconstruct(self): clean = read(os.path.join(self.out, "clean_store.jsonl")).splitlines(keepends=True) quar = [json.loads(l) for l in read(os.path.join(self.out, "quarantine.jsonl")).decode().splitlines()] return clean, quar def test_union_is_byte_exact_and_nothing_deleted(self): rc, out, err = self.quarantine() self.assertEqual(rc, 0, err) clean, quar = self.reconstruct() original = read(self.store).splitlines(keepends=True) by_pos = {r["original_position"]: r["original_line"].encode() for r in quar} rebuilt = clean + [by_pos[p] for p in sorted(by_pos)] self.assertEqual(rebuilt, original) self.assertEqual(len(clean) + len(quar), len(original)) def test_flagged_entries_quarantined_with_reason_and_position(self): self.quarantine() _, quar = self.reconstruct() rec = {r["id"]: r for r in quar} self.assertEqual(rec["e7"]["reason"], "contaminated") self.assertEqual(rec["e7"]["original_position"], 7) self.assertEqual(rec["e8"]["reason"], "contaminated") self.assertEqual(rec["e6"]["reason"], "after_checkpoint_unverified") def test_clean_store_is_verified_prefix(self): self.quarantine() data = read(os.path.join(self.out, "clean_store.jsonl")) self.assertEqual(vc.verify(data, vc.load_checkpoint(self.cp)), []) rc, _, _ = run(VERIFY, os.path.join(self.out, "clean_store.jsonl"), self.cp) self.assertEqual(rc, 0) self.assertNotIn(b"SYSTEM NOTE", data) def test_bundle_minimal_and_reverifiable(self): self.quarantine() b = json.loads(read(os.path.join(self.out, "clean_bundle.json"))) self.assertEqual(set(b), {"schema", "checkpoint_ref", "retained_ids", "quarantined_ids", "contaminated_ids", "bundle_sha256", "reverify"}) self.assertEqual(b["retained_ids"], ["e1", "e2", "e3", "e4", "e5"]) self.assertEqual(b["contaminated_ids"], ["e7", "e8"]) self.assertEqual(b["checkpoint_ref"]["file_sha256"], hashlib.sha256(read(self.cp)).hexdigest()) self.assertEqual(b["bundle_sha256"], q.bundle_hash(b)) b["retained_ids"].append("evil") self.assertNotEqual(b["bundle_sha256"], q.bundle_hash(b)) def test_explicit_ids(self): rc, _, _ = self.quarantine("--ids", "e7") self.assertEqual(rc, 0) b = json.loads(read(os.path.join(self.out, "clean_bundle.json"))) self.assertEqual(b["contaminated_ids"], ["e7"]) def test_no_overwrite_without_force(self): self.assertEqual(self.quarantine()[0], 0) path = os.path.join(self.out, "clean_store.jsonl") with open(path, "ab") as f: f.write(b"marker\n") before = read(path) rc, _, err = self.quarantine() self.assertEqual(rc, 2) self.assertIn("--force", err) self.assertEqual(read(path), before) self.assertEqual(self.quarantine("--force")[0], 0) self.assertNotEqual(read(path), before) def test_refuses_ids_inside_checkpoint(self): rc, _, err = self.quarantine("--ids", "e3") self.assertEqual(rc, 1) self.assertIn("checkpoint", err) self.assertFalse(os.path.exists(self.out)) def test_refuses_to_overwrite_checkpoint_or_store(self): d = self.tmp.name cp = os.path.join(d, "clean_store.jsonl") with open(cp, "wb") as f: f.write(read(self.cp)) before = read(cp) rc, _, _ = run(QUAR, self.store, cp, "--out-dir", d, "--force") self.assertEqual(rc, 1) self.assertEqual(read(cp), before) def test_refuses_tampered_store(self): rc, _, _ = run(QUAR, fx("tampered_store.jsonl"), fx("clean_checkpoint.json"), "--out-dir", self.out) self.assertEqual(rc, 1) self.assertFalse(os.path.exists(self.out)) def test_unknown_id_and_usage(self): self.assertEqual(self.quarantine("--ids", "nope")[0], 1) self.assertEqual(run(QUAR)[0], 2) def test_checkpoint_untouched(self): before = read(self.cp) self.quarantine() self.assertEqual(read(self.cp), before) class PackageTests(unittest.TestCase): def test_fixture_regeneration_byte_equal(self): with tempfile.TemporaryDirectory() as d: mf.main([d]) for name in os.listdir(d): self.assertEqual(read(os.path.join(d, name)), read(fx(name)), name) self.assertEqual(sorted(os.listdir(d)), sorted(n for n in os.listdir(FX) if n != "make_fixtures.py" and not n.startswith("__"))) def test_fixtures_synthetic_only(self): for n in os.listdir(FX): if n.endswith(".jsonl"): text = read(fx(n)).decode() for url in re.findall(r"https?://[^\s\"]+", text): self.assertIn("example.invalid", url) def test_skill_frontmatter_name_matches_dir(self): d = os.path.join(ROOT, "skills", "bilateral-reset") text = read(os.path.join(d, "SKILL.md")).decode() m = re.match(r"---\n(.*?)\n---\n", text, re.S) self.assertTrue(m) keys = dict(l.split(": ", 1) for l in m.group(1).splitlines()) self.assertEqual(set(keys), {"name", "description"}) self.assertEqual(keys["name"], os.path.basename(d)) self.assertIn("not an enforcement gate", text) BANNED = re.compile(r"cortisol|\bdiagnos\w*|\btreat\w*|\bcure\b|\btherap\w*|\bhertz\b|\bhz\b|binaural|clinically", re.I) def test_operator_docs_claims_lint(self): for name in ("attention_shift_audio_script.md", "reset_card.md"): text = read(os.path.join(ROOT, "operator", name)).decode() self.assertEqual(text.count(""), 1, name) self.assertEqual(text.count(""), 1, name) stripped = re.sub(r".*?", "", text, flags=re.S) self.assertIsNone(self.BANNED.search(stripped), (name, self.BANNED.search(stripped))) def test_script_is_labelled_and_optional(self): text = read(os.path.join(ROOT, "operator", "attention_shift_audio_script.md")).decode() self.assertIn("SCRIPT", text.splitlines()[0]) self.assertIn("No audio file ships", text) self.assertRegex(text, r"skip it or stop") def test_lint_would_catch_banned_terms(self): for bad in ("lowers cortisol", "a therapy", "432 Hz tone", "binaural beats", "clinically proven"): self.assertTrue(self.BANNED.search(bad), bad) self.assertIsNone(self.BANNED.search("a secure store")) def test_manifest(self): m = json.loads(read(os.path.join(ROOT, "manifest.json"))) self.assertEqual(m["name"], "bilateral-reset") self.assertEqual(m["version"], "0.1.0") self.assertEqual(m["tools_permissions"]["verifier"]["network"], "none") self.assertEqual(m["tools_permissions"]["verifier"]["subprocess"], False) if __name__ == "__main__": unittest.main()