stack/tests/lead/test_manifest.py

"""The enrichment manifest loads into local DuckDB. It does not download."""

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 open_ledger
from lead.manifest import RELEASE_DDL, load_manifest

runner = CliRunner()

_HEADER = (
    "source_id,title,publisher,kind,url,api_endpoint,landing_page,release,"
    "fetched_utc,status,path,sha256,rows,size_bytes,license_terms,lead_use"
)


def _csv(tmp_path, body: str):
    path = tmp_path / "manifest.csv"
    path.write_text(_HEADER + "\n" + body)
    return path


def test_load_is_idempotent_and_tags_synthetic_rows(tmp_path):
    csv_path = _csv(
        tmp_path,
        "ahrq,Title,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,abc,1,2,public,benchmark\n"
        "tuva,Title,Pub,table,https://example.test/b,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,def,1,2,public,\n"
        "cms,Title,Pub,table,https://example.test/c,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,ghi,1,2,public,synthetic person file\n",
    )
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        assert load_manifest(con, str(csv_path), actor="tester") == 3
        assert load_manifest(con, str(csv_path), actor="tester") == 0
        rows = con.execute(
            """
            SELECT source_id, url, path, layer
            FROM lead_evidence.public_release
            ORDER BY source_id
            """
        ).fetchall()
    finally:
        con.close()
    by_source = {row[0]: row for row in rows}
    assert by_source["ahrq"][1] == "https://example.test/a"
    assert by_source["ahrq"][2] == "/no/such/lead-file"
    assert by_source["ahrq"][3] == ""
    assert by_source["tuva"][3] == "synthetic"
    assert by_source["cms"][3] == "synthetic"


def test_a_blank_sha256_raises_and_inserts_nothing(tmp_path):
    csv_path = _csv(
        tmp_path,
        "ahrq,Title,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,,1,2,public,benchmark\n",
    )
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        with pytest.raises(ValueError, match=r"row 1:.*ahrq.*2024"):
            load_manifest(con, str(csv_path), actor="tester")
        count = con.execute(
            "SELECT count(*) FROM lead_evidence.public_release"
        ).fetchone()[0]
        assert count == 0
    finally:
        con.close()


def test_a_blank_hash_among_real_rows_is_stored_as_a_gap(tmp_path):
    csv_path = _csv(
        tmp_path,
        "ahrq,Title,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,abc,1,2,public,benchmark\n"
        "cms_mds,Full file,CMS,data,https://example.test/full,,,2026-04-01,"
        "2026-10-31,over_cap:10,,,10,public,benchmark\n",
    )
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        assert load_manifest(con, str(csv_path), actor="tester") == 2
        assert load_manifest(con, str(csv_path), actor="tester") == 0
        gap = con.execute(
            """
            SELECT sha256, layer, path
            FROM lead_evidence.public_release
            WHERE source_id = 'cms_mds'
            """
        ).fetchone()
        assert gap == ("", "gap", "")
    finally:
        con.close()


def test_the_same_bytes_at_two_paths_both_stay(tmp_path):
    csv_path = _csv(
        tmp_path,
        "ahrq,Office A,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
        "ahrq/a.csv,abc,1,2,public,benchmark\n"
        "ahrq,Office B,Pub,table,https://example.test/b,,,2024,2026-10-31,ok,"
        "ahrq/b.csv,abc,1,2,public,benchmark\n",
    )
    con = open_ledger(str(tmp_path / "ledger.duckdb"))
    try:
        assert load_manifest(con, str(csv_path), actor="tester") == 2
        assert load_manifest(con, str(csv_path), actor="tester") == 0
        paths = con.execute(
            """
            SELECT path FROM lead_evidence.public_release
            WHERE source_id = 'ahrq' ORDER BY path
            """
        ).fetchall()
        assert paths == [("ahrq/a.csv",), ("ahrq/b.csv",)]
    finally:
        con.close()


def test_release_ddl_and_delta_schema_omit_the_catalog_token():
    assert "aco" not in RELEASE_DDL
    assert "aco" not in schema_sql("delta")


def test_manifest_load_command_prints_the_insert_count(tmp_path):
    csv_path = _csv(
        tmp_path,
        "ahrq,Title,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
        "/no/such/lead-file,abc,1,2,public,benchmark\n",
    )
    db = tmp_path / "ledger.duckdb"
    result = runner.invoke(
        app,
        [
            "lead",
            "manifest",
            "load",
            "--csv",
            str(csv_path),
            "--db",
            str(db),
        ],
    )
    assert result.exit_code == 0
    assert "inserted\t1" in result.stdout