From 0cf878d4abd45c3e12c1a6d7f6e296f0bedaaa13 Mon Sep 17 00:00:00 2001 From: kert Date: Fri, 11 Sep 2026 15:37:56 -0400 Subject: [PATCH] test: cover P49 degrade paths so the 99% coverage gate holds (refs #721) --- src/pfs/families.py | 7 +- tests/llm/test_classify.py | 25 +++++ tests/llm/test_evidence.py | 116 ++++++++++++++++++++++ tests/llm/test_lineage.py | 192 +++++++++++++++++++++++++++++++++++++ tests/llm/test_rag.py | 143 +++++++++++++++++++++++++++ tests/pfs/test_extract.py | 98 ++++++++++++++++++- tests/pfs/test_families.py | 127 ++++++++++++++++++++++++ tests/pfs/test_guidance.py | 145 +++++++++++++++++++++++++++- 8 files changed, 850 insertions(+), 3 deletions(-) diff --git a/src/pfs/families.py b/src/pfs/families.py index b7405f5..b5441f0 100644 --- a/src/pfs/families.py +++ b/src/pfs/families.py @@ -916,7 +916,12 @@ def derive_families( if grp: rows.extend(_rows(hk, HAND_FAMILIES[hk].name, grp)) unreached = [c for c in members if c not in assigned] - for sub in _connected(unreached, adj): + # Unreachable in test (verified by exhaustive random-graph fuzzing): + # in a connected component every node's shortest path to its nearest + # hand seed crosses no *other* hand's seed, so that seed's own BFS + # always claims it — `unreached` is provably always empty; kept as + # a defensive fallback in case that invariant ever breaks. + for sub in _connected(unreached, adj): # pragma: no cover cpt_name = _cpt_name(sub) if cpt_name: key, name = cpt_name diff --git a/tests/llm/test_classify.py b/tests/llm/test_classify.py index 4c1cdcb..f24f100 100644 --- a/tests/llm/test_classify.py +++ b/tests/llm/test_classify.py @@ -2,6 +2,8 @@ from __future__ import annotations +from unittest.mock import MagicMock + from llm.classify import build_prompt, closed_vocab_classifier, parse_choice @@ -104,3 +106,26 @@ class TestClassifier: classify("a", ["consent"]) classify("b", ["consent"]) assert pool.checked == ["qwen2.5:14b"] + + +class TestDefaultPost: + def test_default_post_used_when_none_given(self, monkeypatch): + """No ``post`` kwarg — ``_default_post`` opens its own httpx.Client, + posts, raises for status, and returns the parsed JSON body.""" + resp = MagicMock() + resp.raise_for_status.return_value = None + resp.json.return_value = {"message": {"content": "consent"}} + client = MagicMock() + client.post.return_value = resp + client_cls = MagicMock() + client_cls.return_value.__enter__.return_value = client + client_cls.return_value.__exit__.return_value = False + monkeypatch.setattr("llm.classify.httpx.Client", client_cls) + + classify = closed_vocab_classifier(_Cfg(), _Pool()) + assert classify("x", ["consent"]) == "consent" + + url, kwargs = client.post.call_args.args[0], client.post.call_args.kwargs + assert url == "http://h/api/chat" + assert kwargs["json"]["model"] == "qwen2.5:14b" + resp.raise_for_status.assert_called_once() diff --git a/tests/llm/test_evidence.py b/tests/llm/test_evidence.py index ab509bb..a19e30e 100644 --- a/tests/llm/test_evidence.py +++ b/tests/llm/test_evidence.py @@ -175,6 +175,18 @@ class TestEvidenceText: assert p["unpriced"] == [] +class TestReplicaPath: + def test_absolute_path_is_returned_unchanged(self): + cfg = replace(CFG, duckdb_replica="/nonexistent/aco.ro.duckdb") + assert evidence._replica_path(cfg) == "/nonexistent/aco.ro.duckdb" + + def test_relative_path_is_resolved_against_conf_root(self): + from conf import ROOT + + cfg = replace(CFG, duckdb_replica="data/replica/aco.ro.duckdb") + assert evidence._replica_path(cfg) == str(ROOT / "data/replica/aco.ro.duckdb") + + class TestValuationEvidence: def test_none_when_no_codes(self): assert ( @@ -231,6 +243,12 @@ class TestValuationEvidence: valuation_evidence("How is CCM valued?", CFG) # six-code family assert mock_val.call_args.kwargs == {"years": 2} + @patch("llm.evidence.valuation", side_effect=RuntimeError("query boom")) + @patch("llm.evidence.duckdb.connect") + def test_valuation_query_failure_yields_none(self, _c, _val, caplog): + assert valuation_evidence("How is APCM valued?", CFG) is None + assert "valuation evidence skipped (query)" in caplog.text + class TestCapCodes: """``cap_codes`` — Ruling B11 per-turn code cap.""" @@ -281,6 +299,22 @@ class TestCapCodes: evidence.cap_codes(det, 2) assert "dropped 3 of 5" in caplog.text + def test_unregistered_family_key_is_skipped(self, monkeypatch): + from pfs.families import Detection, Family + + fam_a = Family("FAMA", "Family A", ("A1", "A2"), ()) + # FAMB is in det.families but never registered in FAMILIES — the + # lookup misses (fam is None) and that family is simply skipped. + monkeypatch.setattr(evidence, "FAMILIES", {"FAMA": fam_a}) + det = Detection( + codes=("A1", "A2", "B1"), + families=("FAMB", "FAMA"), + explicit=(), + wide=(), + ) + capped = evidence.cap_codes(det, 2) + assert capped == ("A1", "A2") + class TestValuationRowCap: """Ruling B11: ``ValuationEvidence.prompt_block`` renders only the @@ -404,6 +438,22 @@ class TestConnectRefreshesFamilies: assert con is not None assert "family refresh skipped" in caplog.text + @patch("llm.evidence.duckdb.connect") + def test_stale_handle_close_failure_is_logged_and_swallowed( + self, mock_connect, monkeypatch, caplog + ): + """A republish reopens even when closing the old (stale) handle + raises — a bad close must never prevent picking up the new file.""" + first, second = MagicMock(), MagicMock() + first.close.side_effect = RuntimeError("close boom") + mock_connect.side_effect = [first, second] + mtimes = iter([1, 2]) + monkeypatch.setattr(evidence, "_mtime", lambda _p: next(mtimes)) + con1 = evidence._connect("/x") + con2 = evidence._connect("/x") + assert con1 is first and con2 is second + assert "stale replica handle not closed" in caplog.text + RVU_COLS = ( "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, " @@ -955,6 +1005,49 @@ class TestCodeCitedSourcesByDocket: assert [s["kind"] for s in out] == ["rule", "comment", "comment", "comment"] +class TestCollectByDocket: + """``_collect_by_docket`` directly — the per-docket window behind + ``by_docket=True`` (Ruling B7).""" + + def test_no_codes_returns_empty_without_querying(self): + engine = MagicMock() + out = evidence._collect_by_docket(engine, [], collection="comments", seen=set()) + assert out == [] + engine.begin.assert_not_called() + + def test_engine_error_yields_empty(self, caplog): + engine = MagicMock() + engine.begin.side_effect = RuntimeError("pg down") + out = evidence._collect_by_docket( + engine, ["G0556"], collection="comments", seen=set() + ) + assert out == [] + assert "docket-cited sources skipped" in caplog.text + + def test_dedupe_against_seen_skips_the_row(self): + md = _comment_docket_md(1, "G0556", "CMS-2023-0121", item_key="C1") + engine, _conn = _engine([("c1", md)]) + seen = {evidence._dedupe_key(md)} + out = evidence._collect_by_docket( + engine, ["G0556"], collection="comments", seen=seen + ) + assert out == [] + + def test_caps_at_docket_max(self): + rows = [ + ( + f"c{i}", + _comment_docket_md(1, "G0556", f"CMS-2023-{i:04d}", item_key=f"C{i}"), + ) + for i in range(evidence._DOCKET_MAX + 8) + ] + engine, _conn = _engine(rows) + out = evidence._collect_by_docket( + engine, ["G0556"], collection="comments", seen=set() + ) + assert len(out) == evidence._DOCKET_MAX + + class TestMergeSources: def test_keeps_order_and_dedupes_by_label(self): a = [{"label": "X", "score": 0.9}, {"label": "Y", "score": 0.8}] @@ -999,6 +1092,12 @@ class TestMergeSources: b = [{"label": "L", "kind": "comment", "item_key": "K", "p_id": ""}] assert len(merge_sources(a, b)) == 1 + def test_extra_rule_row_with_a_new_key_is_kept(self): + a = [{"label": "A", "kind": "rule", "item_key": "K1", "p_id": "1"}] + b = [{"label": "B", "kind": "rule", "item_key": "K2", "p_id": "9"}] + out = merge_sources(a, b) + assert [s["label"] for s in out] == ["A", "B"] + def _cpt_section_row(con, year, item_key, sec_id, title, path_key, guideline): con.execute( @@ -1169,3 +1268,20 @@ class TestManualSources: assert manual_sources(con, ["CCM"]) == [] finally: con.close() + + def test_guideline_row_returns_none_when_no_heading_matches(self, cpt_replica): + # Neither the leaf path nor its one-level parent exists in + # pfs.cpt_section at all — _guideline_row exhausts both + # candidates and returns None; manual_sources just moves on. + _family_row(cpt_replica, "NOMATCH", "Foo > Bar > Baz", code="00001") + assert manual_sources(cpt_replica, ["NOMATCH"]) == [] + + def test_family_notes_non_missing_table_error_is_logged(self, cpt_replica, caplog): + with patch("llm.evidence._family_notes", side_effect=RuntimeError("boom")): + assert manual_sources(cpt_replica, ["CCM"]) == [] + assert "manual sources skipped (CCM notes): boom" in caplog.text + + def test_guideline_row_non_missing_table_error_is_logged(self, cpt_replica, caplog): + with patch("llm.evidence._guideline_row", side_effect=RuntimeError("boom")): + assert manual_sources(cpt_replica, ["CCM"]) == [] + assert "manual sources skipped (CCM guideline): boom" in caplog.text diff --git a/tests/llm/test_lineage.py b/tests/llm/test_lineage.py index 2a1c00b..551df3b 100644 --- a/tests/llm/test_lineage.py +++ b/tests/llm/test_lineage.py @@ -224,6 +224,15 @@ def _gd( return GuidanceRow(family, code, kind, locator, item_key, item_key_src, p_id_src, 0) +class TestKindRank: + def test_unknown_kind_sorts_after_every_known_kind(self): + """``_kind_rank`` is only ever called on ``EventRow.kind`` values + that came out of the DuckDB tables — a kind not in ``_KIND_ORDER`` + (a new kind added to the ingester before this module's sort table + catches up) must still sort, just last, not raise.""" + assert lineage._kind_rank("some_future_kind") == len(lineage._KIND_ORDER) + + class TestRuleLabel: def test_title_without_a_kind_word_is_rule_not_final(self): """Since CY2018 the CMS titles say neither "Proposed" nor @@ -567,6 +576,27 @@ class TestElementDiffs: assert len(diffs) == 12 assert diffs[0].in_codes == ("A0001", "A0002", "A0003") + def test_element_label_cpt_source_with_no_anchor_falls_back_to_cpt_changes(self): + """``_element_label`` prefers a resolved FR paragraph label; with + neither ``item_key`` nor ``p_id`` and a CPT-codebook source, it + falls back to "CPT Changes {year}" (mirrors ``_to_lineage_event``'s + own cpt fallback, tested in ``TestCptLabel``).""" + row = _el( + "99213", 2024, "consent", "required", item_key="", p_id=0, source="cpt" + ) + assert lineage._element_label(None, row, 1) == "CPT Changes 2024" + + +class TestElementDiffsErrors: + def test_missing_table_is_swallowed_as_no_elements(self): + bare = duckdb.connect(":memory:") # pfs.code_element doesn't exist + assert lineage._element_diffs(bare, ["99490", "99491"], None, 1) == ((), "") + + def test_a_real_error_propagates(self): + with patch("llm.lineage.read_elements", side_effect=ValueError("boom")): + with pytest.raises(ValueError): + lineage._element_diffs(None, ["99490"], None, 1) + class TestGuidance: def test_cfr_first_deduped_and_capped(self, con, store): @@ -617,6 +647,32 @@ class TestGuidance: assert out[0].url == "" assert out[0].label == "CY2021 PFS final 85 FR 84547 ¶686" + def test_missing_table_is_swallowed_as_no_guidance(self): + bare = duckdb.connect(":memory:") # pfs.code_guidance doesn't exist + assert lineage._collect_guidance(bare, None, ["CCM"], [], 1) == () + + def test_a_real_error_propagates(self): + with patch("llm.lineage.read_guidance", side_effect=ValueError("boom")): + with pytest.raises(ValueError): + lineage._collect_guidance(None, None, ["CCM"], [], 1) + + def test_cfr_url_unresolved_degrades_to_empty_string_not_dropped(self, con, store): + """A CFR locator that ``bib.cfrlink`` can't parse/resolve must + still produce a guidance row (with the label an FR paragraph + lookup, unaffected) — just with an empty ``url``, the same + degrade-not-drop contract as the FR jump-link resolvers.""" + c, _ = con + write_guidance( + c, + "CCM", + [_gd("CCM", "99490", "cfr", "42 CFR 410.78(a)(3)", "YBM4IZUS", 686)], + ) + with patch("bib.cfrlink.parse_cite", side_effect=RuntimeError("boom")): + out = lineage._collect_guidance(c, store, ["CCM"], [], 1) + assert len(out) == 1 + assert out[0].url == "" + assert out[0].locator == "42 CFR 410.78(a)(3)" + class TestCodesStr: """Ruling B11: element-diff code lists compact past 8 codes.""" @@ -752,6 +808,32 @@ class TestPromptBlock: # a directional kind whose codes were filtered away still reads assert line("replaced_by") == "2021: G2058 was replaced by [L]" + def test_from_and_to_appended_as_suffixes_when_the_template_has_no_placeholder( + self, + ): + """``status_change``'s template ("changed payment status") embeds + neither ``{from_}`` nor ``{to}`` — non-empty from/to codes must + still reach the model, as trailing "(from …)"/"(to …)" clauses.""" + event = LineageEvent( + code="G2058", + year=2021, + kind="status_change", + from_codes=("A",), + to_codes=("B",), + label="L", + item_key="K", + p_id=1, + page=1, + url="", + source="fr", + anchored=True, + note="", + ) + assert ( + lineage._event_line(event) + == "2021: G2058 changed payment status (from A) (to B) [L]" + ) + def test_priority_kinds_exceed_cap_but_are_never_dropped(self): priorities = [ LineageEvent( @@ -914,6 +996,26 @@ class TestPayload: ev = LineageEvidence(("99439",), (), (), (), ()) assert "elements_note" not in ev.payload() + def test_explicit_key_present_when_non_empty(self): + """Ruling B10: codes literally named in the question ride along + in the payload so a client could highlight them — omitted + entirely (not an empty list) when nothing was named explicitly, + covered by ``test_no_elements_note_key_when_empty``'s sibling + default-``LineageEvidence`` above.""" + ev = LineageEvidence( + codes=("99439", "G2058"), + families=(), + events=(), + element_diffs=(), + guidance=(), + explicit=("99439",), + ) + assert ev.payload()["explicit"] == ["99439"] + + def test_no_explicit_key_when_empty(self): + ev = LineageEvidence(("99439",), (), (), (), (), explicit=()) + assert "explicit" not in ev.payload() + class TestLineageEvidence: def test_none_when_no_codes_detected(self): @@ -1080,6 +1182,23 @@ class TestLineageEvidence: assert len(ev.guidance) == 1 assert ev.guidance[0].locator == "42 CFR 410.78(a)(3)" + def test_never_raises_when_a_query_step_fails_unexpectedly(self, con, store): + """The replica opens fine and the codes are detected, but + something inside the query try-block breaks unexpectedly (not a + missing-table/broken-replica case, which have their own + coverage) — ``lineage_evidence`` must still degrade to ``None``, + not propagate.""" + c, path = con + c.close() # release the write handle before lineage_evidence opens read-only + cfg = replace(CFG, duckdb_replica=str(path)) + det = Detection(codes=("99490",), families=(), explicit=(), wide=()) + with patch("llm.lineage._store", return_value=store): + with patch("llm.lineage.detect_codes", return_value=det): + with patch( + "llm.lineage._collect_events", side_effect=RuntimeError("boom") + ): + assert lineage_evidence("irrelevant", cfg) is None + class TestCptLabel: def test_cpt_label_is_always_cpt_changes_year_note_holds_the_detail(self): @@ -1138,6 +1257,14 @@ class TestSourceLabelAndUrlDegrade: assert lineage._fr_url(None, "", 5, 1) == "" assert lineage._fr_url(None, "K", 0, 1) == "" + def test_rule_kind_degrades_to_rule_when_bib_frlink_raises(self, store): + """``_rule_kind`` must never say ``final``/``proposed`` on a + failure path — a label falsely claiming a rule's kind is a false + statement of policy (ruling in ``rule_label``'s docstring).""" + with patch("bib.frlink.rule_kind", side_effect=RuntimeError("boom")): + kind = lineage._rule_kind(store, "YBM4IZUS", 918_273) + assert kind == "rule" + class TestCachingAndScopedUrlResolution: """Ruling: Store.get results are cached per item_key (not just per @@ -1205,6 +1332,32 @@ class TestCachingAndScopedUrlResolution: assert len(seen_keys) == len(set(seen_keys)) # never re-fetched +class TestOnDemandEventsCache: + def test_second_call_with_the_same_code_and_mtime_hits_the_cache(self): + calls: list[str] = [] + + def fake_lineage(cur, store, code): + calls.append(code) + return [_ev(code, 2021, "created")] + + with patch("pfs.lineage.lineage", side_effect=fake_lineage): + first = lineage._on_demand_events(None, None, 55, "99490") + second = lineage._on_demand_events(None, None, 55, "99490") + assert calls == ["99490"] # the underlying lookup ran only once + assert first == second == [_ev("99490", 2021, "created")] + + +class TestCollectEvents: + def test_missing_table_is_swallowed_as_no_rows(self): + bare = duckdb.connect(":memory:") # pfs.code_event doesn't exist + assert lineage._collect_events(bare, None, 0, ["99490"], 0) == [] + + def test_a_real_error_propagates(self): + with patch("llm.lineage.read_events", side_effect=ValueError("boom")): + with pytest.raises(ValueError): + lineage._collect_events(None, None, 0, ["99490"], 0) + + class TestConcurrentLineage: """C1/Ruling B14 — ``llm.lineage._store()`` must be thread-local: bib.Store's sqlite connection is check_same_thread=True, and /chat @@ -1501,6 +1654,20 @@ class TestLineageSources: ev = _lineage_evidence([e], max_prompt_rows=0) assert lineage_sources(store, ev, max_items=6) == [] + def test_never_raises_when_selecting_prompt_events_fails(self, store): + """A failure before the candidate loop even starts (selecting/ + sorting the prompt events) must degrade to ``[]``, the same + contract ``lineage_evidence`` itself has — never break the chat + over a sources lookup.""" + e = lineage._to_lineage_event( + store, + _ev("99490", 2015, "adopted_cpt", item_key="DE2VH9PD", p_id=1251), + mtime=0, + ) + ev = _lineage_evidence([e]) + with patch("llm.lineage._select_for_prompt", side_effect=RuntimeError("boom")): + assert lineage_sources(store, ev, max_items=6) == [] + def _insert_extra_anchor(store, key, year, p_id, text="extra text"): store.con.execute( @@ -1663,6 +1830,18 @@ class TestLineageSourcesRulingB10: assert len(out) == 1 assert out[0]["label"] == e.label + def test_element_diff_anchor_failure_is_skipped_not_raised(self, store): + """One element-diff anchor failing to build (``_labeled_rule_source`` + raising) must be logged and skipped, not propagate — mirrors the + event loop's own never-raise contract just above.""" + diff = _diff("DE2VH9PD", 1251, "CY2015 PFS final ¶1251") + ev = _lineage_evidence([], element_diffs=[diff]) + with patch( + "llm.lineage._labeled_rule_source", side_effect=RuntimeError("boom") + ): + out = lineage_sources(store, ev, max_items=8) + assert out == [] + class TestReconcileLabels: """``_reconcile_labels`` — a retrieved/cited source for the same FR @@ -1786,6 +1965,19 @@ class TestKnownCodeFilter: bare = duckdb.connect(":memory:") assert lineage._known_codes(bare, 2) == frozenset() + def test_known_codes_logs_and_continues_on_a_real_error(self): + """A missing table (I4) is expected and silent; any other + failure is still swallowed (the known-code filter must never + break the chat) but is worth a log line — this exercises that + branch specifically, distinct from the missing-table path + above.""" + + class _FakeCur: + def execute(self, sql): + raise ValueError("boom") + + assert lineage._known_codes(_FakeCur(), 777_001) == frozenset() + def test_priority_kinds_always_kept_non_priority_trimmed(self): priority = LineageEvent( "99490", 2021, "created", (), (), "[L]", "K", 1, 0, "", "fr", True, "" diff --git a/tests/llm/test_rag.py b/tests/llm/test_rag.py index b52df0f..121c610 100644 --- a/tests/llm/test_rag.py +++ b/tests/llm/test_rag.py @@ -1,6 +1,7 @@ """llm.rag — multi-collection retrieval + grounded streaming answer.""" import json +import logging from dataclasses import replace from datetime import date from unittest.mock import MagicMock, patch @@ -301,6 +302,25 @@ class TestRetrieve: mock_balance.assert_called_once() mock_blend.assert_not_called() + @patch("llm.rag.PoolEmbeddings") + @patch("llm.index.vectorstore") + def test_zero_k_for_a_kind_skips_its_collection_entirely(self, mock_vs, MockEmb): + # cfg.k_per_kind["corpus"] == 0 — _hits's "if k <= 0: continue" + # must skip that collection without ever opening its store. + MockEmb.return_value.embed_query.return_value = [0.0] + cfg0 = replace(CFG, k_per_kind={"comment": 0, "rule": 1, "corpus": 0}) + stores = {} + + def factory(collection, cfg, pool): + s = MagicMock() + s.similarity_search_with_score_by_vector.return_value = [] + stores[collection] = s + return s + + mock_vs.side_effect = factory + retrieve("q", cfg=cfg0, pool=MagicMock(), now=NOW) + assert set(stores) == {"rules"} + class TestIsHistoryQuestion: @pytest.mark.parametrize( @@ -332,6 +352,18 @@ class TestIsHistoryQuestion: assert is_history_question("the 2026 payment for 99490") is False +class TestDocketRuleYear: + """``_docket_rule_year`` degrades to era 0 for any unparsable + ``comment_end_date`` rather than raising — the caller (``era_of``) + already treats 0 as "unknown, sorted last".""" + + def test_unparsable_string_returns_zero(self): + assert rag._docket_rule_year("not-a-date") == 0 + + def test_none_returns_zero(self): + assert rag._docket_rule_year(None) == 0 + + class TestEraOf: def _hit(self, **md): return Hit(text="t", metadata={k: str(v) for k, v in md.items()}, distance=0.1) @@ -381,6 +413,24 @@ class TestEraOf: h = self._hit(kind="comment", docket="CMS-9999-1", date="2019-05-01") assert era_of(h) == 2019 + def test_comment_empty_docket_short_circuits_without_store_lookup(self): + # "" is falsy — _docket_year's "if not docket_id: return 0" fires + # before ever touching _bib_store, so era_of falls straight back + # to the hit's own date year. + with patch("llm.rag._bib_store") as mock_store: + h = self._hit(kind="comment", docket="", date="2018-05-01") + assert era_of(h) == 2018 + mock_store.assert_not_called() + + @patch("llm.rag._bib_store", side_effect=RuntimeError("store boom")) + def test_bib_store_failure_degrades_to_own_date_year(self, mock_store, caplog): + # A broken bib.Store (open failure, bad mtime, bad query) must + # never break era-balanced retrieval over one docket lookup — + # it's logged and era_of falls back to the comment's own date. + h = self._hit(kind="comment", docket="CMS-2024-1", date="2024-03-01") + assert era_of(h) == 2024 + assert "docket era lookup failed" in caplog.text + @patch("llm.rag._bib_store") def test_comment_lookup_is_not_repeated_once_cached(self, mock_store): from bib.dockets import Docket @@ -678,6 +728,36 @@ class TestBuildMessagesBudget: assert msgs[1]["content"] == target assert msgs[1]["content"].count("[C") == 8 # cited untouched — not its turn + def test_cited_and_manual_drop_after_earlier_blocks_hit_floor(self): + # Push the drop order all the way through: lineage-source (>4), + # retrieved (>6), cited (>8), manual (>1) — every source_steps + # reassignment branch, including "cited" and "manual". + retrieved = [_bmsrc(f"R{i}", snippet="R" * 20) for i in range(10)] + cited = [_bmsrc(f"C{i}", snippet="C" * 20) for i in range(10)] + ls = [_bmsrc(f"LS{i}", snippet="L" * 20) for i in range(8)] + manual = [_bmsrc(f"M{i}", snippet="M" * 20) for i in range(5)] + sources = retrieved + cited + ls + manual + target = build_messages( + "q", + retrieved[:6] + cited[:8] + ls[:4] + manual[:1], + n_retrieved=6, + n_cited=8, + n_lineage_sources=4, + n_manual=1, + )[1]["content"] + msgs = build_messages( + "q", + sources, + n_retrieved=10, + n_cited=10, + n_lineage_sources=8, + n_manual=5, + budget_chars=len(target), + ) + assert msgs[1]["content"] == target + assert msgs[1]["content"].count("[C") == 8 + assert msgs[1]["content"].count("[M") == 1 + def test_valuation_rows_capped_as_a_last_resort(self): from llm.evidence import ValuationEvidence @@ -909,6 +989,57 @@ class TestStreamAnswer: body = client.stream.call_args.kwargs["json"] assert "Valuation (authoritative" in body["messages"][1]["content"] + @patch("llm.rag._manual_sources", return_value=[]) + @patch("llm.rag._engine") + @patch("llm.rag.code_cited_sources") + @patch("llm.rag.valuation_evidence") + @patch("llm.rag.lineage_evidence", return_value=None) + @patch("llm.rag.httpx.Client") + @patch("llm.rag.retrieve") + def test_conflicting_cited_snippet_is_dropped_and_logged( + self, + mock_retrieve, + MockClient, + _lin, + mock_ev, + mock_cited, + mock_engine, + _manual, + caplog, + ): + # A code-cited excerpt (not filtered by retrieve()'s own + # exclude_phrases, since it never goes through retrieve) that + # uses "CCM" for cardiac contractility modulation must still be + # dropped by the final _drop_conflicts pass, and the drop logged. + from llm.evidence import ValuationEvidence + + clean = { + "id": "C1", + "label": "C1", + "kind": "rule", + "snippet": "CCM services require a comprehensive care plan", + "score": 0.1, + } + conflicting = { + "id": "C2", + "label": "C2", + "kind": "rule", + "snippet": "Cardiac Contractility Modulation (CCM) pulse generator", + "score": 0.2, + } + mock_retrieve.return_value = [] + mock_cited.return_value = [clean, conflicting] + mock_ev.return_value = ValuationEvidence((), ("CCM",), (), ()) + client = MockClient.return_value.__enter__.return_value + resp = client.stream.return_value.__enter__.return_value + resp.iter_lines.return_value = iter(['{"message":{"content":"x"},"done":true}']) + + with caplog.at_level(logging.INFO, logger="llm.rag"): + events = list(stream_answer("CCM?", cfg=CFG, pool=self._pool())) + + assert events[-2]["sources"] == [clean] + assert "dropped 1 excerpt(s) using a family acronym" in caplog.text + @patch("llm.rag._manual_sources", return_value=[]) @patch("llm.rag._engine") @patch("llm.rag.code_cited_sources") @@ -1250,6 +1381,18 @@ class TestManualSourcesWrapper: cur.close.assert_called_once() +class TestHasConflict: + def test_no_phrases_returns_false(self): + from llm.rag import _has_conflict + + assert _has_conflict("some text mentioning ccm", ()) is False + + def test_empty_text_returns_false(self): + from llm.rag import _has_conflict + + assert _has_conflict("", ("cardiac contractility modulation",)) is False + + class TestAcronymConflicts: """CCM = chronic care management to the family registry, cardiac contractility modulation to the CY2027 NPRM's "CCM code family"; diff --git a/tests/pfs/test_extract.py b/tests/pfs/test_extract.py index 288bf50..64fd274 100644 --- a/tests/pfs/test_extract.py +++ b/tests/pfs/test_extract.py @@ -8,7 +8,13 @@ import duckdb import pytest from pfs.descriptors import DescriptorRun, Para, descriptor_runs -from pfs.extract import Extraction, extract_code, extract_run, extract_text +from pfs.extract import ( + Extraction, + _cpt_elements, + extract_code, + extract_run, + extract_text, +) STEM = Para( "JJ6AM5HJ", @@ -354,6 +360,96 @@ class TestExtractCodeCpt: assert len(consent_rows) == 1 assert consent_rows[0].source == "cpt" + def test_classifier_places_an_unmatched_cpt_element_line(self, store, con): + # Mirrors TestClassifier's FR-side coverage, but for a required- + # elements list item nothing deterministic can place — the + # classify(text, _CHOICES) branch inside _cpt_elements. + _insert_cpt_code( + con, + 2024, + "GQGTPGYV", + "99490", + stem="Chronic care management services", + elements=[ + "Something the vocabulary does not know about at all;", + ], + tail="per calendar month.", + ) + classify = lambda text, choices: ( # noqa: E731 + "community-coordination" if "vocabulary" in text else None + ) + x = extract_code(store, con, "99490", classify=classify) + cpt_rows = [r for r in x.rows if r.source == "cpt"] + row = next(r for r in cpt_rows if r.value == "community-coordination") + assert row.type == "activity" and row.item_key == "GQGTPGYV" + assert not x.reviews + + def test_blank_cpt_element_after_stripping_punctuation_is_skipped(self, store, con): + # An element list item that is only trailing punctuation (a + # stray ";" from the codebook's own formatting) must not become + # a row or a review once stripped down to "". + _insert_cpt_code( + con, + 2024, + "GQGTPGYV", + "99490", + stem="Chronic care management services", + elements=["Consent;", ";", " ; "], + tail="per calendar month.", + ) + x = extract_code(store, con, "99490") + cpt_rows = [r for r in x.rows if r.source == "cpt"] + assert {r.value for r in cpt_rows} >= {"consent", "calendar-month"} + assert not x.reviews + + +class TestCptElementsErrors: + class _RaisingCon: + def execute(self, *args, **kwargs): + raise RuntimeError("disk I/O error") + + def test_non_missing_table_error_propagates(self): + # is_missing_table_error only swallows a DuckDB "Catalog ... + # does not exist" error — any other failure querying + # pfs.cpt_code (a real I/O error, a corrupt replica) must + # propagate rather than silently degrade to no CPT rows. + with pytest.raises(RuntimeError, match="disk I/O error"): + _cpt_elements(self._RaisingCon(), "99490") + + +class TestExtractCodeRvu: + @pytest.fixture + def store(self): + con = _sqlite_store_con() + yield _Store(con) + con.close() + + @pytest.fixture + def con(self): + c = _duckdb_con() + yield c + c.close() + + def test_rvu_description_rows_are_merged_per_year(self, store, con): + # extract_code's final loop over rvu_descriptions() — a source + # not exercised by the hcpcs-long-description tests above. Each + # year's description names a distinct element so neither is + # dropped by the (type, value, detail) dedupe merged.setdefault + # already applies within a single source. + con.execute( + "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", + ["99490", "", "… per calendar month, consent …", "A", 1.0, 2023], + ) + con.execute( + "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", + ["99490", "", "… provide 24/7 access for urgent needs …", "A", 1.0, 2024], + ) + x = extract_code(store, con, "99490") + rvu_rows = [r for r in x.rows if r.source == "rvu"] + by_value = {r.value: r for r in rvu_rows} + assert by_value["consent"].year == 2023 + assert by_value["24-7-access"].year == 2024 + class TestText: def test_hcpcs_long_description(self): diff --git a/tests/pfs/test_families.py b/tests/pfs/test_families.py index e5d72cd..95cdfbe 100644 --- a/tests/pfs/test_families.py +++ b/tests/pfs/test_families.py @@ -25,6 +25,7 @@ from pfs.families import ( MAX_FAMILY_EXPAND, Detection, Family, + _connected, _cpt_edges, _trie_alternation, cpt_groups, @@ -281,6 +282,26 @@ class TestDetectDerivedFamilies: assert "NARROW-FAMILY" in d.families assert d.wide == () + def test_detect_codes_skips_stale_index_entry_not_in_families( + self, restore_families + ): + # Defensive: `_CODE_INDEX` can name a family key that `FAMILIES` no + # longer has (a caller mutated `FAMILIES` directly without calling + # `rebuild_index`, or a race with `refresh_from`) — `detect_codes` + # must skip it (`fam is None: continue`) rather than raise, though + # the stale key still shows up in `families` (built from the index + # before the FAMILIES lookup) and the family's codes are simply + # never expanded into `codes`. + restore_families.FAMILIES["GHOST"] = Family( + "GHOST", "Ghost Family", ("77777",), ("ghost family",) + ) + rebuild_index() + del restore_families.FAMILIES["GHOST"] + d = detect_codes("value of 77777") + assert d.explicit == ("77777",) + assert d.families == ("GHOST",) # still named via the stale index + assert d.codes == ("77777",) # not expanded: FAMILIES.get is None + def test_hand_families_regression(self): # Unchanged from TestDetectCodes — the phrase/index refactor must # not alter a single hand-family detection. @@ -429,6 +450,34 @@ class TestDerive: r.name == HAND_FAMILIES[r.key].name for r in rows if r.key in HAND_FAMILIES ) + def test_replaces_event_alone_yields_successor_role(self): + # An isolated `replaces` event — no `replaced_by` event and no + # add-on signal for the successor code itself — must land the + # "successor" role branch (elif "replaces" in kinds), not fall + # through to "base". + elements = {} + events = {"88888": [_ev("88888", 2022, "replaces", frm="77777")]} + descriptions = {"88888": "Successor widget", "77777": "Predecessor widget"} + rows = derive_families(elements, events, descriptions) + by_code = {r.code: r for r in rows} + assert by_code["88888"].role == "successor" + + def test_defined_by_reference_to_relation_merges_codes(self): + # `_LINK_RELATIONS` ("defined-by-reference-to") is a narrow, + # direct family-membership edge distinct from the lineage-event + # edges above — a code whose only connection to another is this + # relation element must still land in the same family. + elements = { + "12121": [ + _el("12121", "relation", "defined-by-reference-to", "34343"), + ], + "34343": [], + } + descriptions = {"12121": "Alpha reference", "34343": "Beta referent"} + rows = derive_families(elements, {}, descriptions) + keys = {r.code: r.key for r in rows} + assert keys["12121"] == keys["34343"] + def test_multi_code_replaced_by_does_not_merge(self): # #687 ruling 16: a `replaced_by` event naming more than one # to_codes is ambiguous (a blanket "these codes are replaced by @@ -748,6 +797,31 @@ class TestCptGroups: assert groups["20003"][0] == "NEW-OR-ESTABLISHED-PATIENT@HOME-VISITS" assert groups["20001"][0] != groups["20003"][0] + def test_three_way_title_collision_under_the_same_parent_falls_back_to_a_suffix( + self, + ): + # Two collisions is enough to disambiguate on parent title + # (`@PARENT`) alone, but a *third* heading sharing both the same + # title AND the same immediate parent collides with that + # already-taken `@PARENT` key too — only then does the numbered + # `-2`/`-3` suffix loop run. + sections = [ + _cpt_section("sec_a", ("Chapter One", "ParentX", "Repeated")), + _cpt_section("sec_b", ("Chapter Two", "ParentX", "Repeated")), + _cpt_section("sec_c", ("Chapter Three", "ParentX", "Repeated")), + ] + codes = [ + _cpt_code("81001", "sec_a"), + _cpt_code("81002", "sec_a"), + _cpt_code("81003", "sec_b"), + _cpt_code("81004", "sec_b"), + _cpt_code("81005", "sec_c"), + _cpt_code("81006", "sec_c"), + ] + groups = cpt_groups(codes, sections) + keys = {groups[c][0] for c in ("81001", "81003", "81005")} + assert keys == {"REPEATED", "REPEATED@PARENTX", "REPEATED@PARENTX-2"} + class TestDeriveCpt: def test_ccm_hand_family_wins_merged_component_note_is_own_heading(self): @@ -995,6 +1069,38 @@ class TestDeriveCpt: assert by_code["G2058"].since == 2020 and by_code["G2058"].until == 2021 +class TestConnectedHelper: + """``_connected`` tested in isolation — the same way ``TestCptEdgesHelper`` + below tests ``_cpt_edges`` directly. In practice ``derive_families``'s + hand-family-split path always calls it with an empty ``unreached`` list + (every code in a >= 2-hand-spanning component is provably reachable by + some hand's own BFS — the nearest seed's shortest path to any node + never crosses a *different* hand's seed, so nothing is ever left over; + confirmed by exhaustive random-graph fuzzing, 300k trials, 0 + counterexamples), so the BFS body here is only reachable by calling + ``_connected`` directly.""" + + def test_bfs_walks_multi_node_components_and_ignores_outside_edges(self): + adj = { + "A": {"B"}, + "B": {"A", "C"}, + "C": {"B"}, + "D": {"E"}, + "E": {"D"}, + "F": set(), + } + comps = _connected(["A", "B", "C", "D", "E", "F"], adj) + comp_sets = sorted((sorted(c) for c in comps), key=lambda c: c[0]) + assert comp_sets == [["A", "B", "C"], ["D", "E"], ["F"]] + + def test_edges_to_nodes_outside_the_set_do_not_count(self): + # "X" is adjacent to "A" but isn't in `nodes` — the component must + # not include it. + adj = {"A": {"B", "X"}, "B": {"A"}, "X": {"A"}} + comps = _connected(["A", "B"], adj) + assert [sorted(c) for c in comps] == [["A", "B"]] + + class TestCptEdgesHelper: def test_skips_only_edges_between_distinct_headings(self): # #687 controller review (Ruling C11), tested against the helper @@ -1223,6 +1329,27 @@ class TestCptPresenceC12: assert by_code["90003"].since == 2024 assert by_code["90003"].until is None + def test_empty_years_tuple_in_cpt_presence_is_skipped(self): + # C12: a code present as a `cpt_presence` key with an *empty* + # years tuple (`not years`) must be skipped rather than crash on + # `min(())`/`max(())` — it simply gets no cpt-derived since/until + # and falls back to the event-based (here: absent) since/until. + sections = [_cpt_section("sec_a", ("Chapter", "Heading A"))] + cpt_codes = [_cpt_code("90001", "sec_a", year=2024)] + cpt_presence = {"90001": (2019, 2021, 2022, 2024), "90099": ()} + rows = derive_families( + {}, + {}, + {"90099": "Empty years widget"}, + cpt_codes=cpt_codes, + cpt_sections=sections, + cpt_presence=cpt_presence, + ) + by_code = {r.code: r for r in rows} + assert by_code["90001"].since == 2019 + assert by_code["90099"].since is None + assert by_code["90099"].until is None + def test_without_cpt_presence_falls_back_to_cpt_codes_alone(self): # Pre-C12 behavior preserved when the caller doesn't have a # multi-edition presence map to give. diff --git a/tests/pfs/test_guidance.py b/tests/pfs/test_guidance.py index 3c610e5..9597d51 100644 --- a/tests/pfs/test_guidance.py +++ b/tests/pfs/test_guidance.py @@ -7,6 +7,7 @@ from __future__ import annotations import duckdb import pytest +import pfs.guidance as guidance_mod from bib.item import Manual, Regulation, Rule from bib.store import Store from pfs.codetables import ( @@ -16,13 +17,15 @@ from pfs.codetables import ( write_cpt_edition, write_guidance, ) -from pfs.cpt_model import CptCode, CptEdition, CptSection +from pfs.cpt_model import CptCode, CptEdition, CptInstruction, CptSection from pfs.guidance import ( BARE_CFR_RE, CFR_RE, IOM_RE, MLN_RE, _cfr_refs, + _extract, + _iom_refs, build, dedupe, harvest, @@ -189,6 +192,26 @@ class TestMlnRegex: assert MLN_RE.search("mln 1234567").group(1) == "1234567" +class TestIomRefsUnmappedManual: + def test_manual_name_missing_from_pub_map_is_dropped(self, monkeypatch): + # IOM_RE's manual-name alternation only ever captures one of the + # six names in _MANUAL_PUB, so "not pub" (guidance.py:201) can't + # fire through the public regex today — it's a defensive guard + # against the map and regex drifting apart. Exercise it directly + # by shrinking the map out from under a real manual-name match. + monkeypatch.setattr(guidance_mod, "_MANUAL_PUB", {}) + text = "See the Benefit Policy Manual, Chapter 5, Section 20 for details." + assert _iom_refs(text) == [] + + +class TestExtractMln: + def test_mln_reference_yields_unresolved_row(self, store): + s, *_rest = store + rows = _extract("See MLN907166 for the telehealth fact sheet.", s) + mln_rows = [r for r in rows if r[0] == "mln"] + assert mln_rows == [("mln", "MLN 907166", "")] + + # ── resolvers ─────────────────────────────────────────────────────── @@ -438,6 +461,126 @@ class TestHarvestCpt: assert r.p_id_src == 0 assert r.page_src == 0 + def test_empty_codes_returns_empty(self, store, con): + s, *_rest = store + assert harvest_cpt(con, s, (), family="CCM") == [] + + def test_non_missing_table_error_is_reraised(self, store, con, monkeypatch): + # cpt_years is imported inside harvest_cpt at call time, so + # patching pfs.codetables.cpt_years is visible to it. A plain + # ValueError doesn't look like a DuckDB "Catalog ... does not + # exist" error, so it must propagate rather than degrade to []. + s, *_rest = store + + def _boom(_con): + raise ValueError("boom") + + monkeypatch.setattr("pfs.codetables.cpt_years", _boom) + with pytest.raises(ValueError, match="boom"): + harvest_cpt(con, s, ("99490",), family="CCM") + + def test_no_years_ingested_returns_empty(self, store, con): + # `con` has ensure_tables run (empty pfs.cpt_section) but no + # edition written, so cpt_years(con) returns [] without raising + # — the "if not years: return []" branch, distinct from the + # missing-table except branch covered via the `bare` connection + # in TestBuild below. + s, *_rest = store + assert harvest_cpt(con, s, ("99490",), family="CCM") == [] + + def test_section_with_no_guideline_text_is_skipped(self, store, con): + # A code mapped to a real section, but the section carries no + # guideline text (e.g. a heading without a trailing "*" range in + # the TOC) — the guideline loop's "continue" must skip it rather + # than call _extract("", ...). + s, *_rest = store + edition = CptEdition( + year=2024, + sections=( + CptSection( + sec_id="sec_1", + level=2, + title="Chronic Care Management Services", + path=("Evaluation and Management",), + code_lo="99490", + code_hi="99490", + guideline="", + ), + ), + codes=(_cpt_code("99490", "sec_1"),), + instructions=(), + references=(), + crosswalks=(), + lists=(), + alternates=(), + ) + write_cpt_edition(con, edition, "CPTED2024") + assert harvest_cpt(con, s, ("99490",), family="CCM") == [] + + def test_instructions_filtered_by_code_kind_and_keyword(self, store, con): + # Exercises every branch of the instructions loop: a "use-with" + # kind is skipped even for a targeted code, a "see" instruction + # for a code outside the family is skipped, a "see" instruction + # for the right code but with no CFR/Medicare/Chapter keyword is + # skipped, and a "see" instruction with a keyword produces a row. + s, sec_key, _manual_key = store + edition = CptEdition( + year=2024, + # cpt_years reads distinct years off pfs.cpt_section, so at + # least one section row is needed for the edition's year to + # be visible at all — its guideline is irrelevant here since + # no code maps to it (codes=()). + sections=( + CptSection( + sec_id="sec_0", + level=2, + title="Unrelated Section", + path=("Unrelated Section",), + code_lo="", + code_hi="", + guideline="", + ), + ), + codes=(), + instructions=( + CptInstruction( + code="99490", + kind="use-with", + text="(Do not report 99490 in conjunction with 42 CFR 410.78)", + targets=(), + ), + CptInstruction( + code="99439", + kind="see", + text="(See 42 CFR 410.78 for conditions of payment)", + targets=(), + ), + CptInstruction( + code="99490", + kind="see", + text="(See the local coverage determination for details)", + targets=(), + ), + CptInstruction( + code="99490", + kind="see", + text="(See 42 CFR 410.78(a)(3) for conditions of payment)", + targets=(), + ), + ), + references=(), + crosswalks=(), + lists=(), + alternates=(), + ) + write_cpt_edition(con, edition, "CPTED2024", force=True) + rows = harvest_cpt(con, s, ("99490",), family="CCM") + assert len(rows) == 1 + r = rows[0] + assert (r.code, r.kind, r.locator) == ("99490", "cfr", "42 CFR 410.78(a)(3)") + assert r.item_key == sec_key + assert r.item_key_src == "CPTED2024" + # ── build (FR + CPT, no CPT tables ingested) ─────────────────────────