229 lines
7.0 KiB
Python
229 lines
7.0 KiB
Python
"""Append-only evidence ledger: schema, idempotent insert, per-target status."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from cli import app
|
|
from lead.ledger.schema import schema_sql
|
|
from lead.ledger.store import append, open_ledger, status
|
|
from lead.targets import TARGETS
|
|
|
|
runner = CliRunner()
|
|
|
|
TIERS = ("rule", "aggregate", "literature", "synthetic")
|
|
|
|
|
|
def claim_row(**overrides: object) -> dict[str, object]:
|
|
row: dict[str, object] = {
|
|
"kind": "precursor",
|
|
"subject": "readmission",
|
|
"target_id": "Q.ACR",
|
|
"tiers": "literature",
|
|
"content_hash": "abc",
|
|
"actor": "tester",
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
def test_append_is_idempotent_and_status_covers_every_target(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
row = claim_row(target_id="Q.ACR", tiers="literature", content_hash="abc")
|
|
assert append(con, "claim", row) is True
|
|
assert append(con, "claim", row) is False
|
|
assert (
|
|
con.execute("select count(*) from lead_evidence.claim").fetchone()[0] == 1
|
|
)
|
|
created_at = con.execute(
|
|
"select created_at from lead_evidence.claim"
|
|
).fetchone()[0]
|
|
assert created_at
|
|
found = {s.target_id: s for s in status(con)}
|
|
assert set(found) == {t.id for t in TARGETS}
|
|
assert found["Q.ACR"].claims_by_tier["literature"] == 1
|
|
assert found["Q.ACR"].gaps == 0
|
|
assert found["HN.1"].gaps == 1
|
|
assert found["HN.1"].claims_by_tier == dict.fromkeys(TIERS, 0)
|
|
assert found["HN.1"].open_contradictions == 0
|
|
assert found["HN.1"].last_run is None
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_unknown_claim_target_raises(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
with pytest.raises(KeyError):
|
|
append(con, "claim", claim_row(target_id="nope", content_hash="missing"))
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_contradicting_finding_bumps_open_contradictions(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
assert append(con, "claim", claim_row(content_hash="abc", id="abc"))
|
|
assert append(
|
|
con,
|
|
"finding",
|
|
{
|
|
"claim_id": "abc",
|
|
"relation": "confirms",
|
|
"grain": "beneficiary",
|
|
"content_hash": "f-ok",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
found = {s.target_id: s for s in status(con)}
|
|
assert found["Q.ACR"].open_contradictions == 0
|
|
assert append(
|
|
con,
|
|
"finding",
|
|
{
|
|
"claim_id": "abc",
|
|
"relation": "contradicts",
|
|
"grain": "beneficiary",
|
|
"content_hash": "f-no",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
found = {s.target_id: s for s in status(con)}
|
|
assert found["Q.ACR"].open_contradictions == 1
|
|
assert found["HN.1"].open_contradictions == 0
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_last_run_is_the_latest_finished_on_for_that_target(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
assert append(
|
|
con,
|
|
"run",
|
|
{
|
|
"target_id": "Q.ACR",
|
|
"where_ran": "local",
|
|
"finished_on": "2026-10-30",
|
|
"content_hash": "run-old",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
assert append(
|
|
con,
|
|
"run",
|
|
{
|
|
"target_id": "Q.ACR",
|
|
"where_ran": "databricks",
|
|
"finished_on": "2026-10-31",
|
|
"content_hash": "run-new",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
found = {s.target_id: s for s in status(con)}
|
|
assert found["Q.ACR"].last_run == "2026-10-31"
|
|
assert found["HN.1"].last_run is None
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_a_tier_list_counts_each_named_tier(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
assert append(
|
|
con, "claim", claim_row(content_hash="both", tiers="rule, synthetic")
|
|
)
|
|
found = {s.target_id: s for s in status(con)}
|
|
assert found["Q.ACR"].claims_by_tier["rule"] == 1
|
|
assert found["Q.ACR"].claims_by_tier["synthetic"] == 1
|
|
assert found["Q.ACR"].claims_by_tier["literature"] == 0
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_delta_sql_names_the_schema_and_not_a_catalog():
|
|
sql = schema_sql("delta")
|
|
assert "lead_evidence" in sql
|
|
assert "aco" not in sql
|
|
with pytest.raises(ValueError):
|
|
schema_sql("nope")
|
|
|
|
|
|
def test_search_exogenous_and_extraction_columns_roundtrip(tmp_path):
|
|
con = open_ledger(str(tmp_path / "ledger.duckdb"))
|
|
try:
|
|
assert append(
|
|
con,
|
|
"search",
|
|
{
|
|
"project": "lead-precursors",
|
|
"database": "pubmed",
|
|
"interface": "eutils",
|
|
"searched_on": "2026-10-31",
|
|
"query": "frailty",
|
|
"hit_count": 2,
|
|
"export_sha256": "deadbeef",
|
|
"content_hash": "s1",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
assert append(
|
|
con,
|
|
"exogenous",
|
|
{
|
|
"project": "lead-external",
|
|
"target_id": "Q.ACR",
|
|
"content_hash": "e1",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
assert append(
|
|
con,
|
|
"extraction",
|
|
{
|
|
"project": "lead-precursors",
|
|
"zotero_key": "ABCD1234",
|
|
"extraction_version": "v1",
|
|
"body": "target_id: Q.ACR",
|
|
"content_hash": "x1",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
assert append(
|
|
con,
|
|
"source",
|
|
{
|
|
"zotero_key": "ABCD1234",
|
|
"doi": "10.1/x",
|
|
"database": "pubmed",
|
|
"content_hash": "src1",
|
|
"actor": "tester",
|
|
},
|
|
)
|
|
searched = con.execute(
|
|
"select project, searched_on from lead_evidence.search"
|
|
).fetchone()
|
|
assert searched == ("lead-precursors", "2026-10-31")
|
|
exogenous = con.execute(
|
|
"select target_id from lead_evidence.exogenous"
|
|
).fetchone()
|
|
assert exogenous == ("Q.ACR",)
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_status_command_prints_every_target(tmp_path):
|
|
path = tmp_path / "ledger.duckdb"
|
|
result = runner.invoke(app, ["lead", "ledger", "status", "--db", str(path)])
|
|
assert result.exit_code == 0
|
|
assert "Q.ACR" in result.stdout
|
|
|
|
|
|
def test_status_command_prints_gaps_runs_contradictions_and_tiers(tmp_path):
|
|
path = tmp_path / "ledger.duckdb"
|
|
result = runner.invoke(app, ["lead", "ledger", "status", "--db", str(path)])
|
|
assert result.exit_code == 0
|
|
assert "Q.ACR\t1\t\t0\t0\t0\t0\t0" in result.stdout
|