homelab/stack
1
0
Fork 0
Code Issues 210 Pull Requests 2 Actions Packages Projects 1 Releases 1 Wiki Activity
Files
dfaaf438908eeb106a9e4eafbfb1015c3f9c7d90
stack/tests/cli/test_pfs_cli.py
kert dfaaf43890 fix(pfs): stance R66a/R67/R68 — stance_eval JSON, method record survives reaction reruns, family-named relevance window, packaged golden set, load_duration log 
- reaction --write carries stance_method/stance_agreement/stance_eval over from
  the stored docket row (it used to wipe the PRELIMINARY marker)
- pfs.code_reaction.stance_eval: {cell, recall, direction, n_items,
  n_positions, family, docket, in_sample}, stamped only for the golden set's
  family; new writes leave stance_agreement NULL, older rows stay readable
- the golden set moves to src/pfs/golden/stance_home_visits_cy2023.json
- R67: the +/-1-sentence window must name the event's subject keyword AND this
  family (family code, offered link code/alias, pair phrase); valuation needs a
  family code
- JsonExtractor logs Ollama load_duration at DEBUG
- stance-eval prints recall and direction
2026-11-12 16:04:19 -05:00

2575 lines
96 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
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_: [
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_: [
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])
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_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_VERSIONS['multi']}:qwen-x"
# R66a: the measurement is the stance_eval JSON; the single figure is gone
assert d1.stance_agreement is None
assert json.loads(d1.stance_eval) == {
"cell": 0.83,
"recall": 0.556,
"direction": 0.667,
"n_items": 20,
"n_positions": 27,
"family": "HOME-VISITS",
"docket": "CMS-2022-0113",
"in_sample": True,
}
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_eval_is_stamped_only_for_the_golden_family(
self, con, monkeypatch, tmp_path
):
"""A golden set of another family is never replayed nor stamped."""
t = TestStance()
t._patch(con, monkeypatch)
monkeypatch.setattr(pfs_cli, "_golden_run", lambda *a, **k: 1 / 0)
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"]
assert d1.stance_method and d1.stance_eval is None
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 test_eval_is_none_without_golden_or_with_pending(self, monkeypatch, tmp_path):
F = "HOME-VISITS"
assert pfs_cli._method_eval(tmp_path / "none.json", "multi", "m", F) is None
g = tmp_path / "g.json"
g.write_text(json.dumps({"family": F, "docket": "D1", "items": [{}]}))
monkeypatch.setattr(
pfs_cli, "_golden_run", lambda data, method, **kw: (None, {})
)
assert pfs_cli._method_eval(g, "multi", "m", F) is None
monkeypatch.setattr(
pfs_cli, "_golden_run", lambda data, method, **kw: (SCORES, {})
)
got = json.loads(pfs_cli._method_eval(g, "multi", "m", F))
assert (got["recall"], got["direction"], got["n_items"]) == (0.556, 0.667, 1)
assert pfs_cli._method_eval(g, "multi", "m", "CCM") is None
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
Reference in New Issue View Git Blame Copy Permalink