a44c9982fc
write_cpt_edition is delete-then-insert per edition_year; writing an
empty parse would silently erase whatever real data is already on file
for that year. (parse_epub's own guard — raising when _chapter_groups
finds no chapter content at all — landed in the prior commit, since it
shares the same content_root refactor in pfs/cpt_epub.py.)
write_cpt_edition now refuses (raises ValueError) an edition with zero
codes unless force=True. cpt_load.ingest wraps each edition's parse +
write in try/except ValueError, surfaces the failure as
out[year] = {"error": <message>} and continues with the other
requested editions instead of aborting the whole run.
284 lines
10 KiB
Python
284 lines
10 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 TestTitleMismatchWarnsF4:
|
|
def test_unrecognized_title_logs_a_warning_naming_the_item(self, tmp_path, caplog):
|
|
s = _store(tmp_path)
|
|
key = s.create(
|
|
Item(title="Netter's Atlas for CPT Coding 2015", item_type="book"),
|
|
tags=["source:ama"],
|
|
)
|
|
epub = tmp_path / "Netter.epub"
|
|
_build_epub(epub, _CHAPTER_2021)
|
|
s.attach_file(key, epub, title="Netter.epub")
|
|
try:
|
|
with caplog.at_level("WARNING", logger="pfs.cpt_load"):
|
|
editions = find_editions(s)
|
|
assert editions == []
|
|
messages = [r.getMessage() for r in caplog.records]
|
|
assert any(key in m for m in messages)
|
|
assert any("Netter's Atlas" in m for m in messages)
|
|
finally:
|
|
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()
|
|
|
|
|
|
def _build_epub_with_no_chapters(path: Path) -> None:
|
|
"""An EPUB with content files but none named ``ChapterNN`` — parses
|
|
to zero chapter groups, so ``parse_epub`` raises (F4)."""
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr("mimetype", "application/epub+zip")
|
|
zf.writestr(
|
|
"OPS/Frontmatter.xhtml", "<html><body><p>Not a chapter.</p></body></html>"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def store_with_one_good_one_bad_edition(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 = s.create(
|
|
Item(title="CPT 2021 Professional Edition", item_type="book"),
|
|
tags=["source:ama"],
|
|
)
|
|
bad_epub = tmp_path / "CPT 2021 Professional Edition.epub"
|
|
_build_epub_with_no_chapters(bad_epub)
|
|
s.attach_file(key_2021, bad_epub, title="CPT 2021 Professional Edition.epub")
|
|
yield s, key_2024, key_2021
|
|
s.close()
|
|
|
|
|
|
class TestIngestPerEditionFailureF4:
|
|
"""A CPT edition that fails to parse is a per-edition failure, not a
|
|
whole-run one — ingest surfaces it and continues with the others."""
|
|
|
|
def test_write_surfaces_the_bad_edition_and_writes_the_good_one(
|
|
self, store_with_one_good_one_bad_edition
|
|
):
|
|
store, key_2024, key_2021 = store_with_one_good_one_bad_edition
|
|
con = duckdb.connect(":memory:")
|
|
try:
|
|
out = ingest(store, con, dry_run=False)
|
|
assert out[2024]["codes"] == 2
|
|
assert "error" in out[2021]
|
|
assert "no chapter content found" in out[2021]["error"]
|
|
|
|
years = con.execute(
|
|
"SELECT DISTINCT edition_year FROM pfs.cpt_code"
|
|
).fetchall()
|
|
assert years == [(2024,)]
|
|
finally:
|
|
con.close()
|
|
|
|
def test_dry_run_surfaces_the_bad_edition_too(
|
|
self, store_with_one_good_one_bad_edition
|
|
):
|
|
store, key_2024, key_2021 = store_with_one_good_one_bad_edition
|
|
out = ingest(store, None, dry_run=True)
|
|
assert out[2024]["codes"] == 2
|
|
assert "error" in out[2021]
|