afec69c933
C9: some codes print twice with a real descriptor both times — a guideline "Unlisted Service or Procedure" summary table, a "Qualifying Circumstances for Anesthesia" cross-reference reprint, a section-level variant of the same pattern — distinct from C8's placeholder rows (already dropped, never real text). A title-string denylist was tried and rejected: excluding "Unlisted Service" sections silently drops the sole real entries for codes 90749/91299 in the 2022 edition, proving title matching isn't a safe signal on its own. _resolve_alternates (pfs/cpt_epub.py, run at edition level in parse_epub after every chapter is parsed) scores every entry for a duplicated code instead: +2 if the entry's section (or its nearest ancestor with a TOC code range) has a range containing the code, +1 if that section has any range at all, +1 if the entry carries elements or a reference, -3 if any path component ends with "Guidelines". Highest score wins as canonical (ties: first in document order); the rest become CptAlternate(code, sec_id, reason) rows — kept, not dropped. A code that only ever appears in a guideline section has one entry and is untouched. pfs.cpt_model.CptEdition gains `alternates: tuple[CptAlternate, ...] = ()`. pfs.codetables adds pfs.cpt_code_alt (CptCodeAltRow) and write_cpt_edition writes it; the C8 one-row-per-code assertion now passes on all four real editions (verified: zero duplicate codes in 2019/2021/2022/2024).
201 lines
7.5 KiB
Python
201 lines
7.5 KiB
Python
"""Tests for pfs.cpt_load — CPT EPUB editions -> pfs.cpt_* (task 3, #687)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from bib.item import Item
|
|
from bib.store import Store
|
|
from pfs.cpt_load import find_editions, ingest
|
|
|
|
_CHAPTER_2024 = """
|
|
<div class="h1" id="sec_1">Care Management Services</div>
|
|
<div class="h2" id="sec_2">Chronic Care Management Services</div>
|
|
<div class="noindent">Sample guideline text.</div>
|
|
<table class="table1"><tbody>
|
|
<tr>
|
|
<td class="td-w1" id="code_99490"><div class="table-para"><b>99490</b></div></td>
|
|
<td class="td"><div class="table-para1">Chronic care management services with the following required elements:</div>
|
|
<div class="table-slist">first element;</div>
|
|
<div class="table-para2">first 20 minutes of clinical staff time.</div></td>
|
|
</tr>
|
|
<tr>
|
|
<td class="td-w1" id="code_99439"><div class="table-para"><span class="ama-en">✚</span> <b>99439</b></div></td>
|
|
<td class="td"><div class="table-para1-sub">each additional 20 minutes of clinical staff time (List separately in addition to code for primary procedure)</div>
|
|
<div class="table-para2">(Use 99439 in conjunction with 99490)</div></td>
|
|
</tr>
|
|
</tbody></table>
|
|
"""
|
|
|
|
_CHAPTER_2021 = """
|
|
<div class="h1" id="sec_1">Care Management Services</div>
|
|
<div class="h2" id="sec_2">Chronic Care Management Services</div>
|
|
<div class="noindent">Sample guideline text.</div>
|
|
<table class="table1"><tbody>
|
|
<tr>
|
|
<td class="td-w1" id="code_99490"><div class="table-para"><b>99490</b></div></td>
|
|
<td class="td"><div class="table-para1">Chronic care management services with the following required elements:</div>
|
|
<div class="table-slist">first element;</div>
|
|
<div class="table-para2">first 20 minutes of clinical staff time.</div></td>
|
|
</tr>
|
|
</tbody></table>
|
|
"""
|
|
|
|
|
|
def _build_epub(path: Path, chapter_body: str) -> None:
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr("mimetype", "application/epub+zip")
|
|
zf.writestr("OPS/Chapter01.xhtml", f"<html><body>{chapter_body}</body></html>")
|
|
|
|
|
|
def _store(tmp_path) -> Store:
|
|
return Store(tmp_path / "bib.sqlite3", storage_dir=tmp_path / "storage")
|
|
|
|
|
|
def _add_edition(
|
|
store: Store, tmp_path: Path, *, title: str, filename: str, chapter_body: str
|
|
) -> str:
|
|
item_key = store.create(Item(title=title, item_type="book"), tags=["source:ama"])
|
|
src = tmp_path / filename
|
|
_build_epub(src, chapter_body)
|
|
store.attach_file(item_key, src, title=filename)
|
|
return item_key
|
|
|
|
|
|
@pytest.fixture
|
|
def store_with_editions(tmp_path):
|
|
s = _store(tmp_path)
|
|
key_2024 = _add_edition(
|
|
s,
|
|
tmp_path,
|
|
title="CPT Professional 2024 - American Medical Association",
|
|
filename="CPT Professional 2024 - American Medical Association.epub",
|
|
chapter_body=_CHAPTER_2024,
|
|
)
|
|
key_2021 = _add_edition(
|
|
s,
|
|
tmp_path,
|
|
title="CPT 2021 Professional Edition",
|
|
filename="CPT 2021 Professional Edition.epub",
|
|
chapter_body=_CHAPTER_2021,
|
|
)
|
|
# A 2024 PDF sibling attachment — never matched (no .epub suffix).
|
|
pdf = tmp_path / "CPT Professional 2024.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4 not a real pdf")
|
|
s.attach_file(key_2024, pdf, title="CPT Professional 2024.pdf")
|
|
# The 2023 "CPT Changes" book — tagged source:ama, has an .epub, but
|
|
# its title never matches the edition pattern (C2: skip).
|
|
key_changes = s.create(
|
|
Item(title="CPT Changes 2023", item_type="book"), tags=["source:ama"]
|
|
)
|
|
changes_epub = tmp_path / "CPT Changes 2023.epub"
|
|
_build_epub(changes_epub, _CHAPTER_2021)
|
|
s.attach_file(key_changes, changes_epub, title="CPT Changes 2023.epub")
|
|
# Netter's Atlas — tagged source:ama, has an .epub, title doesn't
|
|
# start with "CPT" at all.
|
|
key_netter = s.create(
|
|
Item(title="Netter's Atlas for CPT Coding 2015", item_type="book"),
|
|
tags=["source:ama"],
|
|
)
|
|
netter_epub = tmp_path / "Netter.epub"
|
|
_build_epub(netter_epub, _CHAPTER_2021)
|
|
s.attach_file(key_netter, netter_epub, title="Netter.epub")
|
|
# 2018 PDF-only edition — tagged source:ama, title matches, but no
|
|
# .epub attachment at all (C2: skip).
|
|
key_2018 = s.create(Item(title="CPT 2018", item_type="book"), tags=["source:ama"])
|
|
pdf_2018 = tmp_path / "CPT 2018.pdf"
|
|
pdf_2018.write_bytes(b"%PDF-1.4 not a real pdf")
|
|
s.attach_file(key_2018, pdf_2018, title="CPT 2018.pdf")
|
|
# An item tagged something else entirely, with an .epub and a
|
|
# matching-looking title — must never surface (wrong tag).
|
|
key_other = s.create(
|
|
Item(title="CPT Professional 2099", item_type="book"), tags=["module:coding"]
|
|
)
|
|
other_epub = tmp_path / "CPT Professional 2099.epub"
|
|
_build_epub(other_epub, _CHAPTER_2021)
|
|
s.attach_file(key_other, other_epub, title="CPT Professional 2099.epub")
|
|
|
|
yield s, key_2024, key_2021
|
|
s.close()
|
|
|
|
|
|
class TestFindEditions:
|
|
def test_finds_only_the_matching_epub_editions(self, store_with_editions):
|
|
store, key_2024, key_2021 = store_with_editions
|
|
editions = find_editions(store)
|
|
assert [(year, key) for year, key, _ in editions] == [
|
|
(2021, key_2021),
|
|
(2024, key_2024),
|
|
]
|
|
|
|
def test_paths_point_at_the_epub_not_the_pdf(self, store_with_editions):
|
|
store, key_2024, _ = store_with_editions
|
|
editions = find_editions(store)
|
|
path = next(p for y, k, p in editions if k == key_2024)
|
|
assert path.suffix == ".epub"
|
|
|
|
|
|
class TestIngestDryRun:
|
|
def test_dry_run_never_touches_con(self, store_with_editions):
|
|
store, key_2024, key_2021 = store_with_editions
|
|
out = ingest(store, None, dry_run=True)
|
|
assert set(out) == {2021, 2024}
|
|
assert out[2024] == {
|
|
"sections": 2,
|
|
"codes": 2,
|
|
"instructions": 1,
|
|
"references": 0,
|
|
"crosswalks": 0,
|
|
"lists": 0,
|
|
"alternates": 0,
|
|
}
|
|
assert out[2021]["codes"] == 1
|
|
|
|
def test_dry_run_can_be_scoped_to_one_year(self, store_with_editions):
|
|
store, _, _ = store_with_editions
|
|
out = ingest(store, None, years=[2024], dry_run=True)
|
|
assert set(out) == {2024}
|
|
|
|
def test_unknown_year_raises(self, store_with_editions):
|
|
store, _, _ = store_with_editions
|
|
with pytest.raises(ValueError, match="2099"):
|
|
ingest(store, None, years=[2099], dry_run=True)
|
|
|
|
|
|
class TestIngestWrite:
|
|
def test_writes_rows_per_year(self, store_with_editions):
|
|
store, key_2024, key_2021 = store_with_editions
|
|
con = duckdb.connect(":memory:")
|
|
try:
|
|
out = ingest(store, con, dry_run=False)
|
|
assert out[2024]["codes"] == 2
|
|
assert out[2021]["codes"] == 1
|
|
|
|
rows = con.execute(
|
|
"SELECT code, item_key FROM pfs.cpt_code WHERE edition_year = 2024 ORDER BY code"
|
|
).fetchall()
|
|
assert rows == [("99439", key_2024), ("99490", key_2024)]
|
|
|
|
rows_2021 = con.execute(
|
|
"SELECT code, item_key FROM pfs.cpt_code WHERE edition_year = 2021"
|
|
).fetchall()
|
|
assert rows_2021 == [("99490", key_2021)]
|
|
finally:
|
|
con.close()
|
|
|
|
def test_scoped_to_one_year_leaves_others_absent(self, store_with_editions):
|
|
store, key_2024, _ = store_with_editions
|
|
con = duckdb.connect(":memory:")
|
|
try:
|
|
ingest(store, con, years=[2024], dry_run=False)
|
|
years = con.execute(
|
|
"SELECT DISTINCT edition_year FROM pfs.cpt_code"
|
|
).fetchall()
|
|
assert years == [(2024,)]
|
|
finally:
|
|
con.close()
|