homelab/stack
1
0
Fork 0
Code Issues 210 Pull Requests 2 Actions Packages Projects 1 Releases 1 Wiki Activity
Files
96b74c545cd20f5ba60d9506ff9629ac2212d0df
stack/tests/cli/test_pfs_cli.py
kert 0f331176f2 fix(pfs): units without money say why; seat notes stored; seat ingest_log rows idempotent (R89) 
A unit with no billing specialty reads "no billing specialty identified" and a
known specialty with no PSPS row "no family billing on record" (stance stays
n/a) instead of claiming it bills under 1%. The seat page's self-disagreement
notes are stored as ingest_log rows (rule_id note) for the dossier, and a rerun
replaces a year's gap and note rows instead of duplicating them.
2026-11-12 21:40:53 -05:00

3594 lines
134 KiB
Python
Raw Blame History

"""stack pfs — elements / lineage / families / guidance / reaction / review."""
from __future__ import annotations
import dataclasses
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
import duckdb
import pytest
import typer
from typer.testing import CliRunner
import cli.pfs as pfs_cli
from cli import app
from pfs.codetables import (
ElementRow,
EventRow,
GuidanceRow,
ReactionRow,
ensure_tables,
read_elements,
read_events,
read_families,
read_guidance,
read_reaction,
read_rounds,
write_cpt_edition,
write_elements,
write_events,
write_guidance,
)
from pfs.cpt_model import CptCode, CptEdition, CptSection
from pfs.extract import Extraction
from pfs.stance import Scores
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())
# A DuckDB cursor, not `c` itself: `stack pfs reaction` (unlike the
# other commands here) calls `_read()` and then, on `--write`,
# `_batch()` in the same invocation — closing a `.close()`d `c`
# would break the batch. A cursor shares `c`'s catalog/data and
# closes independently of it, matching how `_read()`/`_batch()` are
# genuinely separate connections in production.
monkeypatch.setattr(pfs_cli, "_read", lambda: c.cursor())
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()
def _empty_extractions(store, con, codes, *, classify=None):
"""``pfs.extract.extract_codes``'s shape with nothing extracted — the
stand-in for tests that only care which codes the CLI targeted."""
return {code: Extraction(code, (), ()) for code in codes}
class TestElements:
def test_writes_rows_and_publishes(self, con, monkeypatch):
def fake_extract(store, con_, codes, *, classify=None):
return {
code: Extraction(
code,
(
ElementRow(
code,
2026,
"activity",
"consent",
"",
"Consent;",
"K",
1,
2,
"fr",
),
),
(),
)
for code in codes
}
monkeypatch.setattr(pfs_cli, "extract_codes", 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_summary_line_reports_confirmed_count_when_nonzero(self, con, monkeypatch):
# #703: the elements summary surfaces confirmed_by, but only when
# there's something to report.
def fake_extract(store, con_, codes, *, classify=None):
return {
code: Extraction(
code,
(
ElementRow(
code,
2026,
"activity",
"consent",
"",
"Consent;",
"K",
1,
2,
"fr",
"cpt:GQGTPGYV",
),
ElementRow(
code,
2026,
"activity",
"24-7-access",
"",
"24/7;",
"K",
3,
2,
"fr",
),
),
(),
)
for code in codes
}
monkeypatch.setattr(pfs_cli, "extract_codes", fake_extract)
res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
assert res.exit_code == 0, res.output
assert "G0556: 2 elements, 0 for review, 1 confirmed by a second source" in (
res.output
)
def test_summary_line_omits_confirmed_when_none(self, con, monkeypatch):
monkeypatch.setattr(pfs_cli, "extract_codes", _empty_extractions)
res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
assert res.exit_code == 0, res.output
assert "confirmed" not in res.output
def test_dry_run_does_not_write(self, con, monkeypatch):
monkeypatch.setattr(pfs_cli, "extract_codes", _empty_extractions)
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_codes",
lambda s, c, codes, *, classify=None: (
seen.extend(codes) or _empty_extractions(s, c, codes)
),
)
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_codes",
lambda s, c, codes, *, classify=None: (
seen.extend(codes) or _empty_extractions(s, c, codes)
),
)
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 TestCodesForWarnSlow:
"""``_codes_for``'s target-count note is for ``elements``, whose
``--all-payable`` run makes one model call per unplaced line — not for
``lineage --all-payable``, which is pure SQL/regex."""
def test_warns_by_default(self, con, caplog):
pfs_cli._codes_for(con, [], [], True)
assert "targeting 1 codes" in caplog.text
def test_silent_when_warn_slow_is_false(self, con, caplog):
pfs_cli._codes_for(con, [], [], True, warn_slow=False)
assert "targeting" not in caplog.text
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]
def test_prints_cpt_event_with_cpt_anchor_text(self, con, monkeypatch):
ev = EventRow(
"99490",
2015,
"cpt_changed",
"",
"",
"GQGTPGYV",
0,
0,
"cpt",
True,
"CPT Changes 2015",
)
monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [ev])
res = runner.invoke(app, ["pfs", "lineage", "--code", "99490", "--write"])
assert res.exit_code == 0, res.output
assert "2015 cpt_changed" in res.output
assert "cpt 2015 GQGTPGYV" in res.output
def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
# I4: `stack pfs lineage` without --write must read the replica,
# not open a RW duckdb_batch connection (the DuckDB single-writer
# lock a notebook may be holding, #508-#514).
def fail_batch():
raise AssertionError(
"lineage without --write must not open a RW duckdb_batch connection"
)
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [])
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", "lineage", "--code", "G2058"])
assert res.exit_code == 0, res.output
assert read_calls == [True]
assert con.published == []
def test_requires_code_or_all_payable(self, con):
res = runner.invoke(app, ["pfs", "lineage"])
assert res.exit_code != 0
def test_all_payable_limit_write_calls_writer_once_per_code_after_build(
self, con, monkeypatch
):
# Extra A/R/T, newest-year codes so --limit has something to
# trim: sorted, the newest-year A/R/T targets are
# 99490 < A0001 < B0002 ('9' sorts before letters); 99999 is
# status B (excluded) and G2058 is a stale year (excluded).
con.execute(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
["A0001", None, "extra a", "A", 1.0, 2026],
)
con.execute(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
["B0002", None, "extra b", "R", 1.0, 2026],
)
order: list[tuple] = []
def fake_lineage_all(con_, store_, codes):
order.append(("lineage_all", tuple(codes)))
return {c: [] for c in codes}
monkeypatch.setattr(pfs_cli, "lineage_all", fake_lineage_all)
written: list[str] = []
def fake_write_events(con_, code, rows):
order.append(("write", code))
written.append(code)
return 0
monkeypatch.setattr(pfs_cli, "write_events", fake_write_events)
real_batch = pfs_cli._batch
def recording_batch():
order.append(("batch",))
return real_batch()
monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
res = runner.invoke(
app, ["pfs", "lineage", "--all-payable", "--limit", "2", "--write"]
)
assert res.exit_code == 0, res.output
assert order[0] == ("lineage_all", ("99490", "A0001"))
assert order[1] == ("batch",)
assert order[2:] == [("write", "99490"), ("write", "A0001")]
assert written == ["99490", "A0001"]
assert con.published == [True]
def test_all_payable_without_write_never_opens_the_batch(self, con, monkeypatch):
monkeypatch.setattr(
pfs_cli, "lineage_all", lambda c, s, codes: {c: [] for c in codes}
)
def fail_batch():
raise AssertionError(
"lineage --all-payable without --write must not open a RW "
"duckdb_batch connection"
)
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "lineage", "--all-payable"])
assert res.exit_code == 0, res.output
assert con.published == []
assert "lineage: 1 codes targeted" in res.output
def test_lineage_all_seen_targets_deleted_codes(self, con, monkeypatch):
con.execute(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
["99343", None, "old home visit", "A", 1.0, 2022],
)
order: list[tuple] = []
def fake_lineage_all(con_, store_, codes):
order.append(("lineage_all", tuple(codes)))
return {c: [] for c in codes}
monkeypatch.setattr(pfs_cli, "lineage_all", fake_lineage_all)
monkeypatch.setattr(pfs_cli, "write_events", lambda c, code, rows: 0)
real_batch = pfs_cli._batch
monkeypatch.setattr(
pfs_cli, "_batch", lambda: (order.append(("batch",)), real_batch())[1]
)
res = runner.invoke(app, ["pfs", "lineage", "--all-seen", "--write"])
assert res.exit_code == 0, res.output
targets = order[0][1]
assert "99343" in targets and "G2058" in targets and "99999" in targets
assert order[1] == ("batch",)
assert con.published == [True]
def test_lineage_all_seen_limit_and_code_union(self, con, monkeypatch):
seen: list[tuple] = []
def fake(con_, store_, codes):
seen.append(tuple(codes))
return {c: [] for c in codes}
monkeypatch.setattr(pfs_cli, "lineage_all", fake)
res = runner.invoke(
app, ["pfs", "lineage", "--all-seen", "--limit", "1", "--code", "zzzzz"]
)
assert res.exit_code == 0, res.output
assert len(seen[0]) == 2 and "ZZZZZ" in seen[0]
def test_lineage_family_targets_curated_members(self, con, monkeypatch):
seen: list[tuple] = []
def fake(con_, store_, codes):
seen.append(tuple(codes))
return {c: [] for c in codes}
monkeypatch.setattr(pfs_cli, "lineage_all", fake)
res = runner.invoke(app, ["pfs", "lineage", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert len(seen[0]) == 20
assert "lineage: 20 codes targeted" in res.output
def test_lineage_unknown_family_is_rejected(self, con):
res = runner.invoke(app, ["pfs", "lineage", "--family", "NOPE-NOPE"])
assert res.exit_code != 0
def test_codes_for_with_no_targets_names_every_flag(self, con):
with pytest.raises(Exception, match="--all-seen"):
pfs_cli._codes_for(con, [], [], False)
def test_codes_for_all_seen_warns_unless_quiet(self, con, caplog):
import logging
with caplog.at_level(logging.WARNING):
out = pfs_cli._codes_for(con, [], [], False, all_seen=True)
assert "99490" in out and "--all-seen: targeting" in caplog.text
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"}
# Minor: --write prints the summary line by default; the
# per-family dump ("CCM" among them) is --verbose only.
assert "CCM" not in res.output
# No pfs.cpt_* rows seeded in this test — the summary line must
# still print, with cpt-named at 0 (I4: read-only-tolerant path
# also applies to the write path when the CPT tables are empty).
assert (
"families: total 3, multi-code 2, cpt-named 0, hand 1, curated 1, other 1"
in (res.output)
)
def test_verbose_prints_the_per_family_dump(self, con):
write_elements(
con,
"99439",
[
ElementRow(
"99439", 2021, "relation", "addon-of", "99490", "", "K", 1, 1, "fr"
)
],
[],
)
res = runner.invoke(app, ["pfs", "families", "--write", "--verbose"])
assert res.exit_code == 0, res.output
assert "CCM" in res.output
assert "families: total" in res.output
def _ccm_cpt_edition(self, year=2024):
section = CptSection(
sec_id="sec_ccm",
level=3,
title="Chronic Care Management Services",
path=(
"Evaluation and Management",
"Care Management Services",
"Chronic Care Management Services",
),
code_lo="99490",
code_hi="99439",
guideline="",
)
codes = (
CptCode(
code="99490",
sec_id="sec_ccm",
descriptor="Chronic care management services, first 20 minutes.",
stem="Chronic care management services",
elements=(),
tail="first 20 minutes.",
addon=False,
resequenced=False,
new=False,
revised=False,
telemedicine=False,
parent="",
mod51_exempt=False,
audio_only=False,
fda_pending=False,
pla=False,
category="I",
),
CptCode(
code="99439",
sec_id="sec_ccm",
descriptor="each additional 20 minutes.",
stem="Chronic care management services",
elements=(),
tail="each additional 20 minutes.",
addon=True,
resequenced=False,
new=False,
revised=False,
telemedicine=False,
parent="99490",
mod51_exempt=False,
audio_only=False,
fda_pending=False,
pla=False,
category="I",
),
)
return CptEdition(
year=year,
sections=(section,),
codes=codes,
instructions=(),
references=(),
crosswalks=(),
lists=(),
)
def test_feeds_cpt_inputs_when_present_and_prints_summary_line(self, con):
write_cpt_edition(con, self._ccm_cpt_edition(), "ITEM0001")
res = runner.invoke(app, ["pfs", "families", "--write"])
assert res.exit_code == 0, res.output
fams = read_families(con)
ccm = {r.code: r for r in fams if r.key == "CCM"}
assert {"99490", "99439"} <= set(ccm)
assert ccm["99490"].note == (
"Evaluation and Management > Care Management Services > "
"Chronic Care Management Services"
)
assert "families: total" in res.output
assert "cpt-named" in res.output
def test_since_spans_all_ingested_editions_not_just_the_newest(self, con):
# Ruling C12: _cpt_inputs' cpt_presence comes from a full
# SELECT code, edition_year FROM pfs.cpt_code (every edition
# write_cpt_edition has ever written), so `since` is the
# earliest edition, not clamped to whichever one is newest.
write_cpt_edition(con, self._ccm_cpt_edition(year=2019), "ITEM_2019")
write_cpt_edition(con, self._ccm_cpt_edition(year=2024), "ITEM_2024")
res = runner.invoke(app, ["pfs", "families", "--write"])
assert res.exit_code == 0, res.output
fams = read_families(con)
ccm = {r.code: r for r in fams if r.key == "CCM"}
assert ccm["99490"].since == 2019
assert ccm["99490"].until is None
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
def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
# I4: `stack pfs families` without --write must read the replica,
# not open a RW duckdb_batch connection.
def fail_batch():
raise AssertionError(
"families without --write 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", "families"])
assert res.exit_code == 0, res.output
assert read_calls == [True]
assert con.published == []
def test_no_derived_tables_yet_is_a_friendly_no_op(self, monkeypatch):
# I4: a fresh replica with no pfs.code_element/code_event/code_family
# (nothing has run --write yet) is not a bug — print a friendly
# message and exit 0, without ever taking the write lock.
bare = duckdb.connect(":memory:")
monkeypatch.setattr(pfs_cli, "_read", lambda: bare)
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
try:
res = runner.invoke(app, ["pfs", "families"])
assert res.exit_code == 0, res.output
assert "no derived tables yet" in res.output
finally:
bare.close()
def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
# M3: FAMILIES can hold ~15k derived keys once `refresh_from` has
# run — the error text must stay short and human-scannable.
pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
res = runner.invoke(
app, ["pfs", "elements", "--family", "NOPE", "--no-llm", "--dry-run"]
)
assert res.exit_code != 0
assert "ZZZZ-99999" not in res.output
for key in pfs_cli.HAND_FAMILIES:
assert key in res.output
class TestCptIngest:
def test_dry_run_never_opens_batch(self, con, monkeypatch):
seen = {}
def fake_ingest(store, con_, *, years=None, dry_run=False):
seen["store"] = store
seen["con"] = con_
seen["years"] = years
seen["dry_run"] = dry_run
return {2024: {"sections": 3, "codes": 5}}
monkeypatch.setattr(pfs_cli, "cpt_ingest", fake_ingest)
def fail_batch():
raise AssertionError("--dry-run must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(
app, ["pfs", "cpt-ingest", "--edition", "2024", "--dry-run"]
)
assert res.exit_code == 0, res.output
assert seen["dry_run"] is True
assert seen["con"] is None
assert seen["years"] == [2024]
assert "2024: sections=3, codes=5" in res.output
assert con.published == []
def test_all_flag_passes_years_none(self, con, monkeypatch):
seen = {}
monkeypatch.setattr(
pfs_cli,
"cpt_ingest",
lambda store, con_, *, years=None, dry_run=False: (
seen.update(years=years) or {}
),
)
res = runner.invoke(app, ["pfs", "cpt-ingest", "--all", "--dry-run"])
assert res.exit_code == 0, res.output
assert seen["years"] is None
def test_requires_edition_or_all(self, con):
res = runner.invoke(app, ["pfs", "cpt-ingest"])
assert res.exit_code != 0
def test_write_path_opens_batch_ensures_tables_and_publishes(
self, con, monkeypatch
):
seen = {}
def fake_ingest(store, con_, *, years=None, dry_run=False):
seen["con"] = con_
seen["dry_run"] = dry_run
return {2024: {"sections": 1, "codes": 2}}
monkeypatch.setattr(pfs_cli, "cpt_ingest", fake_ingest)
res = runner.invoke(app, ["pfs", "cpt-ingest", "--edition", "2024"])
assert res.exit_code == 0, res.output
assert seen["dry_run"] is False
assert seen["con"] is not None
assert "2024: sections=1, codes=2" in res.output
assert con.published == [True]
class TestGuidance:
def _row(self, family="CCM", code="99490"):
return GuidanceRow(
family, code, "cfr", "42 CFR 410.78(a)(3)", "SEC", "K", 398, 32389
)
def test_write_builds_writes_and_publishes(self, con, monkeypatch):
seen = {}
def fake_build(con_, store_, codes, *, family):
seen["con"] = con_
seen["store"] = store_
seen["codes"] = codes
seen["family"] = family
return [self._row(family=family)]
monkeypatch.setattr(pfs_cli, "build_guidance", fake_build)
res = runner.invoke(app, ["pfs", "guidance", "--family", "ccm", "--write"])
assert res.exit_code == 0, res.output
assert seen["family"] == "CCM"
assert seen["con"] is not None
assert seen["store"] is not None
assert set(seen["codes"]) == set(pfs_cli.HAND_FAMILIES["CCM"].codes)
got = read_guidance(con, "CCM")
assert [r.code for r in got] == ["99490"]
assert con.published == [True]
assert "CCM cfr 42 CFR 410.78(a)(3) SEC ← K ¶398" in res.output
def test_write_scopes_per_family(self, con, monkeypatch):
monkeypatch.setattr(
pfs_cli,
"build_guidance",
lambda con_, store_, codes, *, family: [self._row(family=family)],
)
res = runner.invoke(
app, ["pfs", "guidance", "--family", "CCM", "--family", "APCM", "--write"]
)
assert res.exit_code == 0, res.output
assert [r.family for r in read_guidance(con, "CCM")] == ["CCM"]
assert [r.family for r in read_guidance(con, "APCM")] == ["APCM"]
def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
write_guidance(con, "CCM", [self._row()])
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "guidance", "--family", "CCM"])
assert res.exit_code == 0, res.output
assert "CCM cfr 42 CFR 410.78(a)(3) SEC ← K ¶398" in res.output
assert con.published == []
def test_no_derived_guidance_yet_is_a_friendly_no_op(self, monkeypatch):
# I4: a fresh replica with no pfs.code_guidance (guidance --write
# hasn't run yet) is not a bug — print a friendly message per
# family and exit 0, without ever taking the write lock.
bare = duckdb.connect(":memory:")
monkeypatch.setattr(pfs_cli, "_read", lambda: bare)
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
try:
res = runner.invoke(app, ["pfs", "guidance", "--family", "CCM"])
assert res.exit_code == 0, res.output
assert "no derived guidance yet" in res.output
finally:
bare.close()
def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
res = runner.invoke(app, ["pfs", "guidance", "--family", "NOPE", "--write"])
assert res.exit_code != 0
assert "ZZZZ-99999" not in res.output
for key in pfs_cli.HAND_FAMILIES:
assert key in res.output
def test_unknown_key_among_several_errors_before_any_batch(self, con, monkeypatch):
# Every --family key must be resolved and validated up front — an
# unknown key later in the list must fail before build_guidance
# runs for any family and before the write lock is ever opened.
built: list[str] = []
monkeypatch.setattr(
pfs_cli,
"build_guidance",
lambda con_, store_, codes, *, family: built.append(family) or [],
)
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(
app, ["pfs", "guidance", "--family", "CCM", "--family", "NOPE", "--write"]
)
assert res.exit_code != 0
assert built == []
def test_write_builds_every_family_before_opening_the_batch(self, con, monkeypatch):
order: list[str] = []
def fake_build(con_, store_, codes, *, family):
order.append("build")
return [self._row(family=family)]
real_batch = pfs_cli._batch
def recording_batch():
order.append("batch")
return real_batch()
monkeypatch.setattr(pfs_cli, "build_guidance", fake_build)
monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
res = runner.invoke(
app, ["pfs", "guidance", "--family", "CCM", "--family", "APCM", "--write"]
)
assert res.exit_code == 0, res.output
assert order == ["build", "build", "batch"]
def test_requires_family(self, con):
res = runner.invoke(app, ["pfs", "guidance"])
assert res.exit_code != 0
class TestReaction:
def _row(
self, family="CCM", period="CMS-2023-0121", period_kind="docket", year=2023
):
return ReactionRow(family, period, period_kind, year, 5, 20, 1, 0, 0, 1, "[]")
def test_write_builds_writes_and_publishes(self, con, monkeypatch):
seen = {}
def fake_series(
engine,
store_,
family,
codes,
*,
classify=None,
stance_sample_n=0,
name=None,
):
seen["engine"] = engine
seen["store"] = store_
seen["family"] = family
seen["codes"] = codes
seen["classify"] = classify
seen["n"] = stance_sample_n
seen["name"] = name
return [self._row(family=family)]
monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
res = runner.invoke(app, ["pfs", "reaction", "--family", "ccm", "--write"])
assert res.exit_code == 0, res.output
assert seen["family"] == "CCM"
assert seen["engine"] == "ENGINE"
assert seen["classify"] is None # no --stance-sample: no LLM at all
assert seen["n"] == 0
assert seen["name"] == pfs_cli.HAND_FAMILIES["CCM"].name
assert set(seen["codes"]) == set(pfs_cli.HAND_FAMILIES["CCM"].codes)
got = read_reaction(con, "CCM")
assert [r.period for r in got] == ["CMS-2023-0121"]
assert con.published == [True]
assert "CCM CMS-2023-0121 (2023) 5/20 commenters" in res.output
def test_code_makes_a_single_code_pseudo_family_with_no_display_name(
self, con, monkeypatch
):
seen = {}
def fake_series(
engine,
store_,
family,
codes,
*,
classify=None,
stance_sample_n=0,
name=None,
):
seen["family"] = family
seen["codes"] = codes
seen["name"] = name
return [self._row(family=family)]
monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
res = runner.invoke(app, ["pfs", "reaction", "--code", "g2211", "--write"])
assert res.exit_code == 0, res.output
assert seen["family"] == "G2211"
assert seen["codes"] == ("G2211",)
assert seen["name"] is None
assert [r.family for r in read_reaction(con, "G2211")] == ["G2211"]
def test_stance_sample_builds_the_classifier(self, con, monkeypatch):
seen = {}
monkeypatch.setattr(
pfs_cli,
"reaction_series",
lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0, name=None: ( # noqa: E501
seen.update(classify=classify, n=stance_sample_n) or []
),
)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_classifier", lambda: "CLASSIFIER")
res = runner.invoke(
app, ["pfs", "reaction", "--family", "CCM", "--stance-sample", "5"]
)
assert res.exit_code == 0, res.output
assert seen["classify"] == "CLASSIFIER"
assert seen["n"] == 5
def test_write_computes_classifier_and_series_before_opening_the_batch(
self, con, monkeypatch
):
# Ruling A13: the classifier and every family's series must be
# fully computed BEFORE the DuckDB write lock is taken — the
# batch block only writes and publishes.
order: list[str] = []
def fake_classifier():
order.append("classifier")
return "CLASSIFIER"
def fake_series(
engine,
store_,
family,
codes,
*,
classify=None,
stance_sample_n=0,
name=None,
):
order.append("series")
assert classify == "CLASSIFIER" # already built by the time series runs
return [self._row(family=family)]
real_batch = pfs_cli._batch
def recording_batch():
order.append("batch")
return real_batch()
monkeypatch.setattr(pfs_cli, "_classifier", fake_classifier)
monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
res = runner.invoke(
app,
["pfs", "reaction", "--family", "CCM", "--stance-sample", "5", "--write"],
)
assert res.exit_code == 0, res.output
assert order == ["classifier", "series", "batch"]
def test_without_write_never_opens_a_batch_connection(self, con, monkeypatch):
monkeypatch.setattr(
pfs_cli,
"reaction_series",
lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0, name=None: [ # noqa: E501
self._row(family=family)
],
)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "reaction", "--family", "CCM"])
assert res.exit_code == 0, res.output
assert "CCM CMS-2023-0121 (2023) 5/20 commenters" in res.output
assert con.published == [] # nothing persisted, batch never opened
def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
res = runner.invoke(app, ["pfs", "reaction", "--family", "NOPE", "--write"])
assert res.exit_code != 0
assert "ZZZZ-99999" not in res.output
for key in pfs_cli.HAND_FAMILIES:
assert key in res.output
def test_requires_family_or_code(self, con):
res = runner.invoke(app, ["pfs", "reaction"])
assert res.exit_code != 0
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
def test_source_shown_when_present(self, con):
# #703: CPT-sourced review lines are distinguishable from FR ones.
from pfs.codetables import ReviewRow
write_elements(
con,
"G0556",
[],
[
ReviewRow("G0556", "Odd fr line;", "", "", "K", 9, "fr"),
ReviewRow("G0556", "Odd cpt line;", "", "", "GQGTPGYV", 0, "cpt"),
],
)
res = runner.invoke(app, ["pfs", "review"])
assert res.exit_code == 0, res.output
assert "(fr)" in res.output and "(cpt)" in res.output
def test_code_filter_scopes_the_query(self, con):
from pfs.codetables import ReviewRow
write_elements(
con, "G0556", [], [ReviewRow("G0556", "Odd line;", "", "", "K", 9, "fr")]
)
write_elements(
con, "99490", [], [ReviewRow("99490", "Other line;", "", "", "K", 3, "fr")]
)
res = runner.invoke(app, ["pfs", "review", "--code", "g0556"])
assert res.exit_code == 0, res.output
assert "Odd line" in res.output and "Other line" not in res.output
class TestExposure:
def test_write_builds_whole_calendar_and_publishes(self, con, monkeypatch):
from pfs.codetables import read_exposures
res = runner.invoke(app, ["pfs", "exposure", "--write"])
assert res.exit_code == 0, res.output
rows = read_exposures(con)
got = {(r.code, r.year, r.kind) for r in rows}
# 99490 appears (payable) in 2026 after being absent; G2058's only year is
# 2020 so it ends 2021; 99999 is B-status → never payable → no rows.
assert ("99490", 2026, "becomes-payable") in got
assert ("G2058", 2021, "ends") in got
assert not any(r.code == "99999" for r in rows)
assert con.published == [True]
assert "99490 2026-01-01 becomes-payable" in res.output
assert "exposures:" in res.output
def test_filters_only_affect_printing(self, con, monkeypatch):
from pfs.codetables import read_exposures
res = runner.invoke(app, ["pfs", "exposure", "--write", "--code", "g2058"])
assert res.exit_code == 0, res.output
assert len({r.code for r in read_exposures(con)}) == 2 # whole calendar written
assert "G2058" in res.output and "99490" not in res.output
def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
runner.invoke(app, ["pfs", "exposure", "--write"])
con.published.clear()
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "exposure", "--code", "99490"])
assert res.exit_code == 0, res.output
assert "99490 2026-01-01 becomes-payable" in res.output
assert con.published == []
def test_controls_lists_quiet_codes_in_the_same_band(self, con, monkeypatch):
# two more A-status codes in 2026 in the same (single) band, one of
# which is itself exposed that year → only the quiet one is a control
con.executemany(
"INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
[
# 11111 is payable across the whole file span → never treated
*[("11111", None, "quiet", "A", 1.0, y) for y in range(2020, 2027)],
("22222", None, "new", "A", 1.0, 2026), # becomes payable 2026 itself
],
)
runner.invoke(app, ["pfs", "exposure", "--write"])
res = runner.invoke(
app,
["pfs", "exposure", "--code", "99490", "--year", "2026", "--controls", "5"],
)
assert res.exit_code == 0, res.output
assert "11111 A band" in res.output
assert "22222" not in res.output
assert "controls for 99490 @ 2026" in res.output
def test_controls_requires_one_code_and_year(self, con):
res = runner.invoke(app, ["pfs", "exposure", "--controls", "5"])
assert res.exit_code != 0
def test_no_calendar_yet_is_a_friendly_no_op(self, monkeypatch):
bare = duckdb.connect(":memory:")
monkeypatch.setattr(pfs_cli, "_read", lambda: bare)
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
try:
res = runner.invoke(app, ["pfs", "exposure"])
assert res.exit_code == 0, res.output
assert "no exposure calendar yet" in res.output
finally:
bare.close()
class TestUtilization:
_REC = {
"Rndrng_Prvdr_Geo_Lvl": "National",
"HCPCS_Cd": "99490",
"HCPCS_Desc": "CCM first 20 min",
"HCPCS_Drug_Ind": "N",
"Place_Of_Srvc": "O",
"Tot_Rndrng_Prvdrs": "10",
"Tot_Benes": "100",
"Tot_Srvcs": "1000",
"Tot_Bene_Day_Srvcs": "990",
"Avg_Sbmtd_Chrg": "100",
"Avg_Mdcr_Alowd_Amt": "60",
"Avg_Mdcr_Pymt_Amt": "45",
"Avg_Mdcr_Stdzd_Amt": "44",
}
@pytest.fixture
def env(self, con, monkeypatch, tmp_path):
from bib.store import Store
store = Store(":memory:", storage_dir=tmp_path / "storage")
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(
pfs_cli,
"utilization_cache_path",
lambda y, root=None: tmp_path / f"mup-phy-geo-{y}.json",
)
fetched: list[int] = []
def fake_fetch(client, y, **kw):
fetched.append(y)
return [self._REC, dict(self._REC, Place_Of_Srvc="F", Tot_Srvcs="5")]
monkeypatch.setattr(pfs_cli, "fetch_utilization_year", fake_fetch)
monkeypatch.setattr(pfs_cli, "_http", lambda: MagicMock())
try:
yield store, fetched, tmp_path
finally:
store.close()
def test_write_loads_cites_logs_and_publishes(self, con, env):
store, fetched, tmp_path = env
res = runner.invoke(app, ["pfs", "utilization", "--write", "--year", "2023"])
assert res.exit_code == 0, res.output
assert fetched == [2023]
assert con.execute(
"SELECT count(*), sum(n_services) FROM pfs.utilization WHERE year = 2023"
).fetchone() == (2, 1005.0)
key = con.execute("SELECT DISTINCT item_key FROM pfs.utilization").fetchone()[0]
item = store.get(key)
assert "CY2023" in item.title and "year:2023" in item.tags
log = con.execute(
"SELECT module, table_name, rows, pincite_key, source_file FROM cms.ingest_log"
).fetchone()
assert log[:4] == ("pfs.utilization", "pfs.utilization", 2, key)
assert log[4].endswith("mup-phy-geo-2023.json")
assert (tmp_path / "mup-phy-geo-2023.json").exists()
assert con.published == [True]
assert "2023: 2 records → 2 national rows" in res.output
def test_write_is_idempotent_and_offline_reads_the_cache(self, con, env):
store, fetched, tmp_path = env
runner.invoke(app, ["pfs", "utilization", "--write", "--year", "2023"])
res = runner.invoke(
app, ["pfs", "utilization", "--write", "--year", "2023", "--offline"]
)
assert res.exit_code == 0, res.output
assert fetched == [2023] # the offline run never fetched
assert con.execute("SELECT count(*) FROM pfs.utilization").fetchone()[0] == 2
def test_unknown_year_is_rejected_before_any_fetch(self, con, env):
store, fetched, _ = env
res = runner.invoke(app, ["pfs", "utilization", "--write", "--year", "1999"])
assert res.exit_code != 0
assert fetched == []
def test_code_reads_series_without_the_write_lock(self, con, env, monkeypatch):
runner.invoke(app, ["pfs", "utilization", "--write", "--year", "2023"])
con.published.clear()
def fail_batch():
raise AssertionError("must not open a RW duckdb_batch connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "utilization", "--code", "99490"])
assert res.exit_code == 0, res.output
assert "99490 2023 services 1,005" in res.output
assert con.published == []
def test_no_table_yet_is_a_friendly_no_op(self, monkeypatch):
bare = duckdb.connect(":memory:")
monkeypatch.setattr(pfs_cli, "_read", lambda: bare)
try:
res = runner.invoke(app, ["pfs", "utilization", "--code", "99490"])
assert res.exit_code == 0, res.output
assert "no utilization table yet" in res.output
finally:
bare.close()
class TestFamiliesCurated:
def test_families_write_applies_curated(self, con, monkeypatch):
from pfs.codetables import FamilyRow
derived = [
FamilyRow(
"NEW-PATIENT", "New Patient", "99341", "base", 2015, None, "", 0, ""
)
]
monkeypatch.setattr(pfs_cli, "_derive_rows", lambda c: derived)
res = runner.invoke(app, ["pfs", "families", "--write"])
assert res.exit_code == 0, res.output
fams = read_families(con)
assert [r.key for r in fams if r.code == "99341"] == ["HOME-VISITS"]
assert len([r for r in fams if r.key == "HOME-VISITS"]) == 20
assert "curated 1" in res.output
def test_asker_builds_json_extractor_on_instruct_model(monkeypatch):
from llm import config as llm_config
from llm import pool as llm_pool
from llm.structured import JsonExtractor
class Cfg:
instruct_model = "qwen-test"
chat_num_ctx = 1024
class FakePool:
def check(self, model):
return ["http://h"]
monkeypatch.setattr(llm_config, "load", lambda: Cfg())
monkeypatch.setattr(
llm_pool.HostPool, "from_config", classmethod(lambda c, cfg: FakePool())
)
asker = pfs_cli._asker()
assert isinstance(asker, JsonExtractor) and asker.model == "qwen-test"
@dataclasses.dataclass
class _Cfg:
instruct_model: str = "qwen-14"
instruct_model_large: str = "qwen-32"
chat_num_ctx: int = 1024
def test_asker_large_pins_the_large_model(monkeypatch):
from llm import config as llm_config
from llm import pool as llm_pool
seen = []
class FakePool:
def check(self, model):
seen.append(model)
monkeypatch.setattr(llm_config, "load", lambda: _Cfg())
monkeypatch.setattr(
llm_pool.HostPool, "from_config", classmethod(lambda c, cfg: FakePool())
)
asker = pfs_cli._asker(large=True)
assert asker.model == "qwen-32" and seen == ["qwen-32"]
def test_asker_large_without_a_large_model(monkeypatch):
from llm import config as llm_config
monkeypatch.setattr(llm_config, "load", lambda: _Cfg(instruct_model_large=""))
with pytest.raises(Exception, match="no instruct_model_large"):
pfs_cli._asker(large=True)
class TestRounds:
def _row(self, **kw):
from pfs.codetables import RoundRow
base = dict(
family="HOME-VISITS",
code="99341",
rule_year=2023,
item_key="K",
rule_kind="final",
p_id=1,
page=2,
source="fr-para",
table_title="",
ruc_work_rvu=1.0,
ruc_percentile="25th",
survey_intra_min=None,
survey_total_min=None,
comparators="",
surveying_societies="",
requested_work_rvu=None,
proposed_work_rvu=1.0,
final_work_rvu=1.0,
decision="accept",
rationale_p_id=0,
rationale_quote="",
pe_note="",
extractor="regex",
)
return RoundRow(**{**base, **kw})
def _patch(self, monkeypatch, order, counts=None):
rows = [
self._row(),
self._row(rule_year=2008, rule_kind="final"),
self._row(code="99342", rule_year=2023, field_cites='{"decision": 1}'),
]
seen = {}
def fake(con, store, fam, **kw):
order.append("family_rounds")
seen.update(kw, key=fam.key)
return rows, counts or {
"done": 2,
"skipped": 1,
"failed": 0,
"failed_pending": 3,
}
monkeypatch.setattr(pfs_cli, "family_rounds", fake)
monkeypatch.setattr(pfs_cli, "_asker", lambda: "ASK")
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
return seen
def test_rounds_cli_writes_after_extraction(self, con, monkeypatch):
order = []
seen = self._patch(monkeypatch, order)
orig_write = pfs_cli.write_rounds
def write(c, fam, rows):
order.append("write_rounds")
return orig_write(c, fam, rows)
class _B:
def __enter__(self):
order.append("_batch")
return con._real
def __exit__(self, *a):
return False
monkeypatch.setattr(pfs_cli, "write_rounds", write)
monkeypatch.setattr(pfs_cli, "_batch", lambda: _B())
monkeypatch.setattr(pfs_cli, "_publish", lambda: order.append("_publish"))
res = runner.invoke(
app,
[
"pfs",
"rounds",
"--family",
"home-visits",
"--limit",
"5",
"--retry-failed",
"--write",
],
)
assert res.exit_code == 0, res.output
assert order == ["family_rounds", "_batch", "write_rounds", "_publish"]
assert seen["ask"] == "ASK" and seen["limit"] == 5 and seen["retry_failed"]
assert seen["key"] == "HOME-VISITS" and not seen["tables_only"]
assert "CY2008 final: 1 rounds" in res.output
assert "CY2023 final: 2 rounds" in res.output
assert "done 2, skipped 1, failed 0, failed_pending 3" in res.output
assert "wrote 3 rounds" in res.output
got = read_rounds(con._real, "HOME-VISITS")
assert len(got) == 3
assert {r.code: r.field_cites for r in got}["99342"] == '{"decision": 1}'
def test_rounds_without_write_writes_nothing(self, con, monkeypatch):
order = []
self._patch(monkeypatch, order)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(app, ["pfs", "rounds", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert order == ["family_rounds"] and "wrote" not in res.output
@pytest.mark.parametrize("flag", ["--no-llm", "--tables-only"])
def test_no_model_flags_build_no_asker(self, con, monkeypatch, flag):
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
res = runner.invoke(app, ["pfs", "rounds", "--family", "HOME-VISITS", flag])
assert res.exit_code == 0, res.output
assert seen["ask"] is None and seen["tables_only"] == (flag == "--tables-only")
# the finished ledger is replayed, so a --write cannot delete model rows
assert seen["replay_model"] == "qwen-x"
def test_model_run_does_not_replay(self, con, monkeypatch):
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "_model_name", lambda: 1 / 0)
res = runner.invoke(app, ["pfs", "rounds", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert seen["replay_model"] is None
def test_model_name_reads_config_without_a_pool(self, monkeypatch):
from llm import config as llm_config
class Cfg:
instruct_model = "qwen-test"
monkeypatch.setattr(llm_config, "load", lambda: Cfg())
assert pfs_cli._model_name() == "qwen-test"
def test_rounds_unknown_family(self, con):
res = runner.invoke(app, ["pfs", "rounds", "--family", "NOPE"])
assert res.exit_code != 0 and "unknown curated family" in res.output
def test_rounds_dry_run_makes_no_model_calls(self, con, monkeypatch):
monkeypatch.setattr(
pfs_cli, "rounds_inventory", lambda store, fam: ([{}, {}, {}], [1, 2])
)
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "family_rounds", lambda *a, **k: 1 / 0)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(
app, ["pfs", "rounds", "--family", "HOME-VISITS", "--dry-run"]
)
assert res.exit_code == 0, res.output
assert "3 candidate paragraphs, 2 table rows; no model calls made" in res.output
class TestOrigin:
def _row(self, **kw):
from pfs.codetables import OriginRow
base = dict(
family="HOME-VISITS",
code="99341",
kind="statute",
reference="Section 1861(m) of the Act",
year=None,
summary="A paraphrase.",
cite_kind="fr_para",
item_key="K",
p_id=2,
page_label="100",
locator="FR p. 100, ¶2",
extractor="m",
)
return OriginRow(**{**base, **kw})
def _patch(self, monkeypatch, order):
rows = [
self._row(),
self._row(code="99350", kind="regulation", cite_kind="cfr"),
self._row(code="99342", cite_kind="cpt_book", item_key="B", p_id=0),
]
seen = {}
def fake(con, store, fam, **kw):
order.append("family_origin")
seen.update(kw, key=fam.key)
return rows, {"done": 2, "skipped": 1, "failed": 0, "failed_pending": 3}
monkeypatch.setattr(pfs_cli, "family_origin", fake)
monkeypatch.setattr(pfs_cli, "_asker", lambda: "ASK")
monkeypatch.setattr(
pfs_cli,
"resolve_book",
lambda store, key: ("Title " + key, Path("/x") / f"{key}.epub"),
)
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
return seen
def test_origin_cli_write_after_extraction(self, con, monkeypatch):
from pfs.codetables import read_origin, write_origin
order = []
seen = self._patch(monkeypatch, order)
orig_write = write_origin
def write(c, fam, rows):
order.append("write_origin")
return orig_write(c, fam, rows)
class _B:
def __enter__(self):
order.append("_batch")
return con._real
def __exit__(self, *a):
return False
monkeypatch.setattr(pfs_cli, "write_origin", write)
monkeypatch.setattr(pfs_cli, "_batch", lambda: _B())
monkeypatch.setattr(pfs_cli, "_publish", lambda: order.append("_publish"))
res = runner.invoke(
app,
["pfs", "origin", "--family", "home-visits", "--limit", "4", "--write"],
)
assert res.exit_code == 0, res.output
assert order == ["family_origin", "_batch", "write_origin", "_publish"]
assert (
seen["ask"] == "ASK" and seen["limit"] == 4 and seen["key"] == "HOME-VISITS"
)
assert seen["books"] == [
("3ANRT854", "Title 3ANRT854", Path("/x/3ANRT854.epub"))
]
assert seen["replay_model"] is None
assert "statute: 2" in res.output and "regulation: 1" in res.output
assert "cpt_book: 1" in res.output and "fr_para: 1" in res.output
assert "done 2, skipped 1, failed 0, failed_pending 3" in res.output
assert "wrote 3 origins" in res.output
assert len(read_origin(con._real, "HOME-VISITS")) == 3
def test_origin_without_write_writes_nothing(self, con, monkeypatch):
order = []
seen = self._patch(monkeypatch, order)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(
app,
["pfs", "origin", "--family", "HOME-VISITS", "--book", "A", "--book", "B"],
)
assert res.exit_code == 0, res.output
assert order == ["family_origin"] and "wrote" not in res.output
assert [b[0] for b in seen["books"]] == ["A", "B"]
def test_no_llm_replays_the_ledger(self, con, monkeypatch):
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
res = runner.invoke(
app, ["pfs", "origin", "--family", "HOME-VISITS", "--no-llm"]
)
assert res.exit_code == 0, res.output
assert seen["ask"] is None and seen["replay_model"] == "qwen-x"
def test_no_model_host_falls_back_to_replay(self, con, monkeypatch):
seen = self._patch(monkeypatch, [])
def down():
raise RuntimeError("no Ollama host serves qwen")
monkeypatch.setattr(pfs_cli, "_asker", down)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
res = runner.invoke(app, ["pfs", "origin", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert seen["ask"] is None and seen["replay_model"] == "qwen-x"
assert "replaying the ledger" in res.output
def test_origin_unknown_family_and_missing_book(self, con, monkeypatch):
res = runner.invoke(app, ["pfs", "origin", "--family", "NOPE"])
assert res.exit_code != 0 and "unknown curated family" in res.output
seen = self._patch(monkeypatch, [])
monkeypatch.setattr(pfs_cli, "resolve_book", lambda store, key: None)
# a book asked for by name must exist
res = runner.invoke(
app, ["pfs", "origin", "--family", "HOME-VISITS", "--book", "XX"]
)
assert res.exit_code != 0 and "no EPUB attachment" in res.output
# the default book missing only warns: the run goes on FR-only
res = runner.invoke(app, ["pfs", "origin", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert (
"warning: no EPUB attachment for 3ANRT854; skipping the book" in res.output
)
assert seen["books"] == []
def test_origin_dry_run_makes_no_model_calls(self, con, monkeypatch):
self._patch(monkeypatch, [])
monkeypatch.setattr(
pfs_cli,
"origin_inventory",
lambda store, fam, books: (
[{"unit": "a"}, {"unit": "b"}],
{"3ANRT854": ["12", "13", "20"]},
),
)
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "family_origin", lambda *a, **k: 1 / 0)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(
app, ["pfs", "origin", "--family", "HOME-VISITS", "--dry-run"]
)
assert res.exit_code == 0, res.output
assert "2 candidate Federal Register paragraphs" in res.output
assert "3ANRT854: 3 pages" in res.output and "no model calls made" in res.output
class TestCmsSummaries:
def _rows(self):
from pfs.codetables import CmsSummaryRow
return [
CmsSummaryRow(
"HOME-VISITS",
"K1",
2023,
"final",
1679,
1680,
"",
"majority",
"support",
"acknowledged",
"regex",
),
CmsSummaryRow(
"HOME-VISITS",
"K1",
2023,
"final",
1681,
1682,
"99341",
"few",
"modify",
"finalize_as_proposed",
"regex",
),
CmsSummaryRow(
"HOME-VISITS",
"K0",
2008,
"final",
2241,
2242,
"99326",
"unspecified",
"oppose",
"finalize_modified",
"regex",
),
]
def _patch(self, monkeypatch, seen, counts=None):
def fake(store, fam, **kw):
seen.update(kw, key=fam.key)
kw["counts"].update(
{"done": 2, "skipped": 1, "failed": 0, "failed_pending": 3}
if counts is None
else counts
)
return self._rows()
monkeypatch.setattr(pfs_cli, "cms_summaries_rows", fake)
monkeypatch.setattr(
pfs_cli, "_classifier", lambda **kw: "CLS" if kw == {"pin": True} else 1 / 0
)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
def test_writes_and_publishes(self, con, monkeypatch):
from pfs.codetables import read_cms_summaries
seen = {}
self._patch(monkeypatch, seen)
res = runner.invoke(
app,
[
"pfs",
"cms-summaries",
"--family",
"home-visits",
"--limit",
"4",
"--retry-failed",
"--write",
],
)
assert res.exit_code == 0, res.output
assert seen["classify"] == "CLS" and seen["model"] == "qwen-x"
assert (
seen["limit"] == 4 and seen["retry_failed"] and seen["replay_model"] is None
)
assert seen["key"] == "HOME-VISITS"
assert "CY2008 final: 1 summaries" in res.output
assert "CY2023 final: 2 summaries" in res.output
assert "done 2, skipped 1, failed 0, failed_pending 3" in res.output
assert "wrote 3 summaries" in res.output
assert con.published == [True]
assert len(read_cms_summaries(con._real, "HOME-VISITS")) == 3
def test_without_write_writes_nothing(self, con, monkeypatch):
self._patch(monkeypatch, {})
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(app, ["pfs", "cms-summaries", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert "wrote" not in res.output
def test_no_llm_builds_no_classifier_and_replays(self, con, monkeypatch):
seen = {}
self._patch(monkeypatch, seen, counts={})
monkeypatch.setattr(pfs_cli, "_classifier", lambda **kw: 1 / 0)
res = runner.invoke(
app, ["pfs", "cms-summaries", "--family", "HOME-VISITS", "--no-llm"]
)
assert res.exit_code == 0, res.output
assert seen["classify"] is None and seen["replay_model"] == "qwen-x"
assert "done 0, skipped 0, failed 0, failed_pending 0" in res.output
def test_unknown_family(self, con):
res = runner.invoke(app, ["pfs", "cms-summaries", "--family", "NOPE"])
assert res.exit_code != 0 and "unknown curated family" in res.output
class TestClassifierPin:
@pytest.mark.parametrize("pin,expect", [(True, "qwen-test"), (False, None)])
def test_classifier_pins_only_when_asked(self, monkeypatch, pin, expect):
import llm.classify as lc
import llm.pool as lp
from llm import config as llm_config
class Cfg:
instruct_model = "qwen-test"
got = {}
monkeypatch.setattr(llm_config, "load", lambda: Cfg())
monkeypatch.setattr(
lp.HostPool, "from_config", classmethod(lambda c, cfg: "POOL")
)
monkeypatch.setattr(
lc, "closed_vocab_classifier", lambda cfg, pool, **kw: got.update(kw) or "C"
)
assert pfs_cli._classifier(pin=pin) == "C"
assert got == {"model": expect}
class TestCommenters:
def _rows(self):
from pfs.codetables import CommenterRow
return [
CommenterRow(
"K1",
"D1",
"c1",
"AAFP",
"specialty_society",
"",
"",
0,
"both",
"Business",
"regs",
),
CommenterRow(
"K2",
"D1",
"c2",
"",
"individual",
"",
"",
0,
"regs_gov_fields",
"Individual",
"regs",
),
CommenterRow(
"K3",
"D1",
"c3",
"",
"individual",
"",
"",
0,
"regs_gov_fields",
"Individual",
"regs",
),
]
def _patch(self, monkeypatch, seen, pending=0):
def fake(engine, store, con_, fam, **kw):
seen.update(kw, key=fam.key)
kw["progress"]("K1", "done")
return self._rows(), {
"done": 1,
"skipped": 0,
"failed": 0,
"failed_pending": 0,
"pending": pending,
"blocked": 1,
}
monkeypatch.setattr(pfs_cli, "identify", fake)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_mirror", lambda: "MIRROR")
monkeypatch.setattr(pfs_cli, "_asker", lambda: "ASK")
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
def test_dry_run_lists_categories(self, con, monkeypatch):
from types import SimpleNamespace
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_mirror", lambda: "MIRROR")
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(
pfs_cli,
"family_comment_items",
lambda eng, codes, key, phrases, docket: [
("K1", "a"),
("K2", "b"),
("K3", "c"),
("K4", "d"),
("K5", "e"),
],
)
regs = dict(
zip(["K1", "K2", "K3", "K4"], ["Individual", "Individual", "", "Physician"])
)
monkeypatch.setattr(
pfs_cli,
"regs_for",
lambda store, mirror, docket, keys: (
{},
{k: SimpleNamespace(category=regs[k]) for k in keys if k in regs},
),
)
res = runner.invoke(
app,
[
"pfs",
"commenters",
"--family",
"home-visits",
"--docket",
"D1",
"--dry-run",
],
)
assert res.exit_code == 0, res.output
assert "D1: 5 items; no model calls made" in res.output
assert "unreadable mirror JSON: 1" in res.output
assert "category Individual: 2" in res.output
assert "category (none): 1" in res.output
assert "category Physician: 1" in res.output
def test_writes_and_publishes(self, con, monkeypatch):
from pfs.codetables import read_commenters
seen = {}
self._patch(monkeypatch, seen)
res = runner.invoke(
app,
[
"pfs",
"commenters",
"--family",
"home-visits",
"--docket",
"D1",
"--limit",
"5",
"--write",
],
)
assert res.exit_code == 0, res.output
assert seen["key"] == "HOME-VISITS" and seen["docket"] == "D1"
assert (
seen["limit"] == 5 and seen["ask"] == "ASK" and seen["mirror"] == "MIRROR"
)
assert "K1 done" in res.output
assert "individual: 2" in res.output and "specialty_society: 1" in res.output
assert "pending 0, blocked 1" in res.output
assert "wrote 3 commenters" in res.output
assert con.published == [True]
assert len(read_commenters(con._real)) == 3
def test_write_refused_while_pending(self, con, monkeypatch):
self._patch(monkeypatch, {}, pending=2)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(
app,
[
"pfs",
"commenters",
"--family",
"HOME-VISITS",
"--docket",
"D1",
"--write",
],
)
assert res.exit_code != 0 and "2 comments not yet identified" in res.output
def test_without_write_writes_nothing(self, con, monkeypatch):
self._patch(monkeypatch, {})
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(
app, ["pfs", "commenters", "--family", "HOME-VISITS", "--docket", "D1"]
)
assert res.exit_code == 0 and "commenters: 3;" in res.output
def test_unknown_family(self, con):
res = runner.invoke(
app, ["pfs", "commenters", "--family", "NOPE", "--docket", "D1"]
)
assert res.exit_code != 0
def test_mirror_indirection(self):
from bib.regulations_gov import Mirror
assert isinstance(pfs_cli._mirror(), Mirror)
ASK = SimpleNamespace(model="qwen-x")
#: a golden-set score shaped like the CY2023 measurement (R66a)
SCORES = Scores(100, 83, 88, 24, 10, {}, (), positions=27, found=15, right=10)
class TestStance:
"""``stack pfs stance`` and the reaction overlay (task 9)."""
def _patch(self, con, monkeypatch, *, pending=0, seen=None):
from bib.dockets import Docket
from pfs.codetables import CommenterRow, StanceRow, write_commenters
seen = {} if seen is None else seen
write_commenters(
con._real,
[
CommenterRow(k, "D1", "C-" + k, "", "individual", "", "", 0, "regs_gov_fields", "", "regs", "HOME-VISITS")
for k in ("K1", "K2")
],
) # fmt: skip
store = type(
"S",
(),
{"dockets": lambda self: [Docket(id="D1", comment_end_date="2022-09-06")]},
)()
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_asker", lambda large=False: ASK)
monkeypatch.setattr(
pfs_cli, "_golden_run", lambda data, method, **kw: (SCORES, {})
)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
from pfs.stance import EventChoice
monkeypatch.setattr(
pfs_cli,
"docket_events",
lambda con_, fam, year, store_, **kw: [
EventChoice("HOME-VISITS:2023:G0318", "g", year),
EventChoice("HOME-VISITS:2023:other", "o", year),
],
)
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
monkeypatch.setattr(
pfs_cli,
"family_comment_items",
lambda eng, codes, key, phrases, docket, **kw: [("K1", "t1"), ("K2", "t2"), ("K9", "t9")],
) # fmt: skip
def series(engine, store_, family, codes, **kw):
return [
ReactionRow(family, "D1", "docket", 2023, 2, 9, 0, 0, 0, 0, "[]"),
ReactionRow(family, "R1", "fr-pairs", 2023, 3, 7, 0, 0, 0, 0, "[]"),
]
monkeypatch.setattr(pfs_cli, "reaction_series", series)
def fake(commenters, texts, events, ask, log, **kw):
seen.update(
keys=[c.item_key for c in commenters],
texts=texts,
events=[e.event_key for e in events],
ask=ask,
**kw,
)
kw["progress"]("K1", "done")
rows = [
StanceRow("HOME-VISITS", "K1", "D1", "HOME-VISITS:2023:other", "oppose", "", "", "x"),
StanceRow("HOME-VISITS", "K2", "D1", "HOME-VISITS:2023:other", "support", "", "", "x"),
] # fmt: skip
return rows, {
"done": 2, "skipped": 0, "failed": 0, "failed_pending": 0,
"pending": pending, "no_text": 0,
} # fmt: skip
monkeypatch.setattr(pfs_cli, "stances_for", fake)
return seen
def _args(self, *extra):
return ["pfs", "stance", "--family", "home-visits", "--docket", "D1", *extra]
def test_write_replaces_docket_row_and_keeps_fr_pairs(self, con, monkeypatch):
from pfs.codetables import read_stances, write_reaction
seen = self._patch(con, monkeypatch)
write_reaction(
con._real,
"HOME-VISITS",
[
ReactionRow("HOME-VISITS", "D1", "docket", 2023, 2, 9, 0, 0, 0, 0, "[]"),
ReactionRow("HOME-VISITS", "D0", "docket", 2020, 1, 4, 0, 0, 0, 0, "[]"),
ReactionRow("HOME-VISITS", "R1", "fr-pairs", 2023, 3, 7, 0, 0, 0, 0, "[]"),
],
) # fmt: skip
res = runner.invoke(app, self._args("--limit", "3", "--write"))
assert res.exit_code == 0, res.output
assert seen["keys"] == ["K1", "K2"] and seen["texts"] == {
"K1": "t1",
"K2": "t2",
}
assert seen["events"] == ["HOME-VISITS:2023:G0318", "HOME-VISITS:2023:other"]
assert seen["limit"] == 3 and seen["ask"] is ASK
assert "wrote 2 stances" in res.output and "K1 done" in res.output
assert (
"HOME-VISITS:2023:other: support 1, modify 0, oppose 1, unclear 0"
in res.output
)
assert len(read_stances(con._real, "HOME-VISITS")) == 2
rows = {r.period: r for r in read_reaction(con._real, "HOME-VISITS")}
assert set(rows) == {"D1", "D0", "R1"}
assert (rows["D1"].stance_support, rows["D1"].stance_oppose) == (1, 1)
assert (rows["D1"].n_classified, rows["D1"].n_commenters) == (2, 2)
assert (rows["D0"].n_classified, rows["D0"].n_commenters) == (None, None)
assert list(json.loads(rows["D1"].events_json)) == ["HOME-VISITS:2023:other"]
assert rows["D1"].n_items == 2 and rows["D1"].n_total == 9
assert (rows["D0"].stance_support, rows["R1"].stance_oppose) == (0, 0)
assert rows["D0"].events_json == "{}"
assert con.published == [True]
def test_no_llm_replays_the_ledger(self, con, monkeypatch):
seen = self._patch(con, monkeypatch)
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
res = runner.invoke(app, self._args("--no-llm"))
assert res.exit_code == 0, res.output
assert seen["ask"] is None and seen["model"] == "qwen-x"
def test_write_with_no_stored_reaction_rows_uses_series(self, con, monkeypatch):
self._patch(con, monkeypatch)
res = runner.invoke(app, self._args("--write"))
assert res.exit_code == 0, res.output
assert {r.period for r in read_reaction(con._real, "HOME-VISITS")} == {
"D1",
"R1",
}
def test_write_refused_while_pending(self, con, monkeypatch):
self._patch(con, monkeypatch, pending=2)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(app, self._args("--write"))
assert res.exit_code != 0 and "2 comments not yet classified" in res.output
def test_without_write_writes_nothing(self, con, monkeypatch):
self._patch(con, monkeypatch)
monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0)
res = runner.invoke(app, self._args())
assert res.exit_code == 0 and "stances: 2;" in res.output
def test_dry_run_makes_no_model_calls(self, con, monkeypatch):
self._patch(con, monkeypatch)
monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
monkeypatch.setattr(pfs_cli, "stances_for", lambda *a, **k: 1 / 0)
res = runner.invoke(app, self._args("--dry-run"))
assert res.exit_code == 0, res.output
assert "D1: 2 commenters, 2 with comment text, 2 events" in res.output
assert "HOME-VISITS:2023:other" in res.output
assert "no model calls made" in res.output
def test_no_commenter_rows(self, con, monkeypatch):
from pfs.codetables import write_commenters
self._patch(con, monkeypatch)
write_commenters(con._real, [], docket="D1", family="HOME-VISITS")
res = runner.invoke(app, self._args("--write"))
assert res.exit_code == 0 and "D1: no commenter rows" in res.output
def test_unknown_family_and_docket(self, con, monkeypatch):
self._patch(con, monkeypatch)
res = runner.invoke(
app, ["pfs", "stance", "--family", "NOPE", "--docket", "D1"]
)
assert res.exit_code != 0
res = runner.invoke(
app, ["pfs", "stance", "--family", "home-visits", "--docket", "ZZ"]
)
assert res.exit_code != 0 and "unknown docket" in res.output
def test_rollup_ignores_cms_summaries(self, con, monkeypatch):
from pfs.codetables import CmsSummaryRow, write_cms_summaries
self._patch(con, monkeypatch)
monkeypatch.setattr(pfs_cli, "reaction_series", lambda e, s, f, c, **kw: [
ReactionRow(f, "D1", "docket", 2023, 2, 9, 0, 0, 0, 0, "[]")
]) # fmt: skip
def counts():
r = [
x for x in read_reaction(con._real, "HOME-VISITS") if x.period == "D1"
][0]
return (
r.stance_support,
r.stance_modify,
r.stance_oppose,
r.stance_unclear,
r.events_json,
)
assert runner.invoke(app, self._args("--write")).exit_code == 0
before = counts()
write_cms_summaries(
con._real,
"HOME-VISITS",
[CmsSummaryRow("HOME-VISITS", "R1", 2023, "final", 1, 2, "99342", "a majority of commenters", "support", "finalize_as_proposed", "regex")],
) # fmt: skip
res = runner.invoke(
app, ["pfs", "reaction", "--family", "home-visits", "--write"]
)
assert res.exit_code == 0, res.output
assert counts() == before
assert runner.invoke(app, self._args("--write")).exit_code == 0
assert counts() == before == (1, 0, 1, 0, before[4])
def _reaction_patch(self, monkeypatch):
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(
pfs_cli,
"reaction_series",
lambda e, s, f, c, **kw: [
ReactionRow(f, "D1", "docket", 2023, 5, 20, 0, 0, 0, 0, "[]"),
ReactionRow(f, "D2", "docket", 2024, 4, 10, 0, 0, 0, 0, "[]"),
],
)
def test_reaction_rerun_keeps_stance_counts(self, con, monkeypatch):
from pfs.codetables import StanceRow, write_stances
self._reaction_patch(monkeypatch)
from pfs.codetables import CommenterRow, write_commenters, write_reaction
write_commenters(
con._real,
[
CommenterRow(k, "D1", k, "", "individual", "", "", 0, "regs_gov_fields", "", "regs", "CCM")
for k in ("K1", "K2", "K3")
],
) # fmt: skip
ev = json.dumps({"cell": 0.83, "recall": 0.556, "in_sample": True})
write_reaction(
con._real,
"CCM",
[
ReactionRow("CCM", "D1", "docket", 2023, 5, 20, 0, 0, 0, 0, "[]", '{"CCM:2023:other": {}}',
stance_method="stance-v4:qwen2.5:32b", stance_agreement=0.83, stance_eval=ev),
],
) # fmt: skip
write_stances(
con._real,
"CCM",
"D1",
[
StanceRow("CCM", "K1", "D1", "CCM:2023:other", "modify", "", "", "x"),
StanceRow("CCM", "K2", "D1", "CCM:2023:other", "modify", "", "", "x"),
],
) # fmt: skip
res = runner.invoke(app, ["pfs", "reaction", "--family", "ccm", "--write"])
assert res.exit_code == 0, res.output
rows = {r.period: r for r in read_reaction(con._real, "CCM")}
assert rows["D1"].stance_modify == 2 and rows["D1"].n_items == 5
assert (rows["D1"].n_classified, rows["D1"].n_commenters) == (2, 3)
assert (rows["D2"].n_classified, rows["D2"].n_commenters) == (None, None)
assert list(json.loads(rows["D1"].events_json)) == ["CCM:2023:other"]
assert rows["D2"].stance_modify == 0 and rows["D2"].events_json == "{}"
# R66a: a reaction rerun keeps the stored row's stance method record
d1 = rows["D1"]
assert (d1.stance_method, d1.stance_agreement, d1.stance_eval) == (
"stance-v4:qwen2.5:32b",
0.83,
ev,
)
assert (rows["D2"].stance_method, rows["D2"].stance_eval) == (None, None)
def test_stance_sample_refused_when_stances_stored(self, con, monkeypatch):
from pfs.codetables import StanceRow, write_stances
self._reaction_patch(monkeypatch)
monkeypatch.setattr(pfs_cli, "_classifier", lambda **k: 1 / 0)
write_stances(
con._real, "CCM", "D1", [StanceRow("CCM", "K1", "D1", "CCM:2023:other", "modify", "", "", "x")]
) # fmt: skip
res = runner.invoke(
app, ["pfs", "reaction", "--family", "ccm", "--stance-sample", "5"]
)
assert res.exit_code != 0 and "stored stances" in res.output
def test_reaction_tolerates_a_missing_stance_table(self, con, monkeypatch):
self._reaction_patch(monkeypatch)
con._real.execute("DROP TABLE pfs.comment_stance")
res = runner.invoke(app, ["pfs", "reaction", "--family", "ccm"])
assert res.exit_code == 0, res.output
def test_stored_tolerates_only_a_missing_table(self):
import duckdb as dd
c = dd.connect(":memory:")
assert pfs_cli._stored(pfs_cli.read_stances, c, "F") == []
with pytest.raises(RuntimeError):
pfs_cli._stored(
lambda con, f: (_ for _ in ()).throw(RuntimeError("x")), c, "F"
)
def test_method_option_is_passed_and_validated(self, con, monkeypatch):
seen = self._patch(con, monkeypatch)
res = runner.invoke(app, self._args("--method", "per-event"))
assert res.exit_code == 0 and seen["method"] == "per-event", res.output
res = runner.invoke(app, self._args())
assert seen["method"] == pfs_cli.DEFAULT_METHOD
res = runner.invoke(app, self._args("--method", "nope"))
assert res.exit_code != 0 and "--method must be one of" in res.output
def test_text_window_is_the_wide_one(self, con, monkeypatch):
seen = self._patch(con, monkeypatch)
monkeypatch.setattr(
pfs_cli,
"family_comment_items",
lambda eng, codes, key, phrases, **kw: seen.update(window=kw) or [],
)
runner.invoke(app, self._args("--dry-run"))
assert seen["window"]["max_chars"] == pfs_cli.TEXT_CHARS == 6000
assert seen["window"]["max_chunks"] == pfs_cli.TEXT_CHUNKS
class TestStanceEval:
GOLDEN = {
"family": "HOME-VISITS",
"docket": "D1",
"items": [
{
"item_key": "K1",
"cells": {
"HOME-VISITS:2023:G0318": {"label": "support"},
"HOME-VISITS:2023:other": {"label": "none"},
},
},
{
"item_key": "K2",
"cells": {
"HOME-VISITS:2023:G0318": {"label": "oppose"},
"HOME-VISITS:2023:other": {"label": "none"},
},
},
],
}
def _patch(self, monkeypatch, tmp_path, *, pending=0, no_text=0, golden=None):
from bib.dockets import Docket
from pfs.codetables import StanceRow
from pfs.stance import EventChoice
path = tmp_path / "g.json"
path.write_text(json.dumps(golden or self.GOLDEN))
seen = {}
store = type(
"S",
(),
{"dockets": lambda self: [Docket(id="D1", comment_end_date="2022-09-06")]},
)()
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(
pfs_cli,
"_asker",
lambda large=False: (
seen.update(large=large) or type("A", (), {"model": "qwen-x"})()
),
)
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
monkeypatch.setattr(
pfs_cli,
"docket_events",
lambda con_, fam, year, store_, **kw: [
EventChoice("HOME-VISITS:2023:G0318", "g", year),
EventChoice("HOME-VISITS:2023:other", "o", year),
],
)
monkeypatch.setattr(
pfs_cli,
"family_comment_items",
lambda eng, codes, key, phrases, **kw: [("K1", "t1"), ("K2", "t2"), ("K9", "t")],
) # fmt: skip
def fake(commenters, texts, events, ask, log, **kw):
seen.update(keys=[c.item_key for c in commenters], texts=texts, **kw)
kw["progress"]("K1", "done")
rows = [
StanceRow("HOME-VISITS", "K1", "D1", "HOME-VISITS:2023:G0318", "support", "", "", "x"),
StanceRow("HOME-VISITS", "K2", "D1", "HOME-VISITS:2023:G0318", "support", "", "", "x"),
] # fmt: skip
return rows, {
"done": 2, "skipped": 0, "failed": 0, "failed_pending": 0,
"pending": pending, "no_text": no_text,
} # fmt: skip
monkeypatch.setattr(pfs_cli, "stances_for", fake)
return path, seen
def _run(self, path, *extra):
return runner.invoke(app, ["pfs", "stance-eval", "--golden", str(path), *extra])
MULTI = {
"family": "HOME-VISITS",
"dockets": ["D1", "D2"],
"items": [
{
"item_key": "K1",
"docket": "D1",
"cells": {"HOME-VISITS:2023:G0318": {"label": "support"}},
},
{
"item_key": "K2",
"docket": "D2",
"cells": {"HOME-VISITS:2026:G2211": {"label": "support"}},
},
],
}
def test_multi_docket_golden_scores_each_item_against_its_docket(
self, con, monkeypatch, tmp_path
):
from bib.dockets import Docket
from pfs.codetables import StanceRow
from pfs.stance import EventChoice
path, _ = self._patch(monkeypatch, tmp_path, golden=self.MULTI)
store = type(
"S",
(),
{
"dockets": lambda self: [
Docket(id="D1", comment_end_date="2022-09-06"),
Docket(id="D2", comment_end_date="2025-09-12"),
]
},
)()
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
years = []
monkeypatch.setattr(
pfs_cli,
"docket_events",
lambda con_, fam, year, store_, **kw: (
years.append(year)
or [EventChoice(f"HOME-VISITS:{year}:other", "o", year)]
),
)
seen = []
monkeypatch.setattr(
pfs_cli,
"family_comment_items",
lambda eng, codes, key, phrases, docket, **kw: [
("K1", "t1"), ("K2", "t2")
],
) # fmt: skip
def fake(commenters, texts, events, ask, log, **kw):
seen.append(([c.item_key for c in commenters], kw["docket"], kw["limit"]))
k = commenters[0].item_key
y = events[0].rule_year
return (
[StanceRow("HOME-VISITS", k, kw["docket"], f"HOME-VISITS:{y}:G0318" if k == "K1" else f"HOME-VISITS:{y}:G2211", "support", "", "", "x")],
{"done": 1, "skipped": 0, "failed": 0, "failed_pending": 0, "pending": 0, "no_text": 0},
) # fmt: skip
monkeypatch.setattr(pfs_cli, "stances_for", fake)
res = self._run(path, "--limit", "1")
assert res.exit_code == 0, res.output
assert years == [2023, 2026] # each docket's own rule year
assert seen[0] == (["K1"], "D1", 1)
assert seen[1][:2] == (["K2"], "D2")
assert "2 new calls" in res.output
assert "recall 1.000 (2/2)" in res.output
# the call limit is shared: D2 only replays once D1 spent it
assert seen[1][2] == 0
def test_multi_docket_unknown_docket(self, con, monkeypatch, tmp_path):
path, _ = self._patch(monkeypatch, tmp_path, golden=self.MULTI)
res = self._run(path)
assert "unknown docket 'D2'" in res.output
def test_scores_and_prints_the_matrix(self, con, monkeypatch, tmp_path):
path, seen = self._patch(monkeypatch, tmp_path)
res = self._run(
path, "--method", "per-event", "--model", "large", "--limit", "7"
)
assert res.exit_code == 0, res.output
assert seen["large"] is True and seen["method"] == "per-event"
assert seen["limit"] == 7 and seen["keys"] == ["K1", "K2"]
assert seen["texts"] == {"K1": "t1", "K2": "t2"} # only golden items
assert "per-event on qwen-x: 2 new calls" in res.output
assert "cells 4: agreement 0.750 (3)" in res.output
assert "presence 1.000 (4)" in res.output
assert "firm accuracy 0.500 (1/2); gate FAIL" in res.output
assert "recall 1.000 (2/2); direction 0.500 (1/2)" in res.output
assert "confusion (rows golden, columns model)" in res.output
assert "disagree K2 G0318: golden oppose, model support" in res.output
def test_incomplete_run_prints_no_scores(self, con, monkeypatch, tmp_path):
path, _ = self._patch(monkeypatch, tmp_path, pending=3)
res = self._run(path)
assert res.exit_code == 0 and "incomplete: rerun to resume" in res.output
assert "agreement" not in res.output
def test_validation(self, con, monkeypatch, tmp_path):
path, _ = self._patch(monkeypatch, tmp_path)
assert "--method must be one of" in self._run(path, "--method", "x").output
assert (
"--model must be instruct or large"
in self._run(path, "--model", "x").output
)
path, _ = self._patch(
monkeypatch, tmp_path, golden={**self.GOLDEN, "family": "NOPE"}
)
assert "unknown curated family" in self._run(path).output
path, _ = self._patch(
monkeypatch, tmp_path, golden={**self.GOLDEN, "docket": "ZZ"}
)
assert "unknown docket" in self._run(path).output
class TestStanceMethodRecord:
def test_v5_generation_is_recorded_and_threaded(self, con, monkeypatch):
t = TestStance()
seen = t._patch(con, monkeypatch)
from pfs.stance import EventChoice
flags = []
def events(c_, f, y, s_, **kw):
flags.append(kw["guidelines"])
return [
EventChoice("HOME-VISITS:2023:G0318", "g", y),
EventChoice("HOME-VISITS:2023:other", "o", y),
]
monkeypatch.setattr(pfs_cli, "docket_events", events)
res = runner.invoke(
app, t._args("--model", "large", "--generation", "stance-v5", "--write")
)
assert res.exit_code == 0, res.output
assert flags == [True] and seen["generation"] == "stance-v5"
d1 = {r.period: r for r in read_reaction(con._real, "HOME-VISITS")}["D1"]
assert d1.stance_method == "stance-v5:qwen-x"
def test_write_records_method_and_golden_agreement(self, con, monkeypatch):
t = TestStance()
seen = t._patch(con, monkeypatch)
res = runner.invoke(app, t._args("--model", "large", "--write"))
assert res.exit_code == 0, res.output
got = {r.period: r for r in read_reaction(con._real, "HOME-VISITS")}
d1 = got["D1"]
assert d1.stance_method == f"{pfs_cli.prompt_version('multi')}:qwen-x"
# R82: the measurement is the out-of-sample record, the tuning set secondary
assert d1.stance_agreement is None
ev = json.loads(d1.stance_eval)
assert ev["in_sample"] is False
assert ev["generation"] == "stance-v5"
assert ev["in_sample_record"]["docket"] == "CMS-2022-0113"
assert got["R1"].stance_method is None and got["R1"].stance_eval is None
assert seen["ask"] is ASK
res = runner.invoke(app, t._args("--model", "huge"))
assert res.exit_code != 0 and "--model must be instruct or large" in res.output
def test_other_family_tuning_set_adds_no_record(self, con, monkeypatch, tmp_path):
"""A tuning set of another family is never replayed nor recorded."""
t = TestStance()
t._patch(con, monkeypatch)
g = tmp_path / "g.json"
g.write_text(json.dumps({"family": "CCM", "docket": "D9", "items": []}))
res = runner.invoke(app, t._args("--golden", str(g), "--write"))
assert res.exit_code == 0, res.output
d1 = {r.period: r for r in read_reaction(con._real, "HOME-VISITS")}["D1"]
ev = json.loads(d1.stance_eval)
assert ev["in_sample"] is False and "in_sample_record" not in ev
assert d1.stance_agreement is None
def test_large_model_builds_the_large_asker(self, con, monkeypatch):
t = TestStance()
t._patch(con, monkeypatch)
calls = []
monkeypatch.setattr(
pfs_cli, "_asker", lambda large=False: calls.append(large) or ASK
)
runner.invoke(app, t._args("--model", "large"))
runner.invoke(app, t._args("--model", "instruct"))
runner.invoke(app, t._args()) # the default model is the large one
assert calls == [True, False, pfs_cli.DEFAULT_MODEL == "large"]
def _sets(self, tmp_path, family="HOME-VISITS", generation="stance-v5"):
prim = tmp_path / "p.json"
prim.write_text(json.dumps({
"family": family, "dockets": ["H1", "H2"], "in_sample": False,
"role": "primary", "generation": generation, "items": [{}, {}, {}],
})) # fmt: skip
foot = tmp_path / "f.json"
foot.write_text(json.dumps({
"family": family, "docket": "H3", "in_sample": False,
"role": "footnote", "items": [{}],
})) # fmt: skip
tuning = tmp_path / "t.json"
tuning.write_text(json.dumps({
"family": family, "docket": "T1", "items": [{}, {}],
})) # fmt: skip
return tuning, (foot, prim)
def test_eval_is_out_of_sample_with_the_tuning_record(self, monkeypatch, tmp_path):
tuning, held = self._sets(tmp_path)
seen = []
def run(data, method, **kw):
seen.append(data["items"].__len__())
return SCORES, {}
monkeypatch.setattr(pfs_cli, "_golden_run", run)
got = json.loads(
pfs_cli._method_eval(tuning, "multi", "m", "HOME-VISITS", "stance-v5", held)
)
assert seen == [3, 2] # primary then tuning; the footnote is not stamped
assert got == {
"cell": 0.83, "recall": 0.556, "direction": 0.667, "n_items": 3,
"n_positions": 27, "family": "HOME-VISITS", "dockets": ["H1", "H2"],
"in_sample": False, "generation": "stance-v5",
"in_sample_record": {
"cell": 0.83, "recall": 0.556, "direction": 0.667, "n_items": 2,
"n_positions": 27, "docket": "T1",
},
} # fmt: skip
def test_eval_is_none_without_a_matching_primary(self, monkeypatch, tmp_path):
monkeypatch.setattr(pfs_cli, "_golden_run", lambda *a, **k: 1 / 0)
tuning, held = self._sets(tmp_path)
F = "HOME-VISITS"
# another family, another generation, no files: no measurement
assert pfs_cli._method_eval(tuning, "multi", "m", "CCM", None, held) is None
assert pfs_cli._method_eval(tuning, "multi", "m", F, "stance-v4", held) is None
assert pfs_cli._method_eval(tuning, "multi", "m", F, None, ()) is None
assert (
pfs_cli._method_eval(
tuning, "multi", "m", F, None, (tmp_path / "gone.json",)
)
is None
)
def test_eval_without_tuning_set_omits_the_record(self, monkeypatch, tmp_path):
_, held = self._sets(tmp_path)
other = tmp_path / "ccm"
other.mkdir()
monkeypatch.setattr(pfs_cli, "_golden_run", lambda d, m, **kw: (SCORES, {}))
for tuning in (tmp_path / "none.json", self._sets(other, "CCM")[0]):
got = json.loads(
pfs_cli._method_eval(tuning, "multi", "m", "HOME-VISITS", None, held)
)
assert "in_sample_record" not in got
def test_partial_ledger_fails_loudly(self, monkeypatch, tmp_path):
tuning, held = self._sets(tmp_path)
monkeypatch.setattr(pfs_cli, "_golden_run", lambda d, m, **kw: (None, {}))
with pytest.raises(typer.BadParameter, match="held-out"):
pfs_cli._method_eval(tuning, "multi", "m", "HOME-VISITS", None, held)
calls = iter([(SCORES, {}), (None, {})])
monkeypatch.setattr(pfs_cli, "_golden_run", lambda d, m, **kw: next(calls))
with pytest.raises(typer.BadParameter, match="tuning"):
pfs_cli._method_eval(tuning, "multi", "m", "HOME-VISITS", None, held)
def test_model_of_and_slow_post(self, monkeypatch):
import httpx
assert pfs_cli._model_of(SimpleNamespace(model="a"), True) == "a"
monkeypatch.setattr(
pfs_cli, "_model_name", lambda large=False: "L" if large else "S"
)
assert pfs_cli._model_of(None, True) == "L"
assert pfs_cli._model_of(None, False) == "S"
class Client:
def __init__(self, timeout):
seen["timeout"] = timeout
def __enter__(self):
return self
def __exit__(self, *a):
return False
def post(self, url, json):
return SimpleNamespace(
raise_for_status=lambda: None, json=lambda: {"u": url}
)
seen = {}
monkeypatch.setattr(httpx, "Client", Client)
assert pfs_cli._slow_post("http://h", {}) == {"u": "http://h"}
assert seen["timeout"].read == 600.0
def test_model_name_large(self, monkeypatch):
from llm import config as llm_config
monkeypatch.setattr(llm_config, "load", lambda: _Cfg())
assert pfs_cli._model_name() == "qwen-14"
assert pfs_cli._model_name(large=True) == "qwen-32"
class TestStanceEvalOffline:
def test_no_llm_and_no_relevance_reach_the_run(self, monkeypatch, tmp_path):
seen = {}
def run(data, method, **kw):
seen.update(method=method, **kw)
return None, {
"done": 0,
"skipped": 0,
"failed": 0,
"failed_pending": 0,
"pending": 1,
"no_text": 0,
}
monkeypatch.setattr(pfs_cli, "_golden_run", run)
monkeypatch.setattr(pfs_cli, "_asker", lambda large=False: 1 / 0)
monkeypatch.setattr(
pfs_cli, "_model_name", lambda large=False: "L" if large else "S"
)
g = tmp_path / "g.json"
g.write_text("{}")
res = runner.invoke(
app,
[
"pfs",
"stance-eval",
"--golden",
str(g),
"--no-llm",
"--no-relevance",
"--model",
"large",
],
)
assert res.exit_code == 0, res.output
assert (
seen["ask"] is None and seen["name"] == "L" and seen["relevance"] is False
)
assert "multi on L (no relevance filter)" in res.output
assert "incomplete" in res.output
class TestGuidelineFacts:
def _row(self, **kw):
from pfs.codetables import GuidelineFactRow
base = dict(
family="HOME-VISITS", code="99341", edition_year=2024,
fact="selection_basis", value="mdm_or_time", cite_item="B",
cite_locator="p. 1", extractor="cpt-book",
) # fmt: skip
return GuidelineFactRow(**{**base, **kw})
def _patch(self, monkeypatch, seen, asker=lambda: "ASK"):
rows = [
self._row(),
self._row(code="99342"),
self._row(fact="mdm_level", value="low"),
]
def fake(con, store, fam, **kw):
seen.update(kw, key=fam.key)
kw["counts"].update(
done=2, skipped=1, failed=0, failed_pending=0, unpaged=0
)
return rows
monkeypatch.setattr(pfs_cli, "guideline_facts", fake)
monkeypatch.setattr(pfs_cli, "_asker", asker)
monkeypatch.setattr(pfs_cli, "_model_name", lambda large=False: "qwen-x")
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
def test_write(self, con, monkeypatch):
from pfs.codetables import read_guideline_facts
seen = {}
self._patch(monkeypatch, seen)
res = runner.invoke(
app,
[
"pfs",
"guideline-facts",
"--family",
"home-visits",
"--limit",
"3",
"--write",
],
)
assert res.exit_code == 0, res.output
assert (
seen["ask"] == "ASK" and seen["limit"] == 3 and seen["replay_model"] is None
)
assert (
"2024 selection_basis: 2" in res.output
and "2024 mdm_level: 1" in res.output
)
assert "facts: 3; done 2, skipped 1" in res.output
assert "wrote 3 facts" in res.output and con.published
assert len(read_guideline_facts(con._real, "HOME-VISITS")) == 3
def test_no_write_and_replay(self, con, monkeypatch):
seen = {}
self._patch(monkeypatch, seen)
res = runner.invoke(
app, ["pfs", "guideline-facts", "--family", "HOME-VISITS", "--no-llm"]
)
assert res.exit_code == 0 and "wrote" not in res.output
assert seen["ask"] is None and seen["replay_model"] == "qwen-x"
def test_no_host_replays(self, con, monkeypatch):
seen = {}
def down():
raise RuntimeError("no hosts")
self._patch(monkeypatch, seen, asker=down)
res = runner.invoke(app, ["pfs", "guideline-facts", "--family", "HOME-VISITS"])
assert res.exit_code == 0 and "no model host available" in res.output
assert seen["ask"] is None
def test_unknown_family(self, con):
res = runner.invoke(app, ["pfs", "guideline-facts", "--family", "NOPE"])
assert res.exit_code != 0
class TestStanceGeneration:
def test_bad_generation(self, con):
for cmd, extra in (
("stance", ["--docket", "D1"]),
("stance-eval", ["--golden", "x.json"]),
):
res = runner.invoke(
app,
["pfs", cmd, "--generation", "stance-v9", *extra]
+ (["--family", "HOME-VISITS"] if cmd == "stance" else []),
)
assert res.exit_code != 0 and "--generation" in res.output
class TestPsps:
_REC = {
"HCPCS_CD": "99350",
"PROVIDER_SPEC_CD": "50",
"PSPS_SUBMITTED_SERVICE_CNT": "10",
"PSPS_ALLOWED_CHARGE_AMT": "100",
"PSPS_NCH_PAYMENT_AMT": "80",
"PSPS_ERROR_IND_CD": "0",
}
@pytest.fixture
def env(self, con, monkeypatch, tmp_path):
from bib.store import Store
store = Store(":memory:", storage_dir=tmp_path / "storage")
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(
pfs_cli,
"psps_cache_path",
lambda y, code, root=None: tmp_path / str(y) / f"{code}.json",
)
con.execute("CREATE SCHEMA IF NOT EXISTS terminology")
con.execute(
"CREATE TABLE terminology.cms_acceptable_provider_specialty_codes "
"(specialty_code VARCHAR, specialty_description VARCHAR, "
"pecos_specialty_description VARCHAR)"
)
con.execute(
"INSERT INTO terminology.cms_acceptable_provider_specialty_codes "
"VALUES ('50', 'Nurse Practitioner', 'NP')"
)
fetched: list[tuple[int, str]] = []
def fake_fetch(client, y, code, **kw):
fetched.append((y, code))
return [dict(self._REC, HCPCS_CD=code)] if code == "99350" else []
monkeypatch.setattr(pfs_cli, "fetch_psps", fake_fetch)
monkeypatch.setattr(pfs_cli, "_http_dossier", lambda: MagicMock())
import time
monkeypatch.setattr(time, "sleep", lambda s: None)
try:
yield store, fetched, tmp_path
finally:
store.close()
def test_write_fetches_every_family_code_loads_cites_logs(self, con, env):
store, fetched, tmp_path = env
res = runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
assert res.exit_code == 0, res.output
assert len(fetched) == 20 and {y for y, _ in fetched} == {2023}
rows = con.execute(
"SELECT hcpcs, specialty_cd, specialty, services, allowed, paid, item_key "
"FROM pfs.utilization_specialty"
).fetchall()
assert rows[0][:6] == ("99350", "50", "Nurse Practitioner", 10.0, 100.0, 80.0)
assert "year:2023" in store.get(rows[0][6]).tags
log = con.execute(
"SELECT module, table_name, rows, source_file FROM cms.ingest_log "
"WHERE source_file LIKE '%99350.json'"
).fetchone()
assert log[:3] == ("pfs.psps", "pfs.utilization_specialty", 1)
assert (tmp_path / "2023" / "99350.json").exists()
assert con.published == [True]
assert "2023 99350: 1 lines -> 1 specialties" in res.output
assert "2023 99349: no lines, kept existing rows" in res.output
def test_offline_uses_cache_and_skips_missing(self, con, env):
store, fetched, tmp_path = env
runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
n = len(fetched)
res = runner.invoke(
app,
[
"pfs",
"psps",
"--family",
"HOME-VISITS",
"--write",
"--offline",
"--year",
"2023",
],
)
assert res.exit_code == 0, res.output
assert len(fetched) == n # no new fetch
assert "2023 99349: no cached file, skipped" in res.output
assert con.execute(
"SELECT count(*) FROM pfs.utilization_specialty"
).fetchone() == (1,)
def test_psps_cli_offline_uses_cache(self, con, env):
# the brief's named test: a cached file alone is enough offline
store, fetched, tmp_path = env
from pfs.psps import save_cache
save_cache(tmp_path / "2022" / "99350.json", [self._REC])
res = runner.invoke(
app,
[
"pfs",
"psps",
"--family",
"HOME-VISITS",
"--write",
"--offline",
"--year",
"2022",
],
)
assert res.exit_code == 0, res.output
assert fetched == []
assert con.execute(
"SELECT year, hcpcs, services FROM pfs.utilization_specialty"
).fetchall() == [(2022, "99350", 10.0)]
def test_empty_fetch_keeps_rows_and_logs_nothing(self, con, env, monkeypatch):
store, fetched, tmp_path = env
runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
before = con.execute("SELECT count(*) FROM cms.ingest_log").fetchone()[0]
monkeypatch.setattr(pfs_cli, "fetch_psps", lambda c, y, code, **k: [])
res = runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
assert res.exit_code == 0, res.output
assert con.execute(
"SELECT count(*) FROM pfs.utilization_specialty"
).fetchone() == (1,)
assert con.execute("SELECT count(*) FROM cms.ingest_log").fetchone() == (
before,
)
def test_4xx_stops_the_source_keeps_staged_rows_and_notes_the_gap(
self, con, env, monkeypatch
):
import httpx
calls: list[str] = []
def fetch(client, y, code, **kw):
calls.append(code)
if code == "99326":
raise httpx.HTTPStatusError(
"x", request=MagicMock(), response=httpx.Response(429)
)
return [dict(self._REC, HCPCS_CD=code)]
monkeypatch.setattr(pfs_cli, "fetch_psps", fetch)
res = runner.invoke(
app,
[
"pfs",
"psps",
"--family",
"HOME-VISITS",
"--write",
"--year",
"2023",
"--year",
"2024",
],
)
assert res.exit_code == 0, res.output
assert "gap: 2023 99326: HTTP 429" in res.output
assert calls[-1] == "99326"
n = con.execute("SELECT count(*) FROM pfs.utilization_specialty").fetchone()[0]
assert n == len(calls) - 1 # codes before the 429 were kept, no 2024 pass
assert con.published == [True]
def test_offline_reads_the_newest_snapshot(self, con, env):
import os
store, fetched, tmp_path = env
old = tmp_path / "2022" / "99350.json"
new = tmp_path / "2022" / "99350.abcd1234.json"
old.parent.mkdir(parents=True)
old.write_text(json.dumps([self._REC]))
new.write_text(json.dumps([dict(self._REC, PSPS_SUBMITTED_SERVICE_CNT="77")]))
os.utime(old, (1, 1))
res = runner.invoke(
app,
[
"pfs",
"psps",
"--family",
"HOME-VISITS",
"--write",
"--offline",
"--year",
"2022",
],
)
assert res.exit_code == 0, res.output
assert con.execute(
"SELECT services FROM pfs.utilization_specialty"
).fetchone() == (77.0,)
assert (
con.execute("SELECT source_file FROM cms.ingest_log")
.fetchone()[0]
.endswith("99350.abcd1234.json")
)
def test_names_missing_when_terminology_table_absent(self, con, env):
con.execute("DROP TABLE terminology.cms_acceptable_provider_specialty_codes")
res = runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
assert res.exit_code == 0, res.output
assert con.execute(
"SELECT specialty FROM pfs.utilization_specialty"
).fetchone() == ("",)
def test_terminology_error_other_than_missing_table_propagates(self, monkeypatch):
class Bad:
def execute(self, *a):
raise RuntimeError("boom")
with pytest.raises(RuntimeError):
pfs_cli._psps_names(Bad())
def test_unknown_year_and_family_rejected_before_any_fetch(self, con, env):
store, fetched, _ = env
res = runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "1999"]
)
assert res.exit_code != 0
res = runner.invoke(app, ["pfs", "psps", "--family", "NOPE", "--write"])
assert res.exit_code != 0 and "unknown family" in res.output
assert fetched == []
def test_read_prints_stored_rows_without_the_write_lock(
self, con, env, monkeypatch
):
runner.invoke(
app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"]
)
def fail_batch():
raise AssertionError("must not open a RW connection")
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"])
assert res.exit_code == 0, res.output
assert "Nurse Practitioner" in res.output and "psps: 1 rows" in res.output
def test_read_prints_dash_for_suppressed_and_reports_missing_table(self, con, env):
con.execute(
"INSERT INTO pfs.utilization_specialty VALUES "
"(2023, '99350', '50', 'NP', NULL, NULL, NULL, 'd', 'K', NULL, 2), "
"(2023, '99350', '08', 'FP', NULL, NULL, NULL, 'd', 'K', 1234, 2)"
)
res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"])
assert res.exit_code == 0 and " -" in res.output
assert ">=1,234" in res.output
con.execute("DROP TABLE pfs.utilization_specialty")
res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"])
assert "no PSPS table yet" in res.output
def test_read_reraises_other_errors(self, con, env, monkeypatch):
def boom(*a, **k):
raise RuntimeError("boom")
monkeypatch.setattr(pfs_cli, "read_psps_rows", boom)
res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"])
assert res.exit_code != 0 and isinstance(res.exception, RuntimeError)
def test_http_client_sends_the_honest_agent(self):
c = pfs_cli._http_dossier()
try:
assert (
c.headers["User-Agent"]
== "stack-pfs-dossier/1.0"
== pfs_cli.PSPS_USER_AGENT
)
finally:
c.close()
class TestImpacts:
_NAME = (
"CY 2027 PFS Final Rule Estimated Impact on Total Allowed Charges "
"by Specialty_508.txt"
)
@pytest.fixture
def sources(self, con, monkeypatch, tmp_path):
from pfs.stakes import ImpactSource
fix = Path(__file__).parents[1] / "pfs" / "fixtures"
p508 = tmp_path / self._NAME
p508.write_text((fix / "impact_cy2027_508.txt").read_text())
pfr = tmp_path / "2016-26668.txt"
pfr.write_text((fix / "impact_cy2017_fr.txt").read_text())
monkeypatch.setattr(pfs_cli, "_store", lambda: object())
monkeypatch.setattr(
pfs_cli,
"discover_508",
lambda store, root: [ImpactSource(p508, 2027, "final", "BJG", "508")],
)
monkeypatch.setattr(
pfs_cli,
"discover_fr",
lambda store: [ImpactSource(pfr, 2017, "final", "FRK", "fr")],
)
return p508, pfr
def test_scan_write_loads_both_kinds_and_logs(self, con, sources):
res = runner.invoke(app, ["pfs", "impacts", "--scan", "--write"])
assert res.exit_code == 0, res.output
got = con.execute(
"SELECT rule_year, vintage, source, count(*) FROM pfs.specialty_impact "
"GROUP BY ALL ORDER BY 1"
).fetchall()
assert got == [(2017, "final", "fr", 4), (2027, "final", "508", 6)]
logs = con.execute(
"SELECT module, table_name, rows, pincite_key FROM cms.ingest_log ORDER BY rows"
).fetchall()
assert logs == [
("pfs.stakes", "pfs.specialty_impact", 4, "FRK"),
("pfs.stakes", "pfs.specialty_impact", 6, "BJG"),
]
assert con.published == [True]
assert "CY2027 final (508): 6 rows" in res.output
def test_scan_without_write_only_reports(self, con, sources):
res = runner.invoke(app, ["pfs", "impacts", "--scan"])
assert res.exit_code == 0, res.output
assert "CY2017 final (fr): 4 rows" in res.output
assert con.published == []
assert con.execute("SELECT count(*) FROM pfs.specialty_impact").fetchone() == (
0,
)
def test_path_form_reads_one_file(self, con, sources):
p508, pfr = sources
res = runner.invoke(
app,
["pfs", "impacts", "--path", str(p508), "--year", "2027", "--write"],
)
assert res.exit_code == 0, res.output
res = runner.invoke(
app,
[
"pfs",
"impacts",
"--path",
str(pfr),
"--year",
"2016",
"--vintage",
"proposed",
"--write",
],
)
assert res.exit_code == 0, res.output
assert con.execute(
"SELECT rule_year, vintage, source, count(*) FROM pfs.specialty_impact "
"GROUP BY ALL ORDER BY 1"
).fetchall() == [(2016, "proposed", "fr", 4), (2027, "final", "508", 6)]
def test_argument_errors(self, con, sources):
p508, _ = sources
for args in (
[],
["--scan", "--path", str(p508)],
["--path", str(p508)],
["--scan", "--vintage", "draft"],
):
res = runner.invoke(app, ["pfs", "impacts", *args])
assert res.exit_code != 0, args
class TestSeats:
URL = "https://www.ama-assn.org/about/rvs-update-committee-ruc/composition-rvs-update-committee-ruc"
CPT_URL = "https://www.ama-assn.org/about/cpt-editorial-panel/purpose-cpt-coding-system-cpt-editorial-panel"
# invented text shaped like the AMA RUC page: seat list then members list
_RUC = (
"<html><body><nav>menu text</nav><main><h1>Composition</h1>"
"<p>The RUC represents the profession, with 3 of its 9 members appointed. "
"The RUC also includes 1 seat that rotates.</p>"
"<p>Chair</p><p>American Medical Association Representative</p>"
"<p>CPT Editorial Panel Representative</p>"
"<p>American Osteopathic Association Representative</p>"
"<p>Health Care Professionals Advisory Committee Representative</p>"
"<p>Practice Expense Review Committee Representative</p>"
"<p>Radiology</p><p>Geriatric Medicine</p><p>Infectious Diseases*</p>"
"<p>* Indicates rotating seat</p><p>Current RUC Members</p>"
"<p>RUC Chairperson: A B, MD</p><p>American Medical Association: C D, MD</p>"
"<p>CPT Editorial Panel: E F, MD</p>"
"<p>Health Care Professional Advisory Committee: G H, PhD</p>"
"<p>Practice Expense Subcommittee: I J, MD</p>"
"<p>American College of Radiology: K L, MD</p>"
"<p>American Geriatrics Society: M N, MD</p>"
"<p>American Osteopathic Association: O P, DO</p>"
"<p>Infectious Diseases Society of America: Q R, MD</p>"
"<p>Static</p></main><footer>footer</footer></body></html>"
).encode()
_CPT = (
"<html><body><main><h2>Makeup of CPT Editorial Panel</h2>"
"<p>The Panel is representative of all, with 12 of its 21 members appointed by the "
"national medical specialty societies. In addition to the specialty seats, the Panel "
"includes a seat for the Panel chair, vice chair, two seats for members of the CPT "
"Health Care Professionals Advisory Committee, as well as representatives from these "
"organizations:</p><p>One seat for the American Hospital Association</p>"
"<p>One seat for an at-large organizational member</p>"
"<p>The AMA Board of Trustees selects all members.</p></main></body></html>"
).encode()
@pytest.fixture
def env(self, con, monkeypatch, tmp_path):
import httpx
from bib.store import Store
store = Store(":memory:", storage_dir=tmp_path / "storage")
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(pfs_cli, "_seat_root", lambda: tmp_path / "rosters")
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda task, fam: tmp_path / f"{task}-{fam}.jsonl"
)
con.execute("CREATE SCHEMA IF NOT EXISTS terminology")
con.execute(
"CREATE TABLE terminology.cms_acceptable_provider_specialty_codes "
"(specialty_code VARCHAR, specialty_description VARCHAR)"
)
con.execute(
"INSERT INTO terminology.cms_acceptable_provider_specialty_codes "
"VALUES ('26', 'Geriatric Medicine')"
)
import time
monkeypatch.setattr(time, "sleep", lambda s: None)
state = {
"cdx": None,
"snap": None,
"asked": 0,
"reply": None,
"bad_year": None,
"page": self._RUC,
"no_capture": "2021",
}
def handler(req):
if "cdx" in req.url.path:
if state["cdx"] is not None:
return httpx.Response(state["cdx"])
y = req.url.params["from"]
if y == state["bad_year"]:
return httpx.Response(503)
if y == state["no_capture"]:
return httpx.Response(200, json=[])
return httpx.Response(
200, json=[["urlkey", "timestamp"], ["k", f"{y}0105123456"]]
)
if state["snap"] is not None:
return httpx.Response(
state["snap"], content=b"<html>Access Denied</html>"
)
return httpx.Response(200, content=state["page"])
monkeypatch.setattr(
pfs_cli,
"_http_dossier",
lambda: httpx.Client(transport=httpx.MockTransport(handler)),
)
def ask(system, user, schema):
state["asked"] += 1
return state["reply"] or {
"seats": [
{
"organization": "American Geriatrics Society",
"seat_type": "permanent",
"specialty": "GERIATRIC MEDICINE",
}
]
}
ask.model = "qwen-test"
def asker(large=False):
state["asker_built"] = True
return ask
monkeypatch.setattr(pfs_cli, "_asker", asker)
try:
yield store, state, tmp_path
finally:
store.close()
def _run(self, *extra, body="RUC", url=None):
return runner.invoke(
app,
[
"pfs",
"seats",
"--body",
body,
"--url",
url or self.URL,
"--from",
"2020",
"--to",
"2022",
*extra,
],
)
def test_dry_run_lists_snapshots_and_touches_nothing(self, con, env):
store, state, tmp_path = env
res = self._run("--dry-run")
assert res.exit_code == 0, res.output
assert (
"2020 20200105123456 https://web.archive.org/web/20200105123456id_/"
in res.output
)
assert "2021 2021" not in res.output
assert "gap: RUC 2021 not captured (no archive capture)" in res.output
assert "seats: 2 snapshots; dry run" in res.output
assert not (tmp_path / "rosters").exists()
def test_cdx_refusal_is_a_stated_gap_for_every_year(self, con, env):
store, state, tmp_path = env
state["cdx"] = 403
res = self._run()
assert res.exit_code == 0, res.output
assert "gap: RUC roster not captured (403)" in res.output
for y in (2020, 2021, 2022):
assert f"gap: RUC {y} not captured (403)" in res.output
dry = self._run("--dry-run")
assert "gap: RUC roster not captured (403)" in dry.output
def test_a_cdx_failure_partway_keeps_the_years_already_found(self, con, env):
store, state, tmp_path = env
state["bad_year"] = "2022"
res = self._run("--write")
assert res.exit_code == 0, res.output
assert "gap: RUC roster not captured (503)" in res.output
assert "gap: RUC 2022 not captured (503)" in res.output
assert "2020 20200105123456:" in res.output
assert con.execute(
"SELECT count(DISTINCT year) FROM pfs.governance_seat"
).fetchone() == (1,)
def test_seats_cli_records_gap_on_403(self, con, env):
store, state, tmp_path = env
state["snap"] = 403
res = self._run("--write")
assert res.exit_code == 0, res.output
assert "gap: RUC roster not captured (403)" in res.output
assert "gap: RUC 2020 not captured (403)" in res.output
assert con.execute("SELECT count(*) FROM pfs.governance_seat").fetchone() == (
0,
)
def test_a_bot_page_with_status_200_is_a_gap_and_is_not_saved(self, con, env):
store, state, tmp_path = env
state["snap"] = 200
res = self._run()
assert "gap: RUC roster not captured (bot page)" in res.output
assert not list((tmp_path / "rosters").rglob("*.html"))
def test_full_run_parses_writes_logs_gaps_and_needs_no_model(self, con, env):
store, state, tmp_path = env
res = self._run("--write")
assert res.exit_code == 0, res.output
assert not state.get("asker_built") and state["asked"] == 0
saved = sorted(p.name for p in (tmp_path / "rosters" / "RUC").glob("*.html"))
assert saved == ["2020.20200105123456.html", "2022.20220105123456.html"]
rows = con.execute(
"SELECT organization, specialty, seat_type, holder_kind, n_seats, item_key "
"FROM pfs.governance_seat WHERE year = 2020 ORDER BY organization"
).fetchall()
by = {r[0]: r for r in rows}
assert len(rows) == 9 # 6 special seats + radiology, geriatrics, IDSA
assert by["RUC Chair"][2] == "chair"
assert by["American College of Radiology"][1] == "DIAGNOSTIC RADIOLOGY"
assert by["Infectious Diseases Society of America"][2] == "rotating"
assert by["American Geriatrics Society"][1] == "GERIATRIC MEDICINE"
assert "doc:ruc-roster" in store.get(rows[0][5]).tags
assert "2020 20200105123456: 9 seats" in res.output
assert "note:" not in res.output
assert con.published == [True]
log = con.execute(
"SELECT rows, source_file FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rows = 0"
).fetchall()
assert len(log) == 1 and "2021" in log[0][1]
assert "gap: RUC 2021 not captured (no archive capture)" in res.output
res2 = self._run("--write")
assert res2.exit_code == 0
assert con.execute(
"SELECT count(*) FROM pfs.governance_seat WHERE year = 2020"
).fetchone() == (len(rows),)
def test_rerun_replaces_gap_and_note_rows_and_stores_notes(self, con, env):
store, state, tmp_path = env
state["page"] = self._RUC.replace(b"3 of its 9 members", b"3 of its 12 members")
for _ in range(2):
assert self._run("--write").exit_code == 0
gaps = con.execute(
"SELECT count(*) FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rule_id = 'gap' AND source_file LIKE '%#2021 %'"
).fetchone()
assert gaps == (1,)
notes = con.execute(
"SELECT source_file, rows FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rule_id = 'note' ORDER BY source_file"
).fetchall()
assert (
len(notes) == 2
and "#2020 note: parsed 9 seats, the page says 12" in notes[0][0]
)
assert con.execute(
"SELECT count(*) FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rows > 0 AND rule_id = ''"
).fetchone() == (2,)
def test_limit_caps_snapshots(self, con, env):
store, state, tmp_path = env
res = self._run("--limit", "1")
assert "seats: 9 rows from 1 snapshots" in res.output
assert con.published == [] # no --write
def test_cpt_panel_is_parsed_from_the_makeup_text(self, con, env):
store, state, tmp_path = env
state["page"] = self._CPT
res = self._run("--write", body="CPT_PANEL", url=self.CPT_URL)
assert res.exit_code == 0, res.output
rows = con.execute(
"SELECT organization, seat_type, holder_kind, n_seats FROM pfs.governance_seat "
"WHERE year = 2022 ORDER BY organization"
).fetchall()
assert ("national medical specialty societies", "other", "category", 12) in rows
assert ("American Hospital Association", "other", "organization", 1) in rows
assert not any("American Medical Association" in r[0] for r in rows)
assert "note:" in res.output # 12+1+1+2+1+1 = 18, not the page's 21
def test_an_unrecognised_layout_is_a_per_year_gap(self, con, env):
store, state, tmp_path = env
state["page"] = (
b"<html><main>" + b"Something else entirely. " * 120 + b"</main></html>"
)
res = self._run("--write")
assert res.exit_code == 0, res.output
assert "gap: RUC 2020 not captured (layout not recognised)" in res.output
assert con.execute("SELECT count(*) FROM pfs.governance_seat").fetchone() == (
0,
)
def test_cross_check_logs_model_disagreements(self, con, env):
store, state, tmp_path = env
state["reply"] = {
"seats": [
{
"organization": "American Geriatrics Society",
"seat_type": "permanent",
"specialty": "GERIATRIC MEDICINE",
},
{
"organization": "Society of Thoracic Surgeons",
"seat_type": "permanent",
"specialty": "",
},
]
}
res = self._run("--cross-check", "--limit", "1")
assert res.exit_code == 0, res.output
assert state["asked"] >= 1
assert "cross-check 2020: model-only" in res.output
assert "Society of Thoracic Surgeons" not in res.output.split("cross-check")[0]
n = state["asked"]
self._run("--cross-check", "--limit", "1")
assert state["asked"] == n # resumable ledger
def test_cross_check_agreement(self, con, env, monkeypatch):
from pfs.seats import ParsedSeat
store, state, tmp_path = env
monkeypatch.setattr(
pfs_cli.seat_lib,
"parse_ruc",
lambda text: (
[
ParsedSeat(
"American Geriatrics Society", "", "permanent", "organization"
)
],
[],
),
)
res = self._run("--cross-check", "--limit", "1")
assert "cross-check 2020: model and parse agree" in res.output
def test_a_network_failure_on_a_snapshot_is_a_gap(self, con, env, monkeypatch):
import httpx
def handler(req):
if "cdx" in req.url.path:
y = req.url.params["from"]
return httpx.Response(200, json=[["u", "t"], ["k", f"{y}0105123456"]])
raise httpx.ConnectError("down")
monkeypatch.setattr(
pfs_cli,
"_http_dossier",
lambda: httpx.Client(transport=httpx.MockTransport(handler)),
)
res = self._run()
assert "gap: RUC 2020 not captured (ConnectError)" in res.output
assert "gap: RUC 2021 not captured (not tried after ConnectError)" in res.output
def test_cross_check_failure_is_reported(self, con, env):
store, state, tmp_path = env
state["reply"] = {"bad": 1}
res = self._run("--cross-check", "--limit", "1")
assert "cross-check 2020: model gave no valid reply" in res.output
def test_bad_arguments(self, con, env):
res = runner.invoke(app, ["pfs", "seats", "--body", "XYZ", "--url", self.URL])
assert res.exit_code != 0 and "--body" in res.output
res = runner.invoke(
app,
[
"pfs",
"seats",
"--body",
"RUC",
"--url",
self.URL,
"--from",
"2022",
"--to",
"2020",
],
)
assert res.exit_code != 0 and "--from" in res.output
def test_a_network_failure_is_a_gap_named_by_its_error(self, con, env, monkeypatch):
import httpx
def boom(req):
raise httpx.ConnectError("down")
monkeypatch.setattr(
pfs_cli,
"_http_dossier",
lambda: httpx.Client(transport=httpx.MockTransport(boom)),
)
res = self._run()
assert "gap: RUC roster not captured (ConnectError)" in res.output
def test_seat_root_sits_beside_the_primary_database():
assert str(pfs_cli._seat_root()).endswith("ama/rosters")
Reference in New Issue View Git Blame Copy Permalink