stack/tests/lead/test_ledger_claims.py

"""Claims from extractions, GRADE start, the pool gate, and the catalog."""

from __future__ import annotations

import pytest

from lead.ledger.claims import (
    claims_from_extraction,
    grade,
    may_pool,
    precursor_catalog,
)
from lead.ledger.store import append, open_ledger


def test_precursor_and_limit_extractions_become_claims():
    precursor = claims_from_extraction(
        {
            "zotero_key": "ABCD2345",
            "extraction_version": "v1",
            "outcome_target_id": "Q.ACR",
            "grain": "beneficiary",
            "lead_time": "12 months",
            "design": "cohort",
        },
        project="lead-precursors",
        actor="tester",
    )
    assert len(precursor) == 1
    claim = precursor[0]
    assert claim["kind"] == "precursor"
    assert claim["tiers"] == "literature"
    assert claim["grain"] == "beneficiary"
    assert claim["target_id"] == "Q.ACR"
    assert claim["lead_time"] == "12 months"
    assert claim["certainty"] == "high"

    limit = claims_from_extraction(
        {"target_id": "Q.ACR", "grain": "ACO"},
        project="lead-limits",
        actor="tester",
    )
    assert len(limit) == 1
    assert limit[0]["kind"] == "limit"
    assert limit[0]["target_id"] == "Q.ACR"
    assert limit[0]["grain"] == "ACO"


def test_grade_starts_high_for_three_designs_and_stops_at_very_low():
    assert grade("cohort", ()) == "high"
    assert grade("case-control", ()) == "high"
    assert grade("quasi-experimental", ()) == "high"
    assert grade("cohort", ("risk of bias",)) == "moderate"
    assert grade("cross-sectional", ()) == "low"
    assert grade("cohort", ("a", "b", "c", "d")) == "very low"
    assert grade("cohort", ("a", "b", "c", "d", "e")) == "very low"


def test_may_pool_needs_three_identical_measures():
    assert may_pool(["OR", "OR", "OR"]) is True
    assert may_pool(["OR", "OR"]) is False
    assert may_pool(["OR", "OR", "RR"]) is False


def test_same_extraction_inserts_one_claim_and_the_catalog_reads_it(tmp_path):
    extraction = {
        "zotero_key": "ABCD2345",
        "extraction_version": "v1",
        "outcome_target_id": "Q.ACR",
        "grain": "beneficiary",
        "lead_time": "12 months",
        "design": "cohort",
    }
    first = claims_from_extraction(
        extraction, project="lead-precursors", actor="tester"
    )[0]
    second = claims_from_extraction(
        extraction, project="lead-precursors", actor="tester"
    )[0]
    limit = claims_from_extraction(
        {"target_id": "Q.ACR", "grain": "ACO", "zotero_key": "EFGH5678"},
        project="lead-limits",
        actor="tester",
    )[0]
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        assert append(con, "claim", first) is True
        assert append(con, "claim", second) is False
        assert (
            con.execute("select count(*) from lead_evidence.claim").fetchone()[0] == 1
        )
        assert append(con, "claim", limit) is True
        catalog = precursor_catalog(con)
        row = next(item for item in catalog if item["target_id"] == "Q.ACR")
        assert row["lead_time"] == first["lead_time"]
        assert row["certainty"] == first["certainty"]
        assert row["grain"] == first["grain"]
        assert row["limit_claim_id"] == limit["content_hash"]
    finally:
        con.close()


def test_claim_keeps_effect_value_interval_and_subject():
    claim = claims_from_extraction(
        {
            "target_id": "Q.ACR",
            "effect_measure": "OR",
            "effect_value": "1.2",
            "effect_ci": "1.1, 1.4",
            "factor_definition": "frailty status",
            "design": "cohort",
        },
        project="lead-limits",
        actor="tester",
    )[0]
    assert claim["effect_measure"] == "OR"
    assert claim["effect_value"] == 1.2
    assert claim["effect_ci_low"] == 1.1
    assert claim["effect_ci_high"] == 1.4
    assert claim["subject"] == "frailty status"

    stated = claims_from_extraction(
        {
            "target_id": "Q.ACR",
            "effect_ci": "1.1 to 1.4",
            "statement": "care changes the status",
        },
        project="lead-limits",
        actor="tester",
    )[0]
    assert stated["effect_ci_low"] == 1.1
    assert stated["effect_ci_high"] == 1.4
    assert stated["subject"] == "care changes the status"


def test_catalog_lists_each_precursor_claim(tmp_path):
    first = claims_from_extraction(
        {
            "zotero_key": "ABCD2345",
            "extraction_version": "v1",
            "outcome_target_id": "Q.ACR",
            "grain": "beneficiary",
            "lead_time": "12 months",
            "design": "cohort",
        },
        project="lead-precursors",
        actor="tester",
    )[0]
    second = claims_from_extraction(
        {
            "zotero_key": "IJKM2345",
            "extraction_version": "v1",
            "outcome_target_id": "Q.ACR",
            "grain": "beneficiary",
            "lead_time": "6 months",
            "design": "cohort",
        },
        project="lead-precursors",
        actor="tester",
    )[0]
    limit = claims_from_extraction(
        {"target_id": "Q.ACR", "grain": "ACO", "zotero_key": "EFGH5678"},
        project="lead-limits",
        actor="tester",
    )[0]
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        assert append(con, "claim", first) is True
        assert append(con, "claim", second) is True
        assert append(con, "claim", limit) is True
        rows = [item for item in precursor_catalog(con) if item["target_id"] == "Q.ACR"]
        assert len(rows) == 2
        assert {row["lead_time"] for row in rows} == {"12 months", "6 months"}
        assert {row["limit_claim_id"] for row in rows} == {limit["content_hash"]}
    finally:
        con.close()


def test_unknown_target_raises_key_error():
    with pytest.raises(KeyError):
        claims_from_extraction(
            {"outcome_target_id": "nope", "grain": "beneficiary"},
            project="lead-precursors",
            actor="tester",
        )