stack/tests/lead/test_load_public.py

"""Public files load into a local DuckDB mirror. Nothing is downloaded."""

from __future__ import annotations

import hashlib
import json
import zipfile
from pathlib import Path

import duckdb
import openpyxl
import pytest
from typer.testing import CliRunner

from cli import app
from lead.load import public as public_load
from lead.load.errors import HashMismatch
from lead.load.frame import LOAD_LOG_DDL
from lead.load.public import _NeedsSerial, load_file, load_manifest_files

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 _sha(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _manifest(root: Path, lines: list[str]) -> Path:
    path = root / "manifest.csv"
    path.write_text(_HEADER + "\n" + "\n".join(lines) + "\n")
    return path


def _row(
    source: str, release: str, rel: str, digest: str, *, kind: str = "data"
) -> str:
    return (
        f"{source},Title,Pub,{kind},http://example.invalid/not-fetched,,,{release},"
        f"2026-11-01,ok,{rel},{digest},1,1,public,benchmark"
    )


def test_a_csv_lands_on_the_spine_and_a_second_load_inserts_nothing(tmp_path):
    src = tmp_path / "rows.csv"
    src.write_text("FIPS,Year,Quarter,n\n01001,2024,1,4\n01000,2024,1,9\n")
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="lead_public.demo",
                source_id="demo",
                release="2024",
                sha256=_sha(src),
                actor="tester",
            )
            == 2
        )
        assert (
            load_file(
                con,
                src,
                table="lead_public.demo",
                source_id="demo",
                release="2024",
                sha256=_sha(src),
                actor="tester",
            )
            == 0
        )
        rows = con.execute(
            """
            SELECT geo_fips, spine_year, spine_quarter, time_key, n
            FROM lead_public.demo
            ORDER BY geo_fips NULLS FIRST
            """
        ).fetchall()
    finally:
        con.close()
    assert rows == [
        (None, 2024, 1, "2024Q1", "9"),
        ("01001", 2024, 1, "2024Q1", "4"),
    ]


def test_a_month_fills_the_quarter(tmp_path):
    src = tmp_path / "m.csv"
    src.write_text("fips,year,month,n\n01001,2024,2,1\n")
    con = duckdb.connect()
    try:
        load_file(
            con,
            src,
            table="lead_public.months",
            source_id="months",
            release="2024",
            sha256=_sha(src),
            actor="tester",
        )
        row = con.execute(
            "SELECT spine_month, spine_quarter, time_key FROM lead_public.months"
        ).fetchone()
    finally:
        con.close()
    assert row == (2, 1, "2024Q1")


def test_a_bls_period_fills_the_month_and_a_blank_value_stays_text(tmp_path):
    src = tmp_path / "p.csv"
    src.write_text(
        "series_id,year,period,value\nWPS512,2024,M02,1.5\nWPS512,2024,M13,.\n"
    )
    con = duckdb.connect()
    try:
        load_file(
            con,
            src,
            table="lead_public.periods",
            source_id="periods",
            release="2024",
            sha256=_sha(src),
            actor="tester",
        )
        rows = con.execute(
            """
            SELECT period, spine_month, spine_quarter, time_key, value
            FROM lead_public.periods
            ORDER BY period
            """
        ).fetchall()
    finally:
        con.close()
    assert rows == [
        ("M02", 2, 1, "2024Q1", "1.5"),
        ("M13", None, None, None, "."),
    ]


def test_a_hash_mismatch_inserts_nothing(tmp_path):
    src = tmp_path / "rows.csv"
    src.write_text("fips,year,n\n01001,2024,1\n")
    con = duckdb.connect()
    try:
        with pytest.raises(HashMismatch):
            load_file(
                con,
                src,
                table="lead_public.demo",
                source_id="demo",
                release="2024",
                sha256="0" * 64,
                actor="tester",
            )
        found = con.execute(
            """
            SELECT COUNT(*) FROM information_schema.tables
            WHERE table_schema = 'lead_public' AND table_name = 'demo'
            """
        ).fetchone()[0]
        logged = con.execute("SELECT COUNT(*) FROM lead_public.load_log").fetchone()[0]
    finally:
        con.close()
    assert found == 0
    assert logged == 0


def test_html_and_a_dictionary_are_not_observation_rows(tmp_path):
    html = tmp_path / "spec.html"
    html.write_text("<!DOCTYPE html><html><body>not a table</body></html>")
    missing = tmp_path / "no-dictionary.pdf"
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                html,
                table="lead_public.spec",
                source_id="spec",
                release="v1",
                sha256=_sha(html),
                actor="tester",
            )
            == 0
        )
        manifest = _manifest(
            tmp_path,
            [
                _row("spec", "v1", "spec.html", _sha(html)),
                _row("dict_src", "v1", "no-dictionary.pdf", "abc", kind="dictionary"),
            ],
        )
        results = load_manifest_files(con, manifest, root=tmp_path, actor="tester")
    finally:
        con.close()
    by_source = {row.source_id: row for row in results}
    assert by_source["spec"].status == "skipped"
    assert by_source["spec"].reason == "not_tabular"
    assert by_source["dict_src"].status == "skipped"
    assert by_source["dict_src"].reason == "kind:dictionary"
    assert not missing.exists()


def test_a_zip_csv_an_xlsx_and_a_bls_payload_load(tmp_path):
    zpath = tmp_path / "pack.zip"
    with zipfile.ZipFile(zpath, "w") as zf:
        zf.writestr(
            "inner/places.csv",
            "LocationName,FIPS,Year,Data_Value\nAutauga,01001,2024,3\n",
        )
        zf.writestr("inner/readme.txt", "This file explains the extract.\n")
    book = tmp_path / "book.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "county"
    sheet.append(["ZCTA", "Year", "agi"])
    sheet.append(["02138", 2023, 10])
    wb.save(book)
    payload = [
        {
            "status": "REQUEST_SUCCEEDED",
            "message": [],
            "Results": {
                "series": [
                    {
                        "seriesID": "CUUR0000SAM",
                        "data": [
                            {"year": "2024", "period": "M01", "value": "500.1"},
                            {"year": "2024", "period": "M13", "value": "."},
                        ],
                    }
                ]
            },
        }
    ]
    blob = tmp_path / "cpi.json"
    blob.write_text(json.dumps(payload))
    pipe = tmp_path / "age.dat"
    pipe.write_text("GEO_ID|B01001_E001\n0500000US01001|100\n")
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="lead_public.pack",
                source_id="pack",
                release="2024",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 1
        )
        zip_row = con.execute(
            "SELECT geo_fips, spine_year, member, data_value FROM lead_public.pack"
        ).fetchone()
        assert (
            load_file(
                con,
                book,
                table="lead_public.book",
                source_id="book",
                release="2023",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        xlsx_row = con.execute(
            "SELECT geo_zcta, spine_year, member, agi FROM lead_public.book"
        ).fetchone()
        assert (
            load_file(
                con,
                blob,
                table="lead_public.bls_cpi_medical",
                source_id="bls_cpi_medical",
                release="2026",
                sha256=_sha(blob),
                actor="tester",
            )
            == 2
        )
        cpi = con.execute(
            """
            SELECT series_id, spine_year, period, value, time_key
            FROM lead_public.bls_cpi_medical
            ORDER BY period
            """
        ).fetchall()
        assert (
            load_file(
                con,
                pipe,
                table="lead_public.acs",
                source_id="acs",
                release="2024",
                sha256=_sha(pipe),
                actor="tester",
            )
            == 1
        )
        acs = con.execute("SELECT geo_id, b01001_e001 FROM lead_public.acs").fetchone()
    finally:
        con.close()
    assert zip_row == ("01001", 2024, "inner/places.csv", "3")
    assert xlsx_row == ("02138", 2023, "county", "10")
    assert cpi == [
        ("CUUR0000SAM", 2024, "M01", "500.1", "2024Q1"),
        ("CUUR0000SAM", 2024, "M13", ".", None),
    ]
    assert acs == ("0500000US01001", "100")


def test_awkward_public_csvs_stay_whole_rows(tmp_path):
    """CMS and HCUP files that DuckDB's sniffer misreads still load whole.

    A Windows-1252 apostrophe is not a reason to drop the row. A suppressed
    ``'-`` cell is text, not a quote. An HCUP export wraps codes in single
    quotes and descriptions that contain commas in double quotes.
    """
    coded = tmp_path / "aip.csv"
    coded.write_bytes(
        b'aco_id,aco_name\nA5655,"St. David\x92s Care Partners ACO, LLC"\n'
    )
    suppressed = tmp_path / "reach.csv"
    suppressed.write_text(
        "aco_id,county_name,elig_months_ad,aligned\nD0004,Danville city,'-,'-\n",
        encoding="utf-8",
    )
    sas = tmp_path / "prccsr.csv"
    sas.write_text(
        "'ICD-10-PCS','ICD-10-PCS DESCRIPTION','PRCCSR'\n"
        "'0016070',\"Bypass Cerebral Ventricle to Nasopharynx with "
        "Autologous Tissue Substitute, Open Approach\",'CNS010'\n",
        encoding="utf-8",
    )
    con = duckdb.connect()
    try:
        for path, table in (
            (coded, "aip"),
            (suppressed, "reach"),
            (sas, "prccsr"),
        ):
            load_file(
                con,
                path,
                table=table,
                source_id=table,
                release="2026",
                sha256=_sha(path),
                actor="tester",
            )
        name = con.execute("SELECT aco_name FROM lead_public.aip").fetchone()[0]
        reach = con.execute(
            "SELECT elig_months_ad, aligned FROM lead_public.reach"
        ).fetchone()
        procedure = con.execute(
            """
            SELECT icd_10_pcs, icd_10_pcs_description, prccsr
            FROM lead_public.prccsr
            """
        ).fetchone()
    finally:
        con.close()
    assert name == "St. David\u2019s Care Partners ACO, LLC"
    assert reach == ("'-", "'-")
    assert procedure == (
        "0016070",
        "Bypass Cerebral Ventricle to Nasopharynx with Autologous Tissue "
        "Substitute, Open Approach",
        "CNS010",
    )


def test_a_quoted_newline_and_a_title_banner_stay_one_table(tmp_path):
    report = tmp_path / "report.csv"
    report.write_text(
        'Title,Abstract\nOne,"line one\nline two"\nTwo,plain\n',
        encoding="utf-8",
    )
    mapping = tmp_path / "mapping.csv"
    mapping.write_bytes(
        b'"Model title",,,\n'
        b"note,,,\n"
        b",,,\n"
        b'"Diagnosis\nCode",Description,Flag,Extra\n'
        b"A0103,Typhoid pneumonia,115,1\n"
    )
    con = duckdb.connect()
    con.execute("SET preserve_insertion_order = false")
    try:
        load_file(
            con,
            report,
            table="report",
            source_id="report",
            release="2026",
            sha256=_sha(report),
            actor="tester",
        )
        load_file(
            con,
            mapping,
            table="mapping",
            source_id="mapping",
            release="2026",
            sha256=_sha(mapping),
            actor="tester",
        )
        abstracts = con.execute(
            "SELECT title, abstract FROM lead_public.report ORDER BY title"
        ).fetchall()
        mapped = con.execute(
            """
            SELECT diagnosis_code, description, flag
            FROM lead_public.mapping
            """
        ).fetchall()
    finally:
        con.close()
    assert abstracts == [("One", "line one\nline two"), ("Two", "plain")]
    assert mapped == [("A0103", "Typhoid pneumonia", "115")]


def test_a_sas_macro_beside_a_coefficient_table_is_skipped(tmp_path):
    zpath = tmp_path / "soft.zip"
    with zipfile.ZipFile(zpath, "w") as archive:
        archive.writestr(
            "V22I0ED4.TXT",
            " %MACRO V22I0ED4(AGE=, SEX=);\n %LET X=1;\n %MEND;\n",
        )
        archive.writestr(
            "C2214O5P.csv",
            "Name,Coeff,Label\nHCC1,1.25,Diabetes\n",
        )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="soft",
                source_id="soft",
                release="2026",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            "SELECT name, coeff, label, member FROM lead_public.soft"
        ).fetchone()
    finally:
        con.close()
    assert row == ("HCC1", "1.25", "Diabetes", "C2214O5P.csv")


def test_a_parallel_reader_refusal_is_loaded_on_one_thread(tmp_path, monkeypatch):
    src = tmp_path / "rows.csv"
    src.write_text("fips,year,n\n01001,2024,4\n")
    real = public_load._read_delimited
    seen: list[bool] = []

    def wrapped(con, path, *, serial=False):
        seen.append(serial)
        if len(seen) == 1:
            raise _NeedsSerial()
        return real(con, path, serial=serial)

    monkeypatch.setattr(public_load, "_read_delimited", wrapped)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="demo",
                source_id="demo",
                release="2024",
                sha256=_sha(src),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            "SELECT geo_fips, spine_year, n FROM lead_public.demo"
        ).fetchone()
    finally:
        con.close()
    assert seen == [False, True]
    assert row == ("01001", 2024, "4")


def test_an_empty_sheet_does_not_drop_the_workbook(tmp_path):
    book = tmp_path / "states.xlsx"
    wb = openpyxl.Workbook()
    blank = wb.active
    blank.title = "doc"
    sheet = wb.create_sheet("zip")
    sheet.append(["zip", "agi"])
    sheet.append(["30301", "7"])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="2005",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        row = con.execute("SELECT zip, agi FROM lead_public.irs").fetchone()
    finally:
        con.close()
    assert row == ("30301", "7")


def test_an_rxnorm_rrf_keeps_its_first_record(tmp_path):
    zpath = tmp_path / "rxnorm.zip"
    with zipfile.ZipFile(zpath, "w") as archive:
        archive.writestr(
            "Readme.txt",
            "October 05, 2026\nThis file explains the release.\n",
        )
        archive.writestr(
            "rrf/RXNCONSO.RRF",
            "38|ENG||||||829|829|38||RXNORM|BN|38|Parlodel||N|4096|\n",
        )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="rxnorm",
                source_id="rxnorm",
                release="current",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            "SELECT rxcui, tty, code, str_ FROM lead_public.rxnorm"
        ).fetchone()
    finally:
        con.close()
    assert row == ("38", "BN", "38", "Parlodel")


def test_two_rxnorm_members_keep_published_names(tmp_path):
    zpath = tmp_path / "rxnorm.zip"
    rel = "|".join(
        [
            "",
            "a1",
            "b1",
            "c1",
            "RO",
            "a2",
            "b2",
            "c2",
            "tradename_of",
            "r1",
            "s1",
            "RXNORM",
            "l",
            "g",
            "d",
            "N",
        ]
    )
    with zipfile.ZipFile(zpath, "w") as archive:
        archive.writestr(
            "rrf/RXNCONSO.RRF",
            "38|ENG||||||829|829|38||RXNORM|BN|38|Parlodel||N|4096|\n",
        )
        archive.writestr("rrf/RXNREL.RRF", rel + "|\n")
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="rxnorm",
                source_id="rxnorm",
                release="current",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 2
        )
        names = {
            row[0] for row in con.execute("DESCRIBE lead_public.rxnorm").fetchall()
        }
        conso = con.execute(
            """
            SELECT rxcui, tty, str_
            FROM lead_public.rxnorm
            WHERE member = 'rrf/RXNCONSO.RRF'
            """
        ).fetchone()
        related = con.execute(
            """
            SELECT rxcui1, rel, rxcui2
            FROM lead_public.rxnorm
            WHERE member = 'rrf/RXNREL.RRF'
            """
        ).fetchone()
    finally:
        con.close()
    assert "parlodel" not in names
    assert conso == ("38", "BN", "Parlodel")
    assert related == (None, "c1", "RO")


def test_manifest_stacks_releases_and_keeps_going_after_a_bad_hash(tmp_path):
    good = tmp_path / "cms_demo" / "2024"
    good.mkdir(parents=True)
    first = good / "a.csv"
    first.write_text("fips,year,quarter,n\n01001,2024,1,4\n")
    second_dir = tmp_path / "cms_demo" / "2025"
    second_dir.mkdir()
    second = second_dir / "b.csv"
    second.write_text("fips,year,quarter,n\n01003,2025,2,7\n")
    bad = tmp_path / "cms_demo" / "bad.csv"
    bad.write_text("fips,year,n\n01001,2024,1\n")
    manifest = _manifest(
        tmp_path,
        [
            _row("cms_demo", "2024", "cms_demo/2024/a.csv", _sha(first)),
            _row("cms_demo", "2025", "cms_demo/2025/b.csv", _sha(second)),
            _row("cms_demo", "bad", "cms_demo/bad.csv", "f" * 64),
            _row("cms_gap", "2024", "cms_gap/missing.csv", ""),
        ],
    )
    con = duckdb.connect()
    try:
        results = load_manifest_files(con, manifest, root=tmp_path, actor="tester")
        stacked = con.execute(
            """
            SELECT release, geo_fips, time_key
            FROM lead_public.cms_demo
            ORDER BY release
            """
        ).fetchall()
    finally:
        con.close()
    by_key = {(row.source_id, row.release): row.status for row in results}
    assert by_key == {
        ("cms_demo", "2024"): "loaded",
        ("cms_demo", "2025"): "loaded",
        ("cms_demo", "bad"): "mismatch",
        ("cms_gap", "2024"): "gap",
    }
    assert stacked == [("2024", "01001", "2024Q1"), ("2025", "01003", "2025Q2")]


def test_a_wrapped_label_row_is_not_the_pfs_header(tmp_path):
    """PPRRVU puts a dense row of wrapped labels one line above HCPCS."""
    src = tmp_path / "PPRRVU.csv"
    src.write_text(
        ",,2025 National Physician Fee Schedule Relative Value File,,,,,,,,\n"
        ",,CPT codes and descriptions only are copyright 2024,,,,,,,,\n"
        ",,,,,,,,,,,\n"
        ",,,,NOT USED,,,,,PHYSICIAN,,,\n"
        ",,,STATUS,MEDICARE,WORK,NON-FAC,NA,FACILITY,NA,MP,TOTAL\n"
        "HCPCS,MOD,DESCRIPTION,CODE,PAYMENT,RVU,PE RVU,INDICATOR,PE RVU,INDICATOR,RVU\n"
        "0001F,,Heart failure composite,I,,0.00,0.00,,0.00,,0.00\n",
        encoding="utf-8",
    )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="pfs",
                source_id="pfs",
                release="2025",
                sha256=_sha(src),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            "SELECT hcpcs, mod, description, code FROM lead_public.pfs"
        ).fetchone()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'pfs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert row == ("0001F", None, "Heart failure composite", "I")
    assert "status" not in columns
    assert "medicare" not in columns


def test_a_year_header_below_a_title_is_kept(tmp_path):
    src = tmp_path / "NHE60-34.csv"
    src.write_text(
        "NATIONAL HEALTH EXPENDITURES BY TYPE OF SERVICE,,,,,,,,\n"
        ",Expenditure Amount (Millions),,,,,,,\n"
        "Type of Service,Source of Funds,1960,1961,1962,1963,1964,1965\n"
        ",,Historical,Historical,Historical,Historical,Historical,Historical\n"
        ",Amount in Billions,,,,,,\n"
        "National Health Expenditures,Total Expenditures,27122,29060,31765,34558,38245,41627\n",
        encoding="utf-8",
    )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="nhe",
                source_id="nhe",
                release="2026",
                sha256=_sha(src),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            """
            SELECT type_of_service, source_of_funds, _1960
            FROM lead_public.nhe
            """
        ).fetchone()
    finally:
        con.close()
    assert row == ("National Health Expenditures", "Total Expenditures", "27122")


def test_an_excel_title_row_is_not_the_header(tmp_path):
    book = tmp_path / "ccsr.xlsx"
    wb = openpyxl.Workbook()
    prose = wb.active
    prose.title = "Table_of_Contents"
    prose.append(["This Excel file provides searchable lists of the CCSR categories"])
    prose.append(["Table of Contents"])
    mapping = wb.create_sheet("DX_to_CCSR_Mapping")
    mapping.append(["Clinical Classifications Software Refined, version 2026.1"])
    mapping.append(
        [
            "ICD-10-CM Code",
            "ICD-10-CM Code Description",
            "CCSR Category",
            "CCSR Category Description",
        ]
    )
    mapping.append(
        [
            "A000",
            "Cholera due to Vibrio cholerae 01, biovar cholerae",
            "DIG001",
            "Intestinal infection",
        ]
    )
    names = wb.create_sheet("Naming_Conventions")
    names.append(["Naming conventions for the body-system abbreviations"])
    names.append(["Body System", "Abbreviation"])
    names.append(["Diseases of the blood", "BLD"])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="ccsr",
                source_id="ccsr",
                release="2026",
                sha256=_sha(book),
                actor="tester",
            )
            == 2
        )
        coded = con.execute(
            """
            SELECT icd_10_cm_code, icd_10_cm_code_description, ccsr_category
            FROM lead_public.ccsr
            WHERE icd_10_cm_code = 'A000'
            """
        ).fetchone()
        named = con.execute(
            """
            SELECT body_system, abbreviation
            FROM lead_public.ccsr
            WHERE abbreviation = 'BLD'
            """
        ).fetchone()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'ccsr'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert coded == (
        "A000",
        "Cholera due to Vibrio cholerae 01, biovar cholerae",
        "DIG001",
    )
    assert named == ("Diseases of the blood", "BLD")
    assert not any("searchable" in name or "provides" in name for name in columns)


def test_an_irs_title_block_drops_the_column_index_row(tmp_path):
    book = tmp_path / "county.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "county income"
    sheet.append(["ALASKA"])
    sheet.append(
        [
            "Individual Income Tax Returns: Selected Income Items by State and County, Tax Year 2009"
        ]
    )
    sheet.append(["(Money amounts are in thousands of dollars)"])
    sheet.append(["Source: IRS Individual Master File"])
    sheet.append([None])
    sheet.append(
        [
            "State Code",
            "County Code",
            "County Name",
            "Number of returns",
            "Adjusted gross income (AGI)",
        ]
    )
    sheet.append([-1, -2, -3, -4, -5])
    sheet.append([2, 0, "ALASKA", 293702, 17595406])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="2009",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            """
            SELECT state_code, county_code, county_name, adjusted_gross_income_agi
            FROM lead_public.irs
            """
        ).fetchone()
    finally:
        con.close()
    assert row == ("2", "0", "ALASKA", "17595406")


def test_a_zip_keeps_each_members_own_header(tmp_path):
    zpath = tmp_path / "rvu.zip"
    with zipfile.ZipFile(zpath, "w") as archive:
        archive.writestr(
            "GPCI2025.csv",
            "ADDENDUM E. GEOGRAPHIC PRACTICE COST INDICES,,,,,,\n"
            ",,,,,,\n"
            "Medicare Administrative Contractor (MAC),State,Locality Number,"
            "Locality Name,Work GPCI,PE GPCI,MP GPCI\n"
            "10112,AL,00,ALABAMA,1,0.869,0.575\n",
        )
        archive.writestr(
            "OPPSCAP.csv",
            "HCPCS,MOD,PROCSTAT,CARRIER,LOCALITY,FACILITY PRICE,NON-FACILTY PRICE\n"
            "0633T,TC,C,01112,05,150.69,150.69\n",
        )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="pfs",
                source_id="pfs",
                release="2025",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 2
        )
        gpci = con.execute(
            """
            SELECT medicare_administrative_contractor_mac, state, locality_name
            FROM lead_public.pfs
            WHERE member = 'GPCI2025.csv'
            """
        ).fetchone()
        opps = con.execute(
            """
            SELECT hcpcs, mod, procstat, non_facilty_price
            FROM lead_public.pfs
            WHERE member = 'OPPSCAP.csv'
            """
        ).fetchone()
    finally:
        con.close()
    assert gpci == ("10112", "AL", "ALABAMA")
    assert opps == ("0633T", "TC", "C", "150.69")


def test_icd_fixed_width_files_keep_commas_in_the_description(tmp_path):
    zpath = tmp_path / "icd.zip"
    short = "Cholera due to Vibrio cholerae 01, biovar cholerae".ljust(60)
    long = "Cholera due to Vibrio cholerae 01, biovar cholerae"
    order = (
        f"00002 {'A000'.ljust(7)} 1 {short}{long}\n"
        f"00001 {'A00'.ljust(7)} 0 {'Cholera'.ljust(60)}Cholera\n"
    )
    with zipfile.ZipFile(zpath, "w") as archive:
        archive.writestr(
            "icd10cm_codes_2026.txt",
            "A000    Cholera due to Vibrio cholerae 01, biovar cholerae\n"
            "C441021 Unspecified malignant neoplasm of skin\n",
        )
        archive.writestr("icd10cm_order_2026.txt", order)
        archive.writestr(
            "icd10cm_codes_addenda_2026.txt",
            "Delete:      B880    Other acariasis\n"
            "Add:         B8801   Infestation by Demodex mites\n"
            "Add:         C50A0   Malignant inflammatory neoplasm of unspecified breast\n"
            "\n"
            "    28 deletions\n",
        )
        archive.writestr(
            "icd10cm_order_addenda_2026.txt",
            "Add:         1 B8801   Infestation by Demodex mites"
            "                                 Infestation by Demodex mites\n"
            "Delete:      0 B880    Other acariasis"
            "                                              Other acariasis\n",
        )
        archive.writestr("icd10cmCodesFile.pdf", b"%PDF-1.4 not a table")
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                zpath,
                table="icd",
                source_id="icd",
                release="2026",
                sha256=_sha(zpath),
                actor="tester",
            )
            == 9
        )
        coded = con.execute(
            """
            SELECT code, description FROM lead_public.icd
            WHERE member = 'icd10cm_codes_2026.txt'
            ORDER BY code
            """
        ).fetchall()
        ordered = con.execute(
            """
            SELECT order_number, code, header_flag, short_description
            FROM lead_public.icd
            WHERE member = 'icd10cm_order_2026.txt' AND code = 'A000'
            """
        ).fetchone()
        added = con.execute(
            """
            SELECT action, code, description FROM lead_public.icd
            WHERE member = 'icd10cm_codes_addenda_2026.txt'
            ORDER BY code
            """
        ).fetchall()
        revised = con.execute(
            """
            SELECT action, header_flag, code, short_description, long_description
            FROM lead_public.icd
            WHERE member = 'icd10cm_order_addenda_2026.txt' AND action = 'Add'
            """
        ).fetchone()
    finally:
        con.close()
    assert coded == [
        ("A000", "Cholera due to Vibrio cholerae 01, biovar cholerae"),
        ("C441021", "Unspecified malignant neoplasm of skin"),
    ]
    assert ordered == (
        "00002",
        "A000",
        "1",
        "Cholera due to Vibrio cholerae 01, biovar cholerae",
    )
    assert added == [
        ("Delete", "B880", "Other acariasis"),
        ("Add", "B8801", "Infestation by Demodex mites"),
        ("Add", "C50A0", "Malignant inflammatory neoplasm of unspecified breast"),
    ]
    assert revised == (
        "Add",
        "1",
        "B8801",
        "Infestation by Demodex mites",
        "Infestation by Demodex mites",
    )


def test_a_group_label_stays_on_the_following_years(tmp_path):
    """CMS program statistics put Aged above a run of year rows."""
    book = tmp_path / "inpatient.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "MDCR INPT HOSP 1"
    sheet.append([None, "All Medicare Inpatient Hospitals"])
    sheet.append([None, "Calendar Years 2019-2024"])
    sheet.append(
        [
            "Type of Entitlement and Calendar Year",
            "Total Enrollees",
            "Discharges",
            "Payments",
        ]
    )
    sheet.append(["BLANK"])
    sheet.append(["All Beneficiaries"])
    sheet.append(["2019", 100, 10, 5])
    sheet.append(["2020", 110, 11, 6])
    sheet.append(["Aged Beneficiaries"])
    sheet.append(["2019", 80, 8, 4])
    sheet.append(["NOTE: Persons with coinsurance are counted once."])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="cps",
                source_id="cps",
                release="2024",
                sha256=_sha(book),
                actor="tester",
            )
            == 3
        )
        rows = con.execute(
            """
            SELECT section, type_of_entitlement_and_calendar_year, total_enrollees
            FROM lead_public.cps
            ORDER BY section, type_of_entitlement_and_calendar_year
            """
        ).fetchall()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'cps'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert rows == [
        ("Aged Beneficiaries", "2019", "80"),
        ("All Beneficiaries", "2019", "100"),
        ("All Beneficiaries", "2020", "110"),
    ]
    assert not any(name.startswith("column") or "unnamed" in name for name in columns)


def test_a_group_label_after_the_first_row_keeps_one_width(tmp_path):
    """Some IRS sheets print the total, then a one-cell group, then the counties."""
    book = tmp_path / "county.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "county"
    sheet.append(["State", "Returns", "Income"])
    sheet.append(["AL", 10, 20])
    sheet.append(["Aged"])
    sheet.append(["AL", 4, 8])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="2002",
                sha256=_sha(book),
                actor="tester",
            )
            == 2
        )
        rows = con.execute(
            """
            SELECT section, state, returns
            FROM lead_public.irs
            ORDER BY returns
            """
        ).fetchall()
    finally:
        con.close()
    assert rows == [("", "AL", "10"), ("Aged", "AL", "4")]


def test_a_blank_header_cell_keeps_the_code_under_it(tmp_path):
    """1990 IRS prints CODES once and leaves the county-code cell blank."""
    book = tmp_path / "ALABAMA90ci.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "A"
    sheet.append(["TAX YEAR 1990"])
    sheet.append([None, None, None, None, "NUMBER OF", "NUMBER OF", "ADJUSTED"])
    sheet.append(
        [None, "CODES", None, "COUNTY", "RETURNS", "EXEMPTIONS", "GROSS INCOME"]
    )
    sheet.append([None, 1, 1, "AUTAUGA", 12883, 31491, 348492])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="1990",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            """
            SELECT codes, codes_2, county, number_of_returns
            FROM lead_public.irs
            """
        ).fetchone()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'irs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert row == ("1", "1", "AUTAUGA", "12883")
    assert not any(name.startswith("column") for name in columns)


def test_a_headerless_icd_map_keeps_its_first_code(tmp_path):
    src = tmp_path / "F2825T1N.TXT"
    src.write_text(
        "A0104\t92\t\nA0105\t92\t\nB377\t2\t\nB377\t6\tD\n",
        encoding="utf-8",
    )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="hcc",
                source_id="hcc",
                release="2026",
                sha256=_sha(src),
                actor="tester",
            )
            == 4
        )
        rows = con.execute(
            """
            SELECT icd10, cc, duplicate_cc
            FROM lead_public.hcc
            ORDER BY icd10, cc
            """
        ).fetchall()
    finally:
        con.close()
    assert rows == [
        ("A0104", "92", None),
        ("A0105", "92", None),
        ("B377", "2", None),
        ("B377", "6", "D"),
    ]


def test_a_backtick_in_a_blank_header_is_not_a_column(tmp_path):
    """1998 Georgia has one stray backtick past the last named field."""
    book = tmp_path / "georgia.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "Sheet 1"
    sheet.append(["State", "Returns", None])
    sheet.append(["AL", 10, "`"])
    sheet.append(["GA", 12, None])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="1998",
                sha256=_sha(book),
                actor="tester",
            )
            == 2
        )
        rows = con.execute(
            "SELECT state, returns FROM lead_public.irs ORDER BY state"
        ).fetchall()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'irs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert rows == [("AL", "10"), ("GA", "12")]
    assert not any(name.startswith("column") or name == "col" for name in columns)


def test_an_empty_trailing_header_is_not_a_column(tmp_path):
    src = tmp_path / "sites.csv"
    src.write_text("site_name,site_id,\nAlpha,1,\nBeta,2,\n", encoding="utf-8")
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="sites",
                source_id="sites",
                release="2026",
                sha256=_sha(src),
                actor="tester",
            )
            == 2
        )
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'sites'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert "site_name" in columns
    assert not any(name.startswith("column") or name == "col" for name in columns)


def test_a_dropdown_sheet_is_not_a_table(tmp_path):
    book = tmp_path / "CMS165.xlsx"
    wb = openpyxl.Workbook()
    dropdown = wb.active
    dropdown.title = "DropdownOptions"
    dropdown.append(
        [
            "Types of TRNs (please do not delete this sheet -- it is used to define the dropdown options.)",
            "Measure Section",
            "Source of Change",
        ]
    )
    dropdown.append(["Header", "eCQM Title", "Annual Update"])
    dropdown.append(["Logic", "CMS ID", "Measure Lead"])
    notes = wb.create_sheet("TRNs")
    notes.append(["CMS165v15 - Controlling High Blood Pressure"])
    notes.append(
        [
            "Technical Release Notes",
            "Type of TRN",
            "Measure Section",
            "Source of Change",
        ]
    )
    notes.append(
        ["Updated the version.", "Header", "eCQM Version Number", "Annual Update"]
    )
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="ecqi",
                source_id="ecqi",
                release="v15",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            """
            SELECT technical_release_notes, type_of_trn
            FROM lead_public.ecqi
            """
        ).fetchone()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'ecqi'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert row == ("Updated the version.", "Header")
    assert not any(
        "do_not_delete" in name or "types_of_trns" in name for name in columns
    )


def test_a_suppressed_zip_row_does_not_become_the_header(tmp_path):
    """IRS 2007 prints ``.`` for a withheld count. That row is data."""
    src = tmp_path / "zipcode07.csv"
    src.write_text(
        '"state","ZIPCODE","agi_class","n1","mars2"\n'
        '"AL",35001,1,116,.\n'
        '"AL",35004,1,916,122\n',
        encoding="utf-8",
    )
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                src,
                table="irs",
                source_id="irs",
                release="2007",
                sha256=_sha(src),
                actor="tester",
            )
            == 2
        )
        rows = con.execute(
            """
            SELECT state, zipcode, agi_class, n1, mars2
            FROM lead_public.irs
            ORDER BY zipcode
            """
        ).fetchall()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'irs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert rows == [
        ("AL", "35001", "1", "116", "."),
        ("AL", "35004", "1", "916", "122"),
    ]
    assert not any(
        name in {"al", "col"} or name.startswith("column") or name.startswith("_")
        for name in columns
    )


def test_a_parenthesized_index_row_is_not_data(tmp_path):
    """1998 IRS puts ``(1), (2)`` under the names, and ``C`` means Schedule C."""
    book = tmp_path / "98zp.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "Sheet 1"
    sheet.append(["ALABAMA"])
    sheet.append(["", "Schedule C", "", "Schedule F", ""])
    sheet.append(["Income Size", "Number of", "Schedules", "Number of", "Schedules"])
    sheet.append(["Income Size", "Returns", "C", "Returns", "F"])
    sheet.append(["", "(1)", "(2)", "(3)", "(4)"])
    sheet.append(["ALABAMA", 10, 30, 40, 50])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="irs",
                source_id="irs",
                release="1998",
                sha256=_sha(book),
                actor="tester",
            )
            == 1
        )
        row = con.execute(
            """
            SELECT income_size, schedules_c, schedules_f
            FROM lead_public.irs
            """
        ).fetchone()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'irs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert row == ("ALABAMA", "30", "50")
    assert "c" not in columns
    assert "f" not in columns
    assert not any(name.startswith("column") for name in columns)


def test_an_action_code_key_keeps_the_first_letter(tmp_path):
    """The HCPCS key is ``A / Add`` with a title and no header row."""
    book = tmp_path / "hcpcs.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "Action Code Key"
    sheet.append(["Action Codes"])
    sheet.append(["A", "Add"])
    sheet.append(["B", "Coverage Change"])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="hcpcs",
                source_id="hcpcs",
                release="2026",
                sha256=_sha(book),
                actor="tester",
            )
            == 2
        )
        rows = con.execute(
            """
            SELECT action_code, action_label
            FROM lead_public.hcpcs
            ORDER BY action_code
            """
        ).fetchall()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'hcpcs'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert rows == [("A", "Add"), ("B", "Coverage Change")]
    assert "a" not in columns
    assert "add" not in columns


def test_a_blank_leading_header_is_the_row_stub(tmp_path):
    """CMS home health leaves the stub cell blank and puts Total under it."""
    book = tmp_path / "hha.xlsx"
    wb = openpyxl.Workbook()
    sheet = wb.active
    sheet.title = "MDCR HHA 6"
    sheet.append([None, "Medicare Home Health Agency"])
    sheet.append([None, "Persons", "Visits"])
    sheet.append(["BLANK"])
    sheet.append(["Number of Service Visits"])
    sheet.append(["Total", 10, 20])
    sheet.append(["1-9", 4, 8])
    wb.save(book)
    con = duckdb.connect()
    try:
        assert (
            load_file(
                con,
                book,
                table="hha",
                source_id="hha",
                release="2024",
                sha256=_sha(book),
                actor="tester",
            )
            == 2
        )
        rows = con.execute(
            """
            SELECT section, category, persons, visits
            FROM lead_public.hha
            ORDER BY category
            """
        ).fetchall()
        columns = [
            name
            for (name,) in con.execute(
                """
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'hha'
                """
            ).fetchall()
        ]
    finally:
        con.close()
    assert rows == [
        ("Number of Service Visits", "1-9", "4", "8"),
        ("Number of Service Visits", "Total", "10", "20"),
    ]
    assert not any(name.startswith("column") or name == "col" for name in columns)


def test_the_log_ddl_has_no_catalog_token():
    assert "aco" not in LOAD_LOG_DDL.casefold()


def test_cli_loads_a_manifest_and_does_not_fetch(tmp_path):
    src_dir = tmp_path / "demo"
    src_dir.mkdir()
    src = src_dir / "rows.csv"
    src.write_text("hsa,year,quarter,n\n12,2024,4,1\n")
    manifest = _manifest(
        tmp_path,
        [_row("demo", "2024", "demo/rows.csv", _sha(src))],
    )
    db = tmp_path / "public.duckdb"
    result = runner.invoke(
        app,
        [
            "lead",
            "load",
            "--db",
            str(db),
            "--manifest",
            str(manifest),
            "--root",
            str(tmp_path),
            "--actor",
            "tester",
        ],
    )
    assert result.exit_code == 0, result.output
    assert "loaded\t1" in result.output
    con = duckdb.connect(str(db), read_only=True)
    try:
        row = con.execute("SELECT geo_hsa, time_key FROM lead_public.demo").fetchone()
    finally:
        con.close()
    assert row == ("12", "2024Q4")