"""Unit tests for the journalism suite (research/journalism.py + layers). Run: .venv/bin/python tests/test_journalism.py """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from research.provenance import ProvenanceLedger, evaluate_source_policy from research.timeline import TimelineAnalyzer from research.framing import FramingAnalyzer from research.patterns import CrossDomainPatterns from research.entitygraph import EntityGraph from research.editorial_review import editorial_review from research.casefile import CaseFile from research.journalism import suite_report def test_provenance_credibility_tiers(): led = ProvenanceLedger(path=None) led.register_source("s1", "leaked filing", tier="verified-leak") led.register_source("s2", "rumor", tier="claim", retrievable=True) led.register_source("s3", "unretrievable", tier="secondary", retrievable=False) assert led.sources["s1"].credibility() > led.sources["s2"].credibility() assert led.sources["s3"].credibility() == round(0.6 * 0.4, 3) assert led.sources["s2"].credibility() == round(0.1, 3) def test_provenance_chain_and_single_source(): led = ProvenanceLedger(path=None) led.register_source("s1", "filing", tier="verified-leak", url="file://a") led.register_source("s2", "republication", tier="secondary", url="file://a", independent=False) led.record_claim("bridge opened 2010", ["s1", "s2"]) corr = led.corroboration("bridge opened 2010") assert len(corr) == 1 # s2 is a derived republication, deduped assert len(led.single_source()) == 1 def test_source_policy_requires_independent_traceable_corroboration(): sources = [ {"source_id": "s1", "url": "https://records.example/filing", "retrieved_at": "2026-08-12T12:00:00Z", "content_sha256": "a" * 64, "independent": True, "retrievable": True, "triage": {"independence": 3, "proximity": 3, "recency": 2, "track": 3, "interest": 3}}, {"source_id": "s2", "url": "https://archive.example/report", "retrieved_at": "2026-08-12T12:01:00Z", "content_sha256": "b" * 64, "independent": True, "retrievable": True, "triage": {"independence": 2, "proximity": 2, "recency": 2, "track": 2, "interest": 2}}, ] policy = evaluate_source_policy(sources) assert policy["verified"] and policy["independent_usable"] == 2 def test_source_policy_rejects_untraceable_or_duplicate_leads(): sources = [ {"source_id": "s1", "url": "https://forum.example/post", "retrieved_at": "", "content_sha256": "not-a-hash", "independent": True, "retrievable": True, "triage": {"independence": 1, "proximity": 0, "recency": 1, "track": 0, "interest": 0}}, {"source_id": "s2", "url": "https://forum.example/repost", "origin": "https://forum.example/post", "retrieved_at": "2026-08-12T12:00:00Z", "content_sha256": "c" * 64, "independent": False, "retrievable": True, "triage": {"independence": 1, "proximity": 1, "recency": 1, "track": 1, "interest": 1}}, ] policy = evaluate_source_policy(sources) assert not policy["verified"] assert policy["independent_usable"] == 0 def test_timeline_gaps_and_cliffs(): tl = TimelineAnalyzer() tl.add_event("2010-01-01", "filing A", "s1") tl.add_event("2010-06-01", "filing A2", "s1") tl.add_event("2011-01-01", "filing B", "s2") tl.add_event("2013-01-01", "filing C", "s3") tl.add_event("2013-06-01", "filing C2", "s3") gaps = tl.gaps() assert len(gaps) == 1 # 2011->2013 is 2 years > floor assert "no recorded event" in gaps[0]["absent"] # 2012 is silent between active 2011 and 2013 cliffs = tl.cliffs() assert any(c["year"] == "2012" for c in cliffs) def test_timeline_anachronism(): tl = TimelineAnalyzer() tl.add_event("2010-06-01", "the 2012 report was sealed", "s1") an = tl.anachronisms() assert len(an) == 1 and an[0]["flag"].startswith("cited year") def test_framing_passive_loaded_hedges(): fr = FramingAnalyzer() fr.add_doc("s1", "The memo was destroyed. The scandal was allegedly covered up.") c = fr.doc_card("s1") assert c["passive_hits"] >= 2 assert any(w == "scandal" for w, _ in c["loaded"]) assert any(w == "allegedly" for w, _ in c["hedges"]) def test_framing_omissions(): fr = FramingAnalyzer() fr.add_doc("s1", "The committee discussed the budget and the bridge.") fr.add_doc("s2", "The committee discussed the bridge only.") om = fr.omissions(["budget"]) assert any(o["source_id"] == "s2" for o in om) def test_patterns_shared_rungs_and_themes(): p = CrossDomainPatterns() p.add_strand("economics", "the serpent of speculation and the 1929 crash") p.add_strand("religion", "the serpent in the garden, then 1929") assert any(c["rung"] == "1929" and "economics" in c["domains"] and "religion" in c["domains"] for c in p.shared_rungs()) assert any(c["theme"] == "serpent" for c in p.theme_overlap()) assert "LEAD, never a verdict" in p.report() def test_entitygraph_edges_and_centrality(): g = EntityGraph() g.add_doc("s1", "Central Bank met Delta Corp. Delta Corp hired Smith. " "Central Bank fired Smith. Central Bank met Delta Corp again.") assert "Central Bank" in g._nodes() edges = g.edges(min_cooccur=2) assert ("Central Bank", "Delta Corp") in edges assert g.central()[0][0] in ("Central Bank", "Delta Corp") def test_editorial_review_flags(): r = editorial_review("Clearly the cover-up is the only explanation and " "nobody disputes it, so it must be the FBI.", sources=1, counter_evidence=False, has_dates=False) assert r["flags"] >= 3 assert r["summary"].startswith("HOLD") kinds = {c["item"] for c in r["cards"]} assert "leading question" in kinds and "overclaim" in kinds def test_editorial_review_clean(): r = editorial_review("The state filing lists the bridge opening year as 2010.", sources=2, counter_evidence=True, has_dates=True) assert r["flags"] == 0 assert r["summary"] == "CLEAR TO PUBLISH (with citation audit)" def test_casefile_roundtrip(): cf = CaseFile("test_case_journalism") cf.add_source("s1", "DOT filing", tier="verified-leak") cf.add_finding("main", "bridge opened 2010", "supports", "HIGH", ["s1"]) md = cf.export_markdown() assert "DOT filing" in md and "bridge opened 2010" in md def test_suite_report_end_to_end(): docs = [ {"source_id": "s1", "title": "DOT filing", "tier": "verified-leak", "url": "file://dot", "date": "2010-06-01", "text": "The bridge opened in 2010. The 1929 crash changed funding. " "Delta Corp signed the contract."}, {"source_id": "s2", "title": "Press release", "tier": "secondary", "date": "2012-06-01", "text": "The bridge was allegedly opened on time. The serpent symbol " "on the plaque was noted. Delta Corp celebrated."}, ] claims = [{"claim": "bridge opened 2010", "source_ids": ["s1", "s2"], "verdict": "supports", "confidence": "HIGH", "counter_evidence": True, "has_dates": True}] md = suite_report("test_case_journalism", docs, claims) for needle in ("Provenance Ledger", "Timeline", "Framing", "Cross-Domain Pattern", "Entity Relationship", "Source Policy Gate", "Pre-Publication Adversarial Review", "CaseFile"): assert needle in md def test_suite_report_fails_closed_without_source_policy_metadata(): docs = [{"source_id": "s1", "title": "Unattributed copy", "text": "The bridge opened in 2010."}] claims = [{"claim": "bridge opened 2010", "source_ids": ["s1"], "verdict": "supports", "confidence": "HIGH"}] md = suite_report("test_source_policy_gate", docs, claims) assert "[LEAD ONLY] bridge opened 2010 -> not enough information" in md if __name__ == "__main__": fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] for fn in fns: fn() print(f"ok {fn.__name__}") print(f"\n{len(fns)} journalism tests passed")