84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
"""pfs.lineage — dated events from RVU diffs and FR paragraphs, cross-checked."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from pfs.lineage import rvu_events, rvu_year_span
|
|
|
|
RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER"
|
|
|
|
|
|
@pytest.fixture
|
|
def con():
|
|
c = duckdb.connect(":memory:")
|
|
c.execute("CREATE SCHEMA pfs")
|
|
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
|
|
rows = [
|
|
# table spans 2015..2026 (filler code keeps the span honest)
|
|
*[("00000", None, "filler", "A", 1.0, y) for y in range(2015, 2027)],
|
|
# 99487: B in 2015-2016, A from 2017; description rename 2021; +25% revalue 2022
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2015),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2016),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.00, 2017),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.02, 2018),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.05, 2019),
|
|
("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.10, 2020),
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.12, 2021),
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, 2022),
|
|
*[
|
|
("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, y)
|
|
for y in range(2023, 2027)
|
|
],
|
|
("99487", "26", "modifier row ignored", "A", 99.0, 2022),
|
|
# G2058: 2020 only
|
|
("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
|
|
# 99439: 2021 onward
|
|
*[
|
|
("99439", "", "Chrnc care mgmt svc ea addl", "A", 1.0, y)
|
|
for y in range(2021, 2027)
|
|
],
|
|
]
|
|
c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", rows)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
class TestSpan:
|
|
def test_span(self, con):
|
|
assert rvu_year_span(con) == (2015, 2026)
|
|
|
|
|
|
class TestRvuEvents:
|
|
def test_status_descriptor_and_revalue(self, con):
|
|
ev = rvu_events(con, "99487")
|
|
kinds = [(e.year, e.kind, e.note) for e in ev]
|
|
assert (2017, "status_change", "B→A") in kinds
|
|
assert (
|
|
2021,
|
|
"descriptor_change",
|
|
"Cmplx chron care w/o pt vsit → Cplx chrnc care 1st 60 min",
|
|
) in kinds
|
|
assert (2022, "revalued", "+25.0%") in kinds
|
|
assert not any(
|
|
k == "appeared" for _, k, _ in kinds
|
|
) # present from the first table year
|
|
assert all(
|
|
e.source == "rvu" and e.anchored is False and e.code == "99487" for e in ev
|
|
)
|
|
|
|
def test_appeared_and_disappeared(self, con):
|
|
ev = rvu_events(con, "G2058")
|
|
assert [(e.year, e.kind) for e in ev] == [
|
|
(2020, "appeared"),
|
|
(2021, "disappeared"),
|
|
]
|
|
|
|
def test_appeared_only(self, con):
|
|
ev = rvu_events(con, "99439")
|
|
assert [(e.year, e.kind) for e in ev] == [(2021, "appeared")]
|
|
|
|
def test_unknown_code(self, con):
|
|
assert rvu_events(con, "99999") == []
|