import copy import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from harness.tests._support.paths import ROOT from harness.tests._support.spec import MODPATH, mod, publication_fixture, result, verify_fixture class PublicationTests(unittest.TestCase): def test_publication_accepts_one_exact_review_and_rejects_competing_event(self): before, after, review, r, paths = publication_fixture() self.assertEqual(verify_fixture(before, after, review, r, paths)["review_id"], 11) bad = copy.deepcopy(after) bad["timeline"].append({"id": 102, "review_id": 12, "user": {"id": 42}, "body": review["body"]}) with self.assertRaises(mod.ValidationError): verify_fixture(before, bad, review, r, paths) def test_publication_requires_exact_readback_not_only_review_id(self): before, after, review, r, paths = publication_fixture() for field, value in (("body", "changed"), ("official", False), ("submitted_at", "bad")): exact = copy.deepcopy(review); exact[field] = value with self.subTest(field=field), self.assertRaises(mod.ValidationError): verify_fixture(before, after, exact, r, paths) def test_publication_rejects_unrelated_drift_and_bad_projections(self): before, after, review, r, paths = publication_fixture() cases = [] for field, value in (("updated_at", "2026-10-03T10:00:00Z"), ("review_comments", 4), ("requested_reviewers", [{"id": 43}]), ("head", {"sha": "d" * 40})): bad = copy.deepcopy(after); bad["pr"][field] = value; cases.append(bad) for field, value in (("updated_at", "2026-10-03T10:00:00Z"), ("title", "changed")): bad = copy.deepcopy(after); bad["issue"][field] = value; cases.append(bad) bad = copy.deepcopy(after); bad["reviews"][0]["stale"] = 0; cases.append(bad) bad = copy.deepcopy(after); bad["reviews"][0]["state"] = "PENDING"; cases.append(bad) bad = copy.deepcopy(after); bad["reviews"].append(copy.deepcopy(review)); cases.append(bad) bad = copy.deepcopy(after); bad["timeline"][0]["type"] = "comment"; cases.append(bad) bad = copy.deepcopy(after); bad["timeline"][0]["review_id"] = True; cases.append(bad) for bad in cases: with self.subTest(snapshot=cases.index(bad)), self.assertRaises(mod.ValidationError): verify_fixture(before, bad, bad["reviews"][0], r, paths) def test_publication_binds_snapshot_repository_issue_and_head(self): before, after, review, r, paths = publication_fixture() for section, field, value in (("issue", "number", 4), ("issue", "repository", {"full_name": "wrong/repo"}), ("pr", "head", {"sha": "d" * 40}), ("pr", "number", 8)): b = copy.deepcopy(before); a = copy.deepcopy(after) b[section][field] = a[section][field] = value with self.subTest(section=section, field=field), self.assertRaises(mod.ValidationError): verify_fixture(b, a, review, r, paths) def test_publication_preserves_previous_reviews_and_bounds_supersession(self): before, after, review, r, paths = publication_fixture() old = copy.deepcopy(review); old.update(id=10, body="previous", commit_id="d" * 40, state="REQUEST_CHANGES", updated_at="2026-10-03T00:00:00Z") before["reviews"] = [old]; after["reviews"].insert(0, copy.deepcopy(old)) verify_fixture(before, after, review, r, paths) after["reviews"][0].update(official=False, dismissed=True, updated_at=review["submitted_at"]) with self.assertRaises(mod.ValidationError): verify_fixture(before, after, review, r, paths) contract = {"version": {"version": "1.27.3"}, "supersession_same_author_only": True, "already_dismissed_review_timestamp_projection": True} verify_fixture(before, after, review, r, paths, contract) for mutation in ("body", "user", "missing"): a = copy.deepcopy(after) if mutation == "missing": a["reviews"].pop(0) elif mutation == "user": a["reviews"][0]["user"] = {"id": 43} else: a["reviews"][0]["body"] = "tampered" with self.subTest(mutation=mutation), self.assertRaises(mod.ValidationError): verify_fixture(before, a, review, r, paths, contract) old["official"] = False; old["dismissed"] = True after["reviews"][0] = copy.deepcopy(old); after["reviews"][0]["updated_at"] = review["submitted_at"] verify_fixture(before, after, review, r, paths, contract) contract["already_dismissed_review_timestamp_projection"] = False with self.assertRaises(mod.ValidationError): verify_fixture(before, after, review, r, paths, contract) def test_publication_preserves_collection_identity_and_normalizes_only_files_order(self): before, after, review, r, paths = publication_fixture() second = {"filename": "tests/test_a.py", "status": "added", "additions": 2} before["files"].append(second); after["files"].insert(0, copy.deepcopy(second)) verify_fixture(before, after, review, r, paths) after["files"].append(copy.deepcopy(second)) with self.assertRaises(mod.ValidationError): verify_fixture(before, after, review, r, paths) def test_cli_verify_publication_and_rejects_raw_crlf_body(self): before, after, review, r, paths = publication_fixture() env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1") with tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") as td: root = Path(td) for name, value in (("before", before), ("after", after), ("exact", review), ("result", r), ("paths", sorted(paths)), ("inline", [])): (root / (name + ".json")).write_text(json.dumps(value, ensure_ascii=False)) cmd = [sys.executable, str(MODPATH), "verify-publication", "--before", str(root / "before.json"), "--after", str(root / "after.json"), "--exact-review", str(root / "exact.json"), "--result", str(root / "result.json"), "--allowed-paths", str(root / "paths.json"), "--inline-comments", str(root / "inline.json"), "--reviewer-id", "42", "--expected-pr-url", "https://example.test/acme/repo/pulls/7"] p = subprocess.run(cmd, capture_output=True, text=True, env=env) self.assertEqual(p.returncode, 0, p.stderr) self.assertEqual(json.loads(p.stdout)["review_id"], 11) body_path = root / "body.md"; body_path.write_bytes(review["body"].replace("\n", "\r\n").encode()) p = subprocess.run([sys.executable, str(MODPATH), "parse", "--body", str(body_path), "--allowed-paths", str(root / "paths.json")], capture_output=True, text=True, env=env) self.assertNotEqual(p.returncode, 0) self.assertEqual(json.loads(p.stderr), {"error": "invalid input"}) def test_publication_accepts_causal_event_timestamp_without_forcing_clock_equality(self): before, after, review, r, paths = publication_fixture() event = after["timeline"][0] event["created_at"] = event["updated_at"] = "2026-10-03T01:24:01Z" after["pr"]["updated_at"] = event["created_at"] verify_fixture(before, after, review, r, paths) def test_publication_requires_independent_canonical_pr_url_anchor(self): before, after, review, r, paths = publication_fixture() for foreign in ("https://evil.test/acme/repo/pulls/7", "https://example.test/prefix/acme/repo/pulls/7"): b = copy.deepcopy(before); a = copy.deepcopy(after) b["pr"]["html_url"] = b["pr"]["url"] = foreign a["pr"]["html_url"] = a["pr"]["url"] = foreign a["reviews"][0]["pull_request_url"] = foreign a["timeline"][0]["pull_request_url"] = foreign with self.subTest(url=foreign), self.assertRaises(mod.ValidationError): verify_fixture(b, a, a["reviews"][0], r, paths) if __name__ == "__main__": unittest.main()