tests/test_journalism.py

"""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")