35 lines
979 B
Python
35 lines
979 B
Python
"""Tests for cms.ingest_log — provenance table writer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
|
|
from cms import ingest_log
|
|
|
|
|
|
def test_log_ingest_appends_row(tmp_path):
|
|
src = tmp_path / "f.csv"
|
|
src.write_text("a,b\n1,2\n")
|
|
con = duckdb.connect()
|
|
ingest_log.log_ingest(
|
|
con,
|
|
module="pfs",
|
|
table_name="pfs.rvu_proposed",
|
|
rows=2,
|
|
source_file=str(src),
|
|
rule_id="CMS-1832-P",
|
|
fr_citation="90 FR 1",
|
|
pincite_key="2KVJ2HKX",
|
|
run_id=ingest_log.new_run_id(),
|
|
)
|
|
row = con.execute(
|
|
"SELECT module, table_name, rows, rule_id, length(sha256) FROM cms.ingest_log"
|
|
).fetchone()
|
|
assert row == ("pfs", "pfs.rvu_proposed", 2, "CMS-1832-P", 64)
|
|
|
|
|
|
def test_missing_source_file_logs_empty_hash():
|
|
con = duckdb.connect()
|
|
ingest_log.log_ingest(con, module="pfs", table_name="t", rows=0)
|
|
assert con.execute("SELECT sha256 FROM cms.ingest_log").fetchone()[0] == ""
|