6bff63b499
Some checks failed
CI / lint (push) Successful in 45s
CI / notebooks-smoke (push) Has been cancelled
Deploy / notebooks (push) Has been cancelled
Deploy / zotero (push) Has been cancelled
Deploy / docs (push) Has been cancelled
Deploy / api (push) Has been cancelled
Deploy / llm (push) Has been cancelled
Deploy / mc (push) Has been cancelled
Deploy / report (push) Has been cancelled
CI / test (push) Has been cancelled
Infra CI / docs (push) Has been cancelled
Infra CI / api (push) Has been cancelled
Infra CI / llm (push) Has been cancelled
Infra CI / mc (push) Has been cancelled
Infra CI / notebooks (push) Has been cancelled
Infra CI / zotero (push) Has been cancelled
- pfs.guidance: MLN_RE accepts ICN/MLN product numbers ("ICN MLN909188",
"ICN 909289"), MLN_MATTERS_RE captures MM/SE article numbers; mln_refs
normalises both to "MLN <n>" / "MLN Matters MM<n>"; resolve_mln finds
the bib item whose URL or title carries the identifier (word-bounded).
- pfs.guidance.iom_section_page: locates an IOM section heading inside the
chapter PDF attached to the resolved item (llm.pages.locate_section —
the body heading is the one followed by its "(Rev." line, which the
table of contents lacks); cached per item/section and per PDF.
- pfs.codetables.GuidanceRow.page (+ column, added in place by
ensure_tables; legacy NULL reads 0).
- llm.lineage: resolved IOM/MLN guidance rows link to the bib item's URL,
with #page=N when the section was located.
801 lines
29 KiB
Python
801 lines
29 KiB
Python
"""pfs.guidance — sub-regulatory crosswalk (CFR/IOM/MLN) per code family
|
|
(#689, task 6). Regex snippets are real ``fr_anchors`` text (see
|
|
task-6-report.md for item_key/p_id provenance)."""
|
|
|
|
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 (
|
|
GuidanceRow,
|
|
ensure_tables,
|
|
read_guidance,
|
|
write_cpt_edition,
|
|
write_guidance,
|
|
)
|
|
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,
|
|
_guidance_page,
|
|
_iom_refs,
|
|
build,
|
|
dedupe,
|
|
harvest,
|
|
harvest_cpt,
|
|
iom_section_page,
|
|
mln_refs,
|
|
resolve_cfr,
|
|
resolve_iom,
|
|
resolve_mln,
|
|
)
|
|
|
|
# ── real fr_anchors snippets ──────────────────────────────────────────
|
|
|
|
BARE_CFR_TEXT = (
|
|
"Step 3. Review the elements of the service as described by the "
|
|
"HCPCS code and determine whether each of them is capable of being "
|
|
"furnished using an interactive telecommunications system as "
|
|
"defined in § 410.78(a)(3)."
|
|
) # 2KVJ2HKX ¶398
|
|
|
|
IOM_MANUAL_TEXT = (
|
|
"Consistent with the established Medicare HPSA physician bonus "
|
|
"program (Medicare Claims Processing Manual, Pub. 100-04, Chapter "
|
|
"12, Section 90.4.4) and the proposed Health Professional Shortage "
|
|
"Area Surgical Incentive Payment Program (HSIP) described in "
|
|
"section III.S.2. of this proposed rule, we are proposing that "
|
|
"PCIP payments would be calculated by the Medicare contractors."
|
|
) # T4N6JINW ¶1502
|
|
|
|
IOM_PUB_TEXT = (
|
|
"A physician may not bill Medicare for a service that is on the "
|
|
"list of “always therapy” services (see Pub. 100-04, the "
|
|
"Medicare Benefit Policy Manual, chapter 5, section 20) if the "
|
|
"service was done by staff that is not qualified to provide a "
|
|
"skilled therapy service."
|
|
) # WBS77GY5 ¶1771
|
|
|
|
|
|
# ── regexes ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCfrRegex:
|
|
def test_bare_section_infers_no_title(self):
|
|
# BARE_CFR_RE never captures a title — resolve_cfr's caller
|
|
# supplies the inferred "42" (Resolutions: title_inferred kept
|
|
# only in memory, never as a stored flag).
|
|
m = BARE_CFR_RE.search(BARE_CFR_TEXT)
|
|
assert m is not None
|
|
assert m.groups() == ("410", "78", "(a)(3)")
|
|
|
|
def test_titled_cfr_captures_title_part_section_paras(self):
|
|
m = CFR_RE.search("under 42 CFR 410.78(a)(3) and elsewhere")
|
|
assert m is not None
|
|
assert (
|
|
m.group("title"),
|
|
m.group("part"),
|
|
m.group("dec"),
|
|
m.group("paras"),
|
|
) == (
|
|
"42",
|
|
"410",
|
|
"78",
|
|
"(a)(3)",
|
|
)
|
|
assert m.group("partword") is None
|
|
|
|
def test_titled_cfr_part_only(self):
|
|
m = CFR_RE.search("see 42 CFR Part 425 for details")
|
|
assert m is not None
|
|
assert (m.group("title"), m.group("partword"), m.group("part")) == (
|
|
"42",
|
|
"Part",
|
|
"425",
|
|
)
|
|
assert m.group("dec") is None and m.group("paras") is None
|
|
|
|
def test_titled_cfr_plural_parts_only(self):
|
|
m = CFR_RE.search("see 42 CFR parts 405 for details")
|
|
assert m is not None
|
|
assert (m.group("title"), m.group("partword"), m.group("part")) == (
|
|
"42",
|
|
"parts",
|
|
"405",
|
|
)
|
|
|
|
def test_dotted_cfr_with_section_sign_resolves_via_bare_fallback(self):
|
|
# CFR_RE (titled) has no "§" between "C.F.R." and the number —
|
|
# the bare regex's title-inferred fallback picks it up instead,
|
|
# landing on the same (title, section) either way.
|
|
assert CFR_RE.search("42 C.F.R. § 414.1425 governs this") is None
|
|
assert _cfr_refs("42 C.F.R. § 414.1425 governs this") == [("42", "414.1425")]
|
|
|
|
|
|
class TestCfrListContinuation:
|
|
"""Ruling A10: a titled CFR head followed by a list continuation
|
|
(comma/and/or/through-separated bare tokens) captures every
|
|
section/part in the run, not just the head."""
|
|
|
|
def test_comma_list_of_sections(self):
|
|
assert _cfr_refs(
|
|
"the current regulations at 42 CFR 410.20, 410.26, and 410.32 for CCM"
|
|
) == [("42", "410.20"), ("42", "410.26"), ("42", "410.32")]
|
|
|
|
def test_and_joined_paragraph_cites(self):
|
|
assert _cfr_refs("… 42 CFR 410.26(a)(3) and 410.26(b) to better define …") == [
|
|
("42", "410.26(a)(3)"),
|
|
("42", "410.26(b)"),
|
|
]
|
|
|
|
def test_plural_parts_list(self):
|
|
assert _cfr_refs("… 42 CFR parts 405, 414, and 426 …") == [
|
|
("42", "Part 405"),
|
|
("42", "Part 414"),
|
|
("42", "Part 426"),
|
|
]
|
|
|
|
def test_bare_double_section_plural(self):
|
|
assert _cfr_refs("§§ 410.26 and 410.32 govern this") == [
|
|
("42", "410.26"),
|
|
("42", "410.32"),
|
|
]
|
|
|
|
def test_no_continuation_across_prose(self):
|
|
# "the" is not a citation token — the continuation grammar must
|
|
# not fire on ordinary prose following "and".
|
|
assert _cfr_refs("42 CFR 410.78 and the physician") == [("42", "410.78")]
|
|
|
|
def test_singular_part_does_not_trigger_continuation(self):
|
|
# Only a plural "parts" head triggers the bare-part-number
|
|
# continuation grammar (Ruling A10) — a singular "Part 425"
|
|
# followed by an unrelated number must not absorb it.
|
|
assert _cfr_refs("42 CFR Part 425 and 42 other things") == [("42", "Part 425")]
|
|
|
|
|
|
class TestIomRegex:
|
|
def test_manual_name_form(self):
|
|
m = IOM_RE.search(IOM_MANUAL_TEXT)
|
|
assert m is not None
|
|
pub_digit, manual_name, chapter, section = m.groups()
|
|
# The pub-number branch fires (it starts later in the string but
|
|
# is the only branch the Manual-name gap can't cross — "Pub."
|
|
# has a literal period the [^.]{0,120}? gap can't skip past).
|
|
assert pub_digit == "4"
|
|
assert manual_name is None
|
|
assert chapter == "12"
|
|
assert section == "90.4.4"
|
|
|
|
def test_pub_number_form(self):
|
|
m = IOM_RE.search(IOM_PUB_TEXT)
|
|
assert m is not None
|
|
pub_digit, manual_name, chapter, section = m.groups()
|
|
assert pub_digit == "4"
|
|
assert chapter == "5"
|
|
assert section == "20"
|
|
|
|
def test_no_match_without_chapter(self):
|
|
assert IOM_RE.search("Pub. 100-04 discusses billing generally.") is None
|
|
|
|
|
|
class TestMlnRegex:
|
|
def test_matches_booklet_number(self):
|
|
m = MLN_RE.search("see MLN907166 for details")
|
|
assert m is not None
|
|
assert m.group(1) == "907166"
|
|
|
|
def test_case_insensitive_with_space(self):
|
|
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 ───────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def store():
|
|
s = Store(":memory:")
|
|
s.create(
|
|
Regulation(
|
|
title="42 CFR Part 410",
|
|
url="https://www.ecfr.gov/current/title-42/part-410",
|
|
)
|
|
)
|
|
sec_key = s.create(
|
|
Regulation(
|
|
title="42 CFR 410.78",
|
|
url="https://www.ecfr.gov/current/title-42/section-410.78",
|
|
)
|
|
)
|
|
manual_key = s.create(
|
|
Manual(
|
|
title=(
|
|
"Medicare Claims Processing Manual — Chapter 12: "
|
|
"Physicians/Nonphysician Practitioners"
|
|
)
|
|
)
|
|
)
|
|
s.add_tag(manual_key, "pub:100-04")
|
|
yield s, sec_key, manual_key
|
|
s.close()
|
|
|
|
|
|
class TestResolveCfr:
|
|
def test_resolves_exact_section(self, store):
|
|
s, sec_key, _ = store
|
|
assert resolve_cfr(s, "42", "410.78(a)(3)") == sec_key
|
|
|
|
def test_unresolved_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_cfr(s, "45", "155.20") == ""
|
|
|
|
|
|
class TestResolveIom:
|
|
def test_resolves_chapter_by_pub_and_title(self, store):
|
|
s, _, manual_key = store
|
|
assert resolve_iom(s, "100-04", "12") == manual_key
|
|
|
|
def test_unresolved_pub_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_iom(s, "100-02", "12") == ""
|
|
|
|
def test_unresolved_chapter_returns_empty(self, store):
|
|
s, _, _ = store
|
|
assert resolve_iom(s, "100-04", "5") == ""
|
|
|
|
|
|
# ── harvest (FR paragraphs) ────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def store_with_anchors(store):
|
|
s, sec_key, manual_key = store
|
|
con = s._con() # noqa: SLF001
|
|
item_key = s.create(Rule(title="CY2020 PFS final rule"))
|
|
con.executemany(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
[
|
|
(item_key, 398, 32389, 1, f"For code 99490, {BARE_CFR_TEXT}"),
|
|
(item_key, 1502, 100, 2, f"Codes 99490 and 99439: {IOM_MANUAL_TEXT}"),
|
|
(item_key, 9, 5, 3, "no codes named here at all"),
|
|
],
|
|
)
|
|
con.commit()
|
|
yield s, item_key, sec_key, manual_key
|
|
s.close()
|
|
|
|
|
|
class TestHarvest:
|
|
def test_bare_cfr_paragraph_resolves(self, store_with_anchors):
|
|
s, item_key, sec_key, _manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
cfr_rows = [r for r in rows if r.kind == "cfr"]
|
|
assert len(cfr_rows) == 1
|
|
r = cfr_rows[0]
|
|
assert r.family == "CCM"
|
|
assert r.code == "99490"
|
|
assert r.locator == "42 CFR 410.78(a)(3)"
|
|
assert r.item_key == sec_key
|
|
assert r.item_key_src == item_key
|
|
assert r.p_id_src == 398
|
|
assert r.page_src == 32389
|
|
|
|
def test_iom_paragraph_names_both_codes(self, store_with_anchors):
|
|
s, item_key, _sec_key, manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
iom_rows = [r for r in rows if r.kind == "iom"]
|
|
codes = {r.code for r in iom_rows}
|
|
assert codes == {"99490", "99439"}
|
|
assert all(r.locator == "100-04 ch.12 §90.4.4" for r in iom_rows)
|
|
assert all(r.item_key == manual_key for r in iom_rows)
|
|
|
|
def test_paragraph_without_codes_is_skipped(self, store_with_anchors):
|
|
s, _item_key, _sec_key, _manual_key = store_with_anchors
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
assert all(r.p_id_src != 9 for r in rows)
|
|
|
|
def test_empty_codes_returns_empty(self, store_with_anchors):
|
|
s, *_rest = store_with_anchors
|
|
assert harvest(s, (), family="CCM") == []
|
|
|
|
def test_fr_citation_page_number_never_fires_as_a_code(self, store_with_anchors):
|
|
# A code equal to an FR citation's page number ("43842") must
|
|
# not be treated as present just because "91 FR 43842" appears.
|
|
s, item_key, _sec_key, _manual_key = store_with_anchors
|
|
con = s._con() # noqa: SLF001
|
|
con.execute(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
(item_key, 500, 1, 4, "See 91 FR 43842 for background."),
|
|
)
|
|
con.commit()
|
|
rows = harvest(s, ("43842",), family="FAKE")
|
|
assert rows == []
|
|
|
|
|
|
# ── dedupe ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestDedupe:
|
|
def test_collapses_identical_keys(self):
|
|
r = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
assert dedupe([r, r]) == [r]
|
|
|
|
def test_keeps_distinct_codes(self):
|
|
r1 = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
r2 = GuidanceRow("CCM", "99439", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
|
|
assert dedupe([r1, r2]) == [r1, r2]
|
|
|
|
|
|
# ── DDL / write / read round trip ────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def con():
|
|
c = duckdb.connect(":memory:")
|
|
ensure_tables(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
class TestGuidanceTable:
|
|
def test_write_then_read_round_trips(self, con):
|
|
rows = [
|
|
GuidanceRow(
|
|
"CCM", "99490", "cfr", "42 CFR 410.78(a)(3)", "SEC", "K", 398, 32389
|
|
),
|
|
GuidanceRow(
|
|
"CCM", "99439", "iom", "100-04 ch.12 §90.4.4", "", "K", 1502, 100
|
|
),
|
|
]
|
|
n = write_guidance(con, "CCM", rows)
|
|
assert n == 2
|
|
got = read_guidance(con, "CCM")
|
|
assert {(r.code, r.kind) for r in got} == {("99490", "cfr"), ("99439", "iom")}
|
|
|
|
def test_write_is_delete_then_insert_per_family(self, con):
|
|
write_guidance(
|
|
con, "CCM", [GuidanceRow("CCM", "99490", "cfr", "x", "", "K", 1, 1)]
|
|
)
|
|
write_guidance(
|
|
con, "CCM", [GuidanceRow("CCM", "99439", "cfr", "y", "", "K", 2, 2)]
|
|
)
|
|
got = read_guidance(con, "CCM")
|
|
assert [r.code for r in got] == ["99439"]
|
|
|
|
def test_other_family_untouched(self, con):
|
|
write_guidance(
|
|
con, "APCM", [GuidanceRow("APCM", "G0556", "cfr", "x", "", "K", 1, 1)]
|
|
)
|
|
write_guidance(con, "CCM", [])
|
|
assert len(read_guidance(con, "APCM")) == 1
|
|
|
|
|
|
# ── harvest_cpt (non-empty: a real CFR cite in the guideline text) ───
|
|
|
|
|
|
def _cpt_code(code: str, sec_id: str) -> CptCode:
|
|
return CptCode(
|
|
code=code,
|
|
sec_id=sec_id,
|
|
descriptor="d",
|
|
stem="s",
|
|
elements=(),
|
|
tail="",
|
|
addon=False,
|
|
resequenced=False,
|
|
new=False,
|
|
revised=False,
|
|
telemedicine=False,
|
|
parent="",
|
|
mod51_exempt=False,
|
|
audio_only=False,
|
|
fda_pending=False,
|
|
pla=False,
|
|
category="I",
|
|
)
|
|
|
|
|
|
class TestHarvestCpt:
|
|
def test_guideline_cfr_cite_produces_a_row(self, store, con):
|
|
s, sec_key, _manual_key = store
|
|
edition = CptEdition(
|
|
year=2024,
|
|
sections=(
|
|
CptSection(
|
|
sec_id="sec_1",
|
|
level=2,
|
|
title="Chronic Care Management Services",
|
|
path=(
|
|
"Evaluation and Management",
|
|
"Chronic Care Management Services",
|
|
),
|
|
code_lo="99490",
|
|
code_hi="99490",
|
|
guideline=(
|
|
"See 42 CFR 410.78(a)(3) for telehealth conditions of payment."
|
|
),
|
|
),
|
|
),
|
|
codes=(_cpt_code("99490", "sec_1"),),
|
|
instructions=(),
|
|
references=(),
|
|
crosswalks=(),
|
|
lists=(),
|
|
alternates=(),
|
|
)
|
|
write_cpt_edition(con, edition, "CPTED2024")
|
|
rows = harvest_cpt(con, s, ("99490",), family="CCM")
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert (r.family, r.code, r.kind, r.locator) == (
|
|
"CCM",
|
|
"99490",
|
|
"cfr",
|
|
"42 CFR 410.78(a)(3)",
|
|
)
|
|
assert r.item_key == sec_key
|
|
assert r.item_key_src == "CPTED2024"
|
|
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) ─────────────────────────
|
|
|
|
|
|
class TestBuild:
|
|
def test_build_tolerates_missing_cpt_tables(self, store_with_anchors):
|
|
# A bare connection (no ensure_tables!) so `cpt_years` actually
|
|
# raises a Catalog "does not exist" error and harvest_cpt's
|
|
# except-is_missing_table_error branch is exercised — the `con`
|
|
# fixture already has an (empty) pfs.cpt_section via
|
|
# ensure_tables, which only exercises the "if not years: return
|
|
# []" branch instead.
|
|
s, *_rest = store_with_anchors
|
|
bare = duckdb.connect(":memory:")
|
|
try:
|
|
rows = build(bare, s, ("99490", "99439"), family="CCM")
|
|
assert rows # the FR-sourced rows still come through
|
|
assert harvest_cpt(bare, s, ("99490",), family="CCM") == []
|
|
finally:
|
|
bare.close()
|
|
|
|
|
|
# ── MLN (#705 item 3) ───────────────────────────────────────────────
|
|
|
|
|
|
class TestMlnRefs:
|
|
def test_icn_prefix_and_icn_mln_combo_normalise_to_mln(self):
|
|
text = (
|
|
"MLN Booklet “Chronic Care Management Services” (ICN MLN909188, July "
|
|
"2019); see also “Advance Care Planning” (ICN 909289, August 2016)."
|
|
)
|
|
assert mln_refs(text) == ["MLN 909188", "MLN 909289"]
|
|
|
|
def test_matters_article_forms(self):
|
|
assert mln_refs("MLN Matters® Number MM9603: https://x") == [
|
|
"MLN Matters MM9603"
|
|
]
|
|
assert mln_refs("as stated in MLN Matters article SE1316, issued") == [
|
|
"MLN Matters SE1316"
|
|
]
|
|
assert mln_refs("documentation is available in MLN Matters SE 1316.") == [
|
|
"MLN Matters SE1316"
|
|
]
|
|
# a bare number after "Matters" is a change-request article
|
|
assert mln_refs("consistent with MLN Matters article 11268.") == [
|
|
"MLN Matters MM11268"
|
|
]
|
|
|
|
def test_prose_mentions_without_a_number_yield_nothing(self):
|
|
assert (
|
|
mln_refs("We will issue an MLN Matters article once the CR is out.") == []
|
|
)
|
|
assert mln_refs("Medicare Learning Network (MLN) Matters® article,") == []
|
|
|
|
def test_first_occurrence_wins(self):
|
|
assert mln_refs("MLN006764 ... again MLN 006764") == ["MLN 006764"]
|
|
|
|
|
|
@pytest.fixture
|
|
def mln_store():
|
|
s = Store(":memory:")
|
|
booklet = s.create(
|
|
Manual(
|
|
title="Chronic Care Management Services (MLN909188, June 2025) — CCM",
|
|
url="https://www.cms.gov/files/document/chroniccaremanagement.pdf",
|
|
)
|
|
)
|
|
em_guide = s.create(
|
|
Manual(
|
|
title="Evaluation and Management Services Guide",
|
|
url=(
|
|
"https://www.cms.gov/Outreach-and-Education/Medicare-Learning-"
|
|
"Network-MLN/MLNProducts/Downloads/eval-mgmt-serv-guide-ICN006764.pdf"
|
|
),
|
|
)
|
|
)
|
|
article = s.create(
|
|
Manual(
|
|
title="MLN Matters MM9603",
|
|
url=(
|
|
"https://www.cms.gov/Outreach-and-Education/Medicare-Learning-"
|
|
"Network-MLN/MLNMattersArticles/Downloads/MM9603.pdf"
|
|
),
|
|
)
|
|
)
|
|
decoy = s.create(
|
|
Manual(title="Unrelated", url="https://www.cms.gov/files/MM96030.pdf")
|
|
)
|
|
yield s, booklet, em_guide, article, decoy
|
|
s.close()
|
|
|
|
|
|
class TestResolveMln:
|
|
def test_title_carries_the_number(self, mln_store):
|
|
s, booklet, *_ = mln_store
|
|
assert resolve_mln(s, "MLN 909188") == booklet
|
|
|
|
def test_pre_2020_icn_spelling_in_url(self, mln_store):
|
|
s, _b, em_guide, *_ = mln_store
|
|
assert resolve_mln(s, "MLN 006764") == em_guide
|
|
|
|
def test_matters_article_by_url_word_bounded(self, mln_store):
|
|
s, _b, _e, article, _decoy = mln_store
|
|
# MM9603 must not match the decoy's MM96030
|
|
assert resolve_mln(s, "MLN Matters MM9603") == article
|
|
assert resolve_mln(s, "MLN Matters MM9604") == ""
|
|
|
|
def test_unresolved_and_empty(self, mln_store):
|
|
s, *_ = mln_store
|
|
assert resolve_mln(s, "MLN 907166") == ""
|
|
assert resolve_mln(s, "") == ""
|
|
|
|
|
|
class TestExtractMlnResolved:
|
|
def test_extract_resolves_against_store(self, mln_store):
|
|
s, booklet, *_ = mln_store
|
|
rows = _extract("Refer to the CCM booklet (ICN MLN909188, July 2019).", s)
|
|
assert rows == [("mln", "MLN 909188", booklet)]
|
|
|
|
|
|
# ── IOM section page locating (#705 item 2) ─────────────────────────
|
|
|
|
|
|
class TestIomSectionPage:
|
|
TOC = "Table of Contents 30.6.4 - E/M Services Incident to 30.6.5 - Groups"
|
|
BODY = "30.6.4 - E/M Services Incident to (Rev. 11288; Issued: 03-31-22) A."
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_caches(self, monkeypatch):
|
|
monkeypatch.setattr(guidance_mod, "_PAGE_CACHE", {})
|
|
monkeypatch.setattr(guidance_mod, "_PDF_CACHE", {})
|
|
|
|
@pytest.fixture
|
|
def chapter(self, tmp_path, monkeypatch):
|
|
# Own store: an in-memory Store's default storage dir is ./storage
|
|
# relative to the cwd, which attach_file would create in the repo.
|
|
s = Store(":memory:", storage_dir=tmp_path / "storage")
|
|
manual_key = s.create(
|
|
Manual(title="Medicare Claims Processing Manual — Chapter 12: X")
|
|
)
|
|
pdf = tmp_path / "clm104c12.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4 stub")
|
|
s.attach_file(manual_key, pdf)
|
|
calls: list[str] = []
|
|
|
|
def fake_pages(path):
|
|
calls.append(str(path))
|
|
return [self.TOC, "filler", self.BODY]
|
|
|
|
monkeypatch.setattr("llm.pages.pdf_pages", fake_pages)
|
|
yield s, manual_key, calls
|
|
s.close()
|
|
|
|
def test_locates_body_heading_past_the_toc(self, chapter):
|
|
s, key, _ = chapter
|
|
assert iom_section_page(s, key, "30.6.4") == 3
|
|
|
|
def test_pdf_read_once_and_page_cached(self, chapter):
|
|
s, key, calls = chapter
|
|
assert iom_section_page(s, key, "30.6.4") == 3
|
|
assert iom_section_page(s, key, "30.6.4") == 3
|
|
assert iom_section_page(s, key, "30.6.5") == 0 # TOC-only
|
|
assert len(calls) == 1
|
|
|
|
def test_no_pdf_attachment_or_blank_args(self, store):
|
|
s, _sec, manual_key = store
|
|
assert iom_section_page(s, manual_key, "30.6.4") == 0
|
|
assert iom_section_page(s, "", "30.6.4") == 0
|
|
assert iom_section_page(s, manual_key, "") == 0
|
|
|
|
def test_guidance_page_only_for_resolved_iom_sections(self, chapter):
|
|
s, key, _ = chapter
|
|
assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", key) == 3
|
|
assert _guidance_page(s, "iom", "100-04 ch.12", key) == 0
|
|
assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", "") == 0
|
|
assert _guidance_page(s, "cfr", "42 CFR 410.78", key) == 0
|
|
|
|
def test_harvest_carries_the_located_page(self, store_with_anchors, monkeypatch):
|
|
s, _item, _sec, manual_key = store_with_anchors
|
|
monkeypatch.setattr(
|
|
guidance_mod,
|
|
"iom_section_page",
|
|
lambda st, k, sec: 39 if k == manual_key else 0,
|
|
)
|
|
rows = harvest(s, ("99490", "99439"), family="CCM")
|
|
iom = [r for r in rows if r.kind == "iom"]
|
|
assert iom and all(r.page == 39 for r in iom)
|
|
assert all(r.page == 0 for r in rows if r.kind != "iom")
|
|
|
|
|
|
class TestGuidanceTablePage:
|
|
def test_page_round_trips(self, con):
|
|
row = GuidanceRow(
|
|
"CCM", "99490", "iom", "100-04 ch.12 §30.6.4", "K", "S", 1, 2, 39
|
|
)
|
|
write_guidance(con, "CCM", [row])
|
|
assert read_guidance(con, "CCM") == [row]
|
|
|
|
def test_legacy_table_without_page_column_reads_zero(self):
|
|
c = duckdb.connect(":memory:")
|
|
c.execute("CREATE SCHEMA pfs")
|
|
c.execute(
|
|
"CREATE TABLE pfs.code_guidance (family VARCHAR, code VARCHAR, kind VARCHAR, "
|
|
"locator VARCHAR, item_key VARCHAR, item_key_src VARCHAR, p_id_src INTEGER, "
|
|
"page_src INTEGER)"
|
|
)
|
|
c.execute(
|
|
"INSERT INTO pfs.code_guidance VALUES ('CCM','99490','iom','x','','S',1,2)"
|
|
)
|
|
ensure_tables(c) # adds the column in place; the old row reads NULL
|
|
got = read_guidance(c, "CCM")
|
|
assert got == [GuidanceRow("CCM", "99490", "iom", "x", "", "S", 1, 2, 0)]
|
|
c.close()
|