File size: 8,373 Bytes
76b78ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """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")
|