264 lines
8.2 KiB
Python
264 lines
8.2 KiB
Python
"""stack pfs — elements / lineage / families / review."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import duckdb
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
import cli.pfs as pfs_cli
|
|
from cli import app
|
|
from pfs.codetables import (
|
|
ElementRow,
|
|
EventRow,
|
|
ensure_tables,
|
|
read_elements,
|
|
read_events,
|
|
read_families,
|
|
write_elements,
|
|
write_events,
|
|
)
|
|
from pfs.extract import Extraction
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def restore_families():
|
|
"""Snapshot ``pfs.families.FAMILIES`` and restore it in teardown, even
|
|
if the test body raises — a test that reaches ``refresh_from`` must
|
|
not leave the live registry clobbered for the rest of the session."""
|
|
from pfs import families as mod
|
|
|
|
before = dict(mod.FAMILIES)
|
|
try:
|
|
yield mod
|
|
finally:
|
|
mod.FAMILIES.clear()
|
|
mod.FAMILIES.update(before)
|
|
|
|
|
|
@pytest.fixture
|
|
def con(monkeypatch, restore_families):
|
|
c = duckdb.connect(":memory:")
|
|
ensure_tables(c)
|
|
c.execute(
|
|
"CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER)"
|
|
)
|
|
c.executemany(
|
|
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
|
|
[
|
|
("99490", None, "Chrnc care mgmt staff 1st 20", "A", 1.0, 2026),
|
|
("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
|
|
("99999", None, "bundled thing", "B", 0.0, 2026),
|
|
],
|
|
)
|
|
|
|
class _Batch:
|
|
def __enter__(self):
|
|
return c
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
monkeypatch.setattr(pfs_cli, "_batch", lambda: _Batch())
|
|
monkeypatch.setattr(pfs_cli, "_read", lambda: c)
|
|
monkeypatch.setattr(pfs_cli, "_store", lambda: object())
|
|
published = []
|
|
monkeypatch.setattr(pfs_cli, "_publish", lambda: published.append(True))
|
|
|
|
class _ConProxy:
|
|
"""``DuckDBPyConnection`` has no ``__dict__`` (C extension type), so
|
|
a plain ``con.published = []`` fails; wrap it to attach the
|
|
publish-tracking list the tests assert on while forwarding
|
|
everything else (execute, executemany, ...) to the real connection."""
|
|
|
|
def __init__(self, real, published):
|
|
object.__setattr__(self, "_real", real)
|
|
object.__setattr__(self, "published", published)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(object.__getattribute__(self, "_real"), name)
|
|
|
|
try:
|
|
yield _ConProxy(c, published)
|
|
finally:
|
|
c.close()
|
|
|
|
|
|
class TestElements:
|
|
def test_writes_rows_and_publishes(self, con, monkeypatch):
|
|
def fake_extract(store, con_, code, *, classify=None):
|
|
return Extraction(
|
|
code,
|
|
(
|
|
ElementRow(
|
|
code,
|
|
2026,
|
|
"activity",
|
|
"consent",
|
|
"",
|
|
"Consent;",
|
|
"K",
|
|
1,
|
|
2,
|
|
"fr",
|
|
),
|
|
),
|
|
(),
|
|
)
|
|
|
|
monkeypatch.setattr(pfs_cli, "extract_code", fake_extract)
|
|
res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
|
|
assert res.exit_code == 0, res.output
|
|
assert "G0556: 1 elements, 0 for review" in res.output
|
|
assert [r.value for r in read_elements(con, "G0556")] == ["consent"]
|
|
assert con.published == [True]
|
|
|
|
def test_dry_run_does_not_write(self, con, monkeypatch):
|
|
monkeypatch.setattr(
|
|
pfs_cli,
|
|
"extract_code",
|
|
lambda s, c, code, *, classify=None: Extraction(code, (), ()),
|
|
)
|
|
|
|
def fail_batch():
|
|
raise AssertionError("--dry-run must not open a RW duckdb_batch connection")
|
|
|
|
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
|
|
|
|
read_calls = []
|
|
orig_read = pfs_cli._read
|
|
|
|
def spy_read():
|
|
read_calls.append(True)
|
|
return orig_read()
|
|
|
|
monkeypatch.setattr(pfs_cli, "_read", spy_read)
|
|
|
|
res = runner.invoke(
|
|
app, ["pfs", "elements", "--code", "G0556", "--no-llm", "--dry-run"]
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
assert con.published == []
|
|
assert read_calls == [True]
|
|
|
|
def test_all_payable_selects_art_status_in_newest_year(self, con, monkeypatch):
|
|
seen = []
|
|
monkeypatch.setattr(
|
|
pfs_cli,
|
|
"extract_code",
|
|
lambda s, c, code, *, classify=None: (
|
|
seen.append(code) or Extraction(code, (), ())
|
|
),
|
|
)
|
|
res = runner.invoke(
|
|
app, ["pfs", "elements", "--all-payable", "--no-llm", "--dry-run"]
|
|
)
|
|
assert res.exit_code == 0 and seen == ["99490"]
|
|
|
|
def test_repeated_family_option_expands_every_family(self, con, monkeypatch):
|
|
# #684 regression: --family used to be a single str, so a second
|
|
# --family silently discarded the first. It must now be repeatable
|
|
# and expand every family named.
|
|
seen = []
|
|
monkeypatch.setattr(
|
|
pfs_cli,
|
|
"extract_code",
|
|
lambda s, c, code, *, classify=None: (
|
|
seen.append(code) or Extraction(code, (), ())
|
|
),
|
|
)
|
|
res = runner.invoke(
|
|
app,
|
|
[
|
|
"pfs",
|
|
"elements",
|
|
"--family",
|
|
"CCM",
|
|
"--family",
|
|
"APCM",
|
|
"--no-llm",
|
|
"--dry-run",
|
|
],
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
from pfs.families import HAND_FAMILIES
|
|
|
|
expected = set(HAND_FAMILIES["CCM"].codes) | set(HAND_FAMILIES["APCM"].codes)
|
|
assert set(seen) == expected
|
|
|
|
|
|
class TestLineage:
|
|
def test_prints_and_writes(self, con, monkeypatch):
|
|
ev = EventRow(
|
|
"G2058",
|
|
2021,
|
|
"replaced_by",
|
|
"",
|
|
"99439",
|
|
"YBM4IZUS",
|
|
1578,
|
|
84639,
|
|
"fr",
|
|
True,
|
|
"",
|
|
)
|
|
monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [ev])
|
|
res = runner.invoke(app, ["pfs", "lineage", "--code", "G2058", "--write"])
|
|
assert res.exit_code == 0, res.output
|
|
assert "2021 replaced_by" in res.output and "YBM4IZUS ¶1578" in res.output
|
|
assert read_events(con, "G2058")[0].to_codes == "99439"
|
|
assert con.published == [True]
|
|
|
|
|
|
class TestFamilies:
|
|
def test_derives_from_tables_and_writes(self, con):
|
|
write_elements(
|
|
con,
|
|
"99439",
|
|
[
|
|
ElementRow(
|
|
"99439", 2021, "relation", "addon-of", "99490", "", "K", 1, 1, "fr"
|
|
)
|
|
],
|
|
[],
|
|
)
|
|
write_events(
|
|
con,
|
|
"G2058",
|
|
[
|
|
EventRow(
|
|
"G2058", 2021, "replaced_by", "", "99439", "K", 2, 1, "fr", True, ""
|
|
)
|
|
],
|
|
)
|
|
res = runner.invoke(app, ["pfs", "families", "--write"])
|
|
assert res.exit_code == 0, res.output
|
|
fams = read_families(con)
|
|
assert {r.code for r in fams if r.key == "CCM"} >= {"99490", "99439", "G2058"}
|
|
assert "CCM" in res.output
|
|
|
|
def test_write_tolerates_null_rvu_description(self, con):
|
|
# #687 regression: a real pfs.rvu row can carry a NULL description
|
|
# (a corpus parsing artifact) at the current max year with a
|
|
# payable status — that row lands in the derivation scope and used
|
|
# to crash `families --write` with an AttributeError.
|
|
con.execute(
|
|
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
|
|
("Z9999", None, None, "A", 1.0, 2026),
|
|
)
|
|
res = runner.invoke(app, ["pfs", "families", "--write"])
|
|
assert res.exit_code == 0, res.output
|
|
|
|
|
|
class TestReview:
|
|
def test_lists_queue(self, con):
|
|
from pfs.codetables import ReviewRow
|
|
|
|
write_elements(
|
|
con, "G0556", [], [ReviewRow("G0556", "Odd line;", "", "", "K", 9)]
|
|
)
|
|
res = runner.invoke(app, ["pfs", "review"])
|
|
assert res.exit_code == 0 and "G0556" in res.output and "Odd line" in res.output
|