308 lines
11 KiB
Python
308 lines
11 KiB
Python
"""Tests for pfs.nprm — NPRM Addendum B loader → pfs.rvu_proposed (#592, #612)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from pfs import nprm
|
|
|
|
|
|
def _fixture(tmp_path, name="CY2026_NPRM_Addendum_B.csv"):
|
|
p = tmp_path / name
|
|
p.write_text(
|
|
"HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
|
|
"NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
|
|
"99213,,Office visit est,A,1.3,1.5,0.55,0.1\n"
|
|
"0001A,,Admin covid,X,0.0,0.0,0.0,0.0\n"
|
|
)
|
|
return p
|
|
|
|
|
|
def _fixture_2027(tmp_path):
|
|
"""CY2027 partition with a distinct HCPCS set so the two are distinguishable."""
|
|
p = tmp_path / "CY2027_NPRM_Addendum_B.csv"
|
|
p.write_text(
|
|
"HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
|
|
"NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
|
|
"99214,,Office visit est L4,A,1.9,2.1,0.75,0.15\n"
|
|
"0002A,,Admin covid dose 2,X,0.0,0.0,0.0,0.0\n"
|
|
)
|
|
return p
|
|
|
|
|
|
def _patch_discovery(monkeypatch, mapping):
|
|
"""monkeypatch._discover_addendum_b(tag) -> path, from a {tag: path} dict."""
|
|
|
|
def _fake(tag):
|
|
try:
|
|
return mapping[tag]
|
|
except KeyError:
|
|
raise FileNotFoundError(
|
|
f"No Addendum B attachment found among bib items tagged {tag!r}"
|
|
) from None
|
|
|
|
monkeypatch.setattr(nprm, "_discover_addendum_b", _fake)
|
|
|
|
|
|
def test_load_rvu_proposed_loads_and_reloads(tmp_path, monkeypatch):
|
|
path = _fixture(tmp_path)
|
|
_patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
out = nprm.load_rvu_proposed(con, year=2026)
|
|
assert out["rows"] == 2
|
|
assert out["cms_rule_id"] == "CMS-1832-P"
|
|
row = con.execute(
|
|
"SELECT hcpcs, work_rvu, cms_rule_id FROM pfs.rvu_proposed WHERE hcpcs='99213'"
|
|
).fetchone()
|
|
assert row == ("99213", 1.3, "CMS-1832-P")
|
|
nprm.load_rvu_proposed(con, year=2026) # idempotent delete-and-reload
|
|
assert con.execute("SELECT count(*) FROM pfs.rvu_proposed").fetchone()[0] == 2
|
|
|
|
|
|
def test_nprm_sources_covers_both_years():
|
|
years = {source[0] for source in nprm.NPRM_SOURCES}
|
|
assert years == {2026, 2027}
|
|
|
|
|
|
def test_load_rvu_proposed_two_partitions_side_by_side(tmp_path, monkeypatch):
|
|
"""Loading both years leaves both rule partitions intact side by side."""
|
|
path_2026 = _fixture(tmp_path)
|
|
path_2027 = _fixture_2027(tmp_path)
|
|
_patch_discovery(
|
|
monkeypatch,
|
|
{"sup:2026_PFS_NPRM": path_2026, "sup:2027_PFS_NPRM": path_2027},
|
|
)
|
|
con = duckdb.connect()
|
|
nprm.load_rvu_proposed(con, year=2026)
|
|
nprm.load_rvu_proposed(con, year=2027)
|
|
|
|
counts = dict(
|
|
con.execute(
|
|
"SELECT cms_rule_id, count(*) FROM pfs.rvu_proposed GROUP BY 1"
|
|
).fetchall()
|
|
)
|
|
assert counts == {"CMS-1832-P": 2, "CMS-1848-P": 2}
|
|
|
|
|
|
def test_load_rvu_proposed_reload_one_leaves_other_intact(tmp_path, monkeypatch):
|
|
"""Reloading the 2026 partition doesn't touch the 2027 partition's rows."""
|
|
path_2026 = _fixture(tmp_path)
|
|
path_2027 = _fixture_2027(tmp_path)
|
|
_patch_discovery(
|
|
monkeypatch,
|
|
{"sup:2026_PFS_NPRM": path_2026, "sup:2027_PFS_NPRM": path_2027},
|
|
)
|
|
con = duckdb.connect()
|
|
nprm.load_rvu_proposed(con, year=2026)
|
|
nprm.load_rvu_proposed(con, year=2027)
|
|
nprm.load_rvu_proposed(con, year=2026) # reload just 2026
|
|
|
|
counts = dict(
|
|
con.execute(
|
|
"SELECT cms_rule_id, count(*) FROM pfs.rvu_proposed GROUP BY 1"
|
|
).fetchall()
|
|
)
|
|
assert counts == {"CMS-1832-P": 2, "CMS-1848-P": 2}
|
|
row_2027 = con.execute(
|
|
"SELECT hcpcs, work_rvu FROM pfs.rvu_proposed WHERE cms_rule_id='CMS-1848-P' "
|
|
"AND hcpcs='99214'"
|
|
).fetchone()
|
|
assert row_2027 == ("99214", 1.9)
|
|
|
|
|
|
def test_load_rvu_proposed_missing_tag_raises_naming_it(tmp_path, monkeypatch):
|
|
"""Discovery failure names the missing tag in the error."""
|
|
_patch_discovery(monkeypatch, {}) # no tags registered at all
|
|
con = duckdb.connect()
|
|
with pytest.raises(FileNotFoundError, match="sup:2026_PFS_NPRM"):
|
|
nprm.load_rvu_proposed(con, year=2026)
|
|
|
|
|
|
def test_load_rvu_proposed_unknown_year_raises():
|
|
con = duckdb.connect()
|
|
with pytest.raises(ValueError, match="2099"):
|
|
nprm.load_rvu_proposed(con, year=2099)
|
|
|
|
|
|
def test_load_rvu_proposed_drops_and_recreates_stale_schema(tmp_path, monkeypatch):
|
|
"""A pre-existing table with old column names is dropped and rebuilt.
|
|
|
|
Regression guard for the pre-#612 table (columns ``modifier`` /
|
|
``nonfac_pe_rvu`` instead of ``mod`` / ``non_fac_pe_rvu``) — CREATE
|
|
TABLE IF NOT EXISTS alone won't fix a live mismatched table.
|
|
"""
|
|
path = _fixture(tmp_path)
|
|
_patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
|
|
con.execute(
|
|
"CREATE TABLE pfs.rvu_proposed ("
|
|
"hcpcs VARCHAR, modifier VARCHAR, description VARCHAR, "
|
|
"status_code VARCHAR, work_rvu DOUBLE, nonfac_pe_rvu DOUBLE, "
|
|
"fac_pe_rvu DOUBLE, mp_rvu DOUBLE, cms_rule_id VARCHAR)"
|
|
)
|
|
out = nprm.load_rvu_proposed(con, year=2026)
|
|
assert out["rows"] == 2
|
|
cols = [r[0] for r in con.execute("DESCRIBE pfs.rvu_proposed").fetchall()]
|
|
assert cols == nprm._TARGET_COLUMNS
|
|
|
|
|
|
def _dirty_fixture(tmp_path):
|
|
"""Real-file quirks: padded cells, "NA" markers, a footnote row.
|
|
|
|
Mirrors what the actual CY2026 Addendum B workbook contains — this
|
|
combination broke silently on the real file (see task-5-report.md)
|
|
and was fixed via manual smoke-testing rather than a fixture test,
|
|
so these rows lock the fix in.
|
|
"""
|
|
p = tmp_path / "CY2026_NPRM_Addendum_B_dirty.csv"
|
|
p.write_text(
|
|
"HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
|
|
"NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
|
|
" 99213 , ,Office visit est,A, 1.30 , 1.50 , 0.55 , 0.10 \n"
|
|
"99214,,Office visit NA test,A,NA,NA,NA,NA\n"
|
|
"1 CPT codes and descriptors only are copyright 2025 American "
|
|
"Medical Association. All Rights Reserved.,,,,,,,\n"
|
|
)
|
|
return p
|
|
|
|
|
|
def test_load_rvu_proposed_strips_padding(tmp_path, monkeypatch):
|
|
"""Whitespace-padded HCPCS/numeric cells load clean, not null."""
|
|
path = _dirty_fixture(tmp_path)
|
|
_patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
nprm.load_rvu_proposed(con, year=2026)
|
|
row = con.execute(
|
|
"SELECT hcpcs, description, work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu "
|
|
"FROM pfs.rvu_proposed WHERE hcpcs='99213'"
|
|
).fetchone()
|
|
assert row == ("99213", "Office visit est", 1.3, 1.5, 0.55, 0.1)
|
|
|
|
|
|
def test_load_rvu_proposed_na_marker_becomes_null(tmp_path, monkeypatch):
|
|
""" "NA" in a numeric cell loads as SQL NULL, not a cast error."""
|
|
path = _dirty_fixture(tmp_path)
|
|
_patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
nprm.load_rvu_proposed(con, year=2026)
|
|
row = con.execute(
|
|
"SELECT work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu "
|
|
"FROM pfs.rvu_proposed WHERE hcpcs='99214'"
|
|
).fetchone()
|
|
assert row == (None, None, None, None)
|
|
|
|
|
|
def test_load_rvu_proposed_drops_footnote_rows(tmp_path, monkeypatch):
|
|
"""A footnote row (first cell = long sentence) is not a HCPCS code."""
|
|
path = _dirty_fixture(tmp_path)
|
|
_patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
out = nprm.load_rvu_proposed(con, year=2026)
|
|
assert out["rows"] == 2 # only 99213 and 99214 — footnote row excluded
|
|
assert con.execute("SELECT count(*) FROM pfs.rvu_proposed").fetchone()[0] == 2
|
|
footnote = con.execute(
|
|
"SELECT count(*) FROM pfs.rvu_proposed WHERE hcpcs LIKE '1 CPT%'"
|
|
).fetchone()[0]
|
|
assert footnote == 0
|
|
|
|
|
|
# ── load_gpci_proposed (#629) ──────────────────────────────────────
|
|
|
|
|
|
def _gpci_fixture_2027(tmp_path):
|
|
"""CY2027 Addendum E layout: title row, header row, data, footnote."""
|
|
p = tmp_path / "Addendum_E_CY2027.csv"
|
|
p.write_text(
|
|
"ADDENDUM E. CY 2027 GEOGRAPHIC PRACTICE COST INDICES,,,,,,\n"
|
|
"Medicare Administrative Contractor,State,Locality Number,Locality Name,"
|
|
"2027 PW GPCI (without 1.0 Floor),2027 PE GPCI,2027 MP GPCI\n"
|
|
"10112,AL,00,ALABAMA,0.975,0.88,0.557\n"
|
|
"02102,AK,01,ALASKA*,1.5,1.049,0.511\n"
|
|
"*Note: localities marked with an asterisk...,,,,,,\n"
|
|
)
|
|
return p
|
|
|
|
|
|
def _gpci_fixture_2026_wide(tmp_path):
|
|
"""CY2026 Addendum E layout: 2025 baseline + 2026 and 2027 proposed."""
|
|
p = tmp_path / "Addendum_E_CY2026.csv"
|
|
p.write_text(
|
|
"ADDENDUM E. PROPOSED CY 2026 GPCIs,,,,,,,,,,,,\n"
|
|
"Medicare Administrative Contractor,State,Locality Number,Locality Name,"
|
|
"2025 PW GPCI (with 1.0 Floor),2025 PE GPCI,2025 MP GPCI,"
|
|
"2026 PW GPCI (without 1.0 Floor),2026 PE GPCI,2026 MP GPCI,"
|
|
"2027 PW GPCI (without 1.0 Floor),2027 PE GPCI,2027 MP GPCI\n"
|
|
"10112,AL,00,ALABAMA,1,0.869,0.575,0.988,0.875,0.566,0.975,0.88,0.557\n"
|
|
"02102,AK,01,ALASKA*,1.5,1.081,0.592,1.5,1.065,0.551,1.5,1.049,0.511\n"
|
|
)
|
|
return p
|
|
|
|
|
|
def _patch_gpci_discovery(monkeypatch, mapping):
|
|
def _fake(tag):
|
|
try:
|
|
return mapping[tag]
|
|
except KeyError:
|
|
raise FileNotFoundError(
|
|
f"No Addendum E attachment found among bib items tagged {tag!r}"
|
|
) from None
|
|
|
|
monkeypatch.setattr(nprm, "_discover_addendum_e", _fake)
|
|
|
|
|
|
def test_load_gpci_proposed_cy2027_single_year(tmp_path, monkeypatch):
|
|
path = _gpci_fixture_2027(tmp_path)
|
|
_patch_gpci_discovery(monkeypatch, {"sup:2027_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
out = nprm.load_gpci_proposed(con, year=2027)
|
|
assert out["rows"] == 2
|
|
assert out["cms_rule_id"] == "CMS-1848-P"
|
|
rows = con.execute(
|
|
"SELECT mac, state, locality, locality_name, work_gpci, pe_gpci, "
|
|
"mp_gpci, gpci_year, cms_rule_id FROM pfs.gpci_proposed ORDER BY mac"
|
|
).fetchall()
|
|
assert rows == [
|
|
("02102", "AK", "01", "ALASKA*", 1.5, 1.049, 0.511, 2027, "CMS-1848-P"),
|
|
("10112", "AL", "00", "ALABAMA", 0.975, 0.88, 0.557, 2027, "CMS-1848-P"),
|
|
]
|
|
|
|
|
|
def test_load_gpci_proposed_cy2026_wide_melts_two_years(tmp_path, monkeypatch):
|
|
path = _gpci_fixture_2026_wide(tmp_path)
|
|
_patch_gpci_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
|
|
con = duckdb.connect()
|
|
out = nprm.load_gpci_proposed(con, year=2026)
|
|
# 2 localities x 2 proposed years (2026, 2027) — the 2025 baseline
|
|
# restatement is skipped.
|
|
assert out["rows"] == 4
|
|
years = con.execute(
|
|
"SELECT DISTINCT gpci_year FROM pfs.gpci_proposed ORDER BY gpci_year"
|
|
).fetchall()
|
|
assert years == [(2026,), (2027,)]
|
|
pe = con.execute(
|
|
"SELECT gpci_year, pe_gpci FROM pfs.gpci_proposed "
|
|
"WHERE state='AL' ORDER BY gpci_year"
|
|
).fetchall()
|
|
assert pe == [(2026, 0.875), (2027, 0.88)]
|
|
|
|
|
|
def test_load_gpci_proposed_partitions_isolated(tmp_path, monkeypatch):
|
|
_patch_gpci_discovery(
|
|
monkeypatch,
|
|
{
|
|
"sup:2026_PFS_NPRM": _gpci_fixture_2026_wide(tmp_path),
|
|
"sup:2027_PFS_NPRM": _gpci_fixture_2027(tmp_path),
|
|
},
|
|
)
|
|
con = duckdb.connect()
|
|
nprm.load_gpci_proposed(con, year=2026)
|
|
nprm.load_gpci_proposed(con, year=2027)
|
|
nprm.load_gpci_proposed(con, year=2027) # reload one partition
|
|
counts = con.execute(
|
|
"SELECT cms_rule_id, count(*) FROM pfs.gpci_proposed "
|
|
"GROUP BY cms_rule_id ORDER BY cms_rule_id"
|
|
).fetchall()
|
|
assert counts == [("CMS-1832-P", 4), ("CMS-1848-P", 2)]
|