diff --git a/notebooks/code_families.py b/notebooks/code_families.py index 773a755..d5938b6 100644 --- a/notebooks/code_families.py +++ b/notebooks/code_families.py @@ -343,12 +343,12 @@ def _(code, mo, not_built, pl, q): def _(code, mo, not_built, q): # ── 2. What the extractor wrote ── _els = q( - "SELECT type, value, detail, source, item_key, p_id, page FROM pfs.code_element " - "WHERE code = ? ORDER BY type, value", + "SELECT type, value, detail, source, confirmed_by, item_key, p_id, page " + "FROM pfs.code_element WHERE code = ? ORDER BY type, value", (code,), ) _rev = q( - "SELECT text, proposed_value, item_key, p_id FROM pfs.code_element_review " + "SELECT text, proposed_value, source, item_key, p_id FROM pfs.code_element_review " "WHERE code = ? ORDER BY p_id", (code,), ) @@ -367,7 +367,9 @@ def _(code, mo, not_built, q): "the closed list, or `none`**; whatever neither pass can place is queued for " "human review rather than guessed. Nothing enters `pfs.code_element` unless " "it is a member of the closed vocabulary, and every row keeps the exact " - "Federal Register paragraph it came from." + "Federal Register paragraph it came from. When a second source (the CPT " + "manual, HCPCS, RVU) independently transcribes the same element, that source " + "isn't a second row — it's recorded in `confirmed_by` on the winning row." ), mo.ui.table( _els, label=f"pfs.code_element — {code} ({_els.height} rows)" diff --git a/src/cli/pfs.py b/src/cli/pfs.py index ffbd309..1b76b02 100644 --- a/src/cli/pfs.py +++ b/src/cli/pfs.py @@ -28,10 +28,12 @@ from pfs.codetables import ( is_missing_table_error, read_all_elements, read_all_events, + read_all_reviews, read_cpt_codes, read_cpt_instructions, read_cpt_sections, read_guidance, + read_reviews, write_elements, write_events, write_families, @@ -113,7 +115,9 @@ def _run_elements( x = extract_code(store, con, c, classify=classify) if write: write_elements(con, c, x.rows, x.reviews) - typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review") + confirmed = sum(1 for r in x.rows if r.confirmed_by) + note = f", {confirmed} confirmed by a second source" if confirmed else "" + typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review{note}") def _codes_for( @@ -642,17 +646,13 @@ def reaction( def review(code: str = typer.Option("", "--code")) -> None: """Element lines the classifier could not place.""" con = _read() - sql = ( - "SELECT code, text, proposed_value, item_key, p_id FROM pfs.code_element_review" - ) - params: list[Any] = [] - if code: - sql += " WHERE code = ?" - params.append(code.upper()) - for c, text, proposed, key, p_id in con.execute( - sql + " ORDER BY code, p_id", params - ).fetchall(): - typer.echo(f"{c} {key} ¶{p_id} [{proposed or '?'}] {text[:120]}") + rows = read_reviews(con, code.upper()) if code else read_all_reviews(con) + for r in rows: + src = f" ({r.source})" if r.source else "" + typer.echo( + f"{r.code} {r.item_key} ¶{r.p_id}{src} [{r.proposed_value or '?'}] " + f"{r.text[:120]}" + ) def _print_cpt_counts(counts: dict[int, dict[str, int]]) -> None: diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index ea2db9c..b075d30 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -47,6 +47,16 @@ class ElementRow: p_id: int page: int source: str # "fr" | "cpt" | "hcpcs" | "rvu" + #: Comma-separated ``source:item_key`` tokens (e.g. ``"cpt:GQGTPGYV"``) + #: — one per additional source whose own independent transcription + #: yielded the same ``(type, value, detail)`` as this row (#703). The + #: winning row (FR over CPT over HCPCS/RVU, unchanged) is the one kept; + #: this column is the cross-check the dedupe used to discard entirely. + #: ``""`` when no second source confirmed this element. Last field, + #: defaulted, so existing callers that build an ``ElementRow`` without + #: it keep working (Ruling C1 pattern), and a replica read before this + #: column existed tolerates the short tuple the same way. + confirmed_by: str = "" @dataclass(frozen=True) @@ -57,6 +67,11 @@ class ReviewRow: proposed_value: str item_key: str p_id: int + #: "fr" | "cpt" | "hcpcs" | "rvu" — which source's pass produced this + #: unplaced line (#703). Last field, defaulted, so existing callers + #: that build a ``ReviewRow`` without it keep working, and a replica + #: read before this column existed tolerates the short tuple. + source: str = "" @dataclass(frozen=True) @@ -233,10 +248,11 @@ _DDL = """ CREATE SCHEMA IF NOT EXISTS pfs; CREATE TABLE IF NOT EXISTS pfs.code_element ( code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, detail VARCHAR, - text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR); + text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, + confirmed_by VARCHAR); CREATE TABLE IF NOT EXISTS pfs.code_element_review ( code VARCHAR, text VARCHAR, proposed_type VARCHAR, proposed_value VARCHAR, - item_key VARCHAR, p_id INTEGER); + item_key VARCHAR, p_id INTEGER, source VARCHAR); CREATE TABLE IF NOT EXISTS pfs.code_event ( code VARCHAR, year INTEGER, kind VARCHAR, from_codes VARCHAR, to_codes VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, anchored BOOLEAN, note VARCHAR); @@ -295,6 +311,15 @@ def ensure_tables(con: Any) -> None: # `note` existed) needs the column added — CREATE TABLE IF NOT EXISTS # above is a no-op once the table already exists. con.execute("ALTER TABLE pfs.code_family ADD COLUMN IF NOT EXISTS note VARCHAR") + # #703, same pattern: pre-existing pfs.code_element/code_element_review + # tables (created before confirmed_by/source existed) need the columns + # added in place. + con.execute( + "ALTER TABLE pfs.code_element ADD COLUMN IF NOT EXISTS confirmed_by VARCHAR" + ) + con.execute( + "ALTER TABLE pfs.code_element_review ADD COLUMN IF NOT EXISTS source VARCHAR" + ) def _insert(con: Any, table: str, rows: Sequence[Any]) -> int: @@ -336,6 +361,20 @@ def read_elements(con: Any, code: str) -> list[ElementRow]: return [ElementRow(*r) for r in rows] +def read_reviews(con: Any, code: str) -> list[ReviewRow]: + rows = con.execute( + "SELECT * FROM pfs.code_element_review WHERE code = ? ORDER BY p_id", [code] + ).fetchall() + return [ReviewRow(*r) for r in rows] + + +def read_all_reviews(con: Any) -> list[ReviewRow]: + rows = con.execute( + "SELECT * FROM pfs.code_element_review ORDER BY code, p_id" + ).fetchall() + return [ReviewRow(*r) for r in rows] + + def read_events(con: Any, code: str) -> list[EventRow]: rows = con.execute( "SELECT * FROM pfs.code_event WHERE code = ? ORDER BY year, kind", [code] diff --git a/src/pfs/extract.py b/src/pfs/extract.py index 77daf23..d57fcf6 100644 --- a/src/pfs/extract.py +++ b/src/pfs/extract.py @@ -7,13 +7,16 @@ description, RVU short description); ``extract_code`` gathers every source for a code — FR, then the newest CPT codebook edition's own ``pfs.cpt_code`` row (stem + tail, then each required-elements list item), then HCPCS, then RVU — with FR rows winning over CPT and CPT -winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. No I/O -here beyond what the caller hands in. +winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. The +losing source isn't just dropped: it's recorded on the winning row's +``confirmed_by`` as a ``source:item_key`` token (#703) — an independent +transcription of the same element cross-checks the winner without +doubling the row count. No I/O here beyond what the caller hands in. """ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, Callable, Sequence from pfs.codetables import ElementRow, ReviewRow, is_missing_table_error @@ -41,6 +44,35 @@ class Extraction: reviews: tuple[ReviewRow, ...] +def _merge_or_confirm( + merged: dict[tuple[str, str, str], ElementRow], r: ElementRow +) -> None: + """Add *r* to *merged*, keyed by ``(type, value, detail)``. A first + sighting of a key is just kept. A later row for a key already held by + a *different* source (#703) doesn't replace the winner — FR still + beats CPT, CPT still beats HCPCS/RVU, and the row count doesn't grow + — it cross-checks it: ``r``'s ``source:item_key`` is appended to the + kept row's ``confirmed_by`` (deduped, so re-running the same source + twice, or a code with more than one CPT hit, doesn't repeat a token). + A later row from the *same* source as the one already kept is a + same-source duplicate (e.g. two FR rule years naming the same + element), not a second, independent source — it's dropped exactly as + before, uncounted as confirmation.""" + key = (r.type, r.value, r.detail) + existing = merged.get(key) + if existing is None: + merged[key] = r + return + if r.source == existing.source: + return + token = f"{r.source}:{r.item_key}" + tokens = [t for t in existing.confirmed_by.split(",") if t] + if token in tokens: + return + tokens.append(token) + merged[key] = replace(existing, confirmed_by=",".join(tokens)) + + def _row(code: str, e: Element, para: Para, *, year: int, source: str) -> ElementRow: return ElementRow( code, @@ -75,7 +107,7 @@ def extract_run( rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr")) else: reviews.append( - ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id) + ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id, "fr") ) # 3. deterministic on the stem (and anything the whole text reveals) for e in parse_descriptor(run.text): @@ -149,7 +181,7 @@ def _cpt_elements( e = Element(_TYPE_OF[choice], choice) rows.append(_cpt_row(code, edition_year, item_key, e, text)) else: - reviews.append(ReviewRow(code, text, "", "", item_key, 0)) + reviews.append(ReviewRow(code, text, "", "", item_key, 0, "cpt")) return rows, reviews @@ -165,20 +197,20 @@ def extract_code( for run in descriptor_runs(store, code): x = extract_run(run, classify=classify) for r in x.rows: - merged.setdefault((r.type, r.value, r.detail), r) + _merge_or_confirm(merged, r) reviews.extend(x.reviews) cpt_rows, cpt_reviews = _cpt_elements(con, code, classify=classify) for r in cpt_rows: - merged.setdefault((r.type, r.value, r.detail), r) + _merge_or_confirm(merged, r) reviews.extend(cpt_reviews) long_desc = hcpcs_long_description(con, code) years = rvu_descriptions(con, code) if long_desc: y = years[-1][0] if years else 0 for r in extract_text(code, long_desc, year=y, source="hcpcs").rows: - merged.setdefault((r.type, r.value, r.detail), r) + _merge_or_confirm(merged, r) for year, _status, desc in years: for r in extract_text(code, desc, year=year, source="rvu").rows: - merged.setdefault((r.type, r.value, r.detail), r) + _merge_or_confirm(merged, r) ordered = sorted(merged.values(), key=lambda r: (r.type, r.value, r.detail)) return Extraction(code, tuple(ordered), tuple(reviews)) diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py index b0b9b30..72af1ab 100644 --- a/tests/cli/test_pfs_cli.py +++ b/tests/cli/test_pfs_cli.py @@ -128,6 +128,59 @@ class TestElements: 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_, code, *, classify=None): + return 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", + ), + ), + (), + ) + + monkeypatch.setattr(pfs_cli, "extract_code", 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_code", + lambda s, c, code, *, classify=None: Extraction(code, (), ()), + ) + 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, @@ -930,3 +983,33 @@ class TestReview: ) 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 diff --git a/tests/pfs/test_codetables.py b/tests/pfs/test_codetables.py index 9958c05..015c9ad 100644 --- a/tests/pfs/test_codetables.py +++ b/tests/pfs/test_codetables.py @@ -15,12 +15,14 @@ from pfs.codetables import ( is_missing_table_error, read_all_elements, read_all_events, + read_all_reviews, read_cpt_codes, read_cpt_instructions, read_cpt_sections, read_elements, read_events, read_families, + read_reviews, write_cpt_edition, write_elements, write_events, @@ -96,6 +98,106 @@ class TestDDL: "cpt_code_alt", } <= names + def test_ensure_tables_upgrades_a_pre_existing_element_table_without_confirmed_by( + self, + ): + # #703, same pattern as code_family/note (Ruling C1): a replica + # created before confirmed_by/source existed has a 10-column + # pfs.code_element and a 6-column pfs.code_element_review — + # ensure_tables must add the columns in place. + con = duckdb.connect(":memory:") + try: + con.execute("CREATE SCHEMA IF NOT EXISTS pfs;") + con.execute( + "CREATE TABLE pfs.code_element (" + "code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, " + "detail VARCHAR, text VARCHAR, item_key VARCHAR, p_id INTEGER, " + "page INTEGER, source VARCHAR)" + ) + con.execute( + "CREATE TABLE pfs.code_element_review (" + "code VARCHAR, text VARCHAR, proposed_type VARCHAR, " + "proposed_value VARCHAR, item_key VARCHAR, p_id INTEGER)" + ) + con.execute( + "INSERT INTO pfs.code_element VALUES " + "('99490', 2015, 'activity', 'consent', '', 'Consent;', " + "'DE2VH9PD', 1245, 67716, 'fr')" + ) + con.execute( + "INSERT INTO pfs.code_element_review VALUES " + "('99490', 'Odd line;', '', '', 'DE2VH9PD', 1246)" + ) + ensure_tables(con) + el_cols = { + r[0] + for r in con.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema='pfs' AND table_name='code_element'" + ).fetchall() + } + rev_cols = { + r[0] + for r in con.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema='pfs' AND table_name='code_element_review'" + ).fetchall() + } + assert "confirmed_by" in el_cols + assert "source" in rev_cols + # The pre-existing row survives the upgrade with confirmed_by + # = NULL (mirrors the note-field precedent, C1) — read_elements + # does not crash on it. + row = read_elements(con, "99490")[0] + assert row.confirmed_by is None + # A fresh write still round-trips a real value. + write_elements( + con, + "99491", + [ + ElementRow( + "99491", + 2015, + "activity", + "consent", + "", + "Consent;", + "DE2VH9PD", + 1245, + 67716, + "fr", + "cpt:GQGTPGYV", + ) + ], + [], + ) + assert read_elements(con, "99491")[0].confirmed_by == "cpt:GQGTPGYV" + finally: + con.close() + + def test_read_elements_tolerates_a_table_with_no_confirmed_by_column(self): + # I4-style tolerance: a read-only path against an old replica must + # not call ensure_tables (that would need a write connection) — it + # must cope with the short tuple SELECT * returns. + con = duckdb.connect(":memory:") + try: + con.execute("CREATE SCHEMA IF NOT EXISTS pfs;") + con.execute( + "CREATE TABLE pfs.code_element (" + "code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, " + "detail VARCHAR, text VARCHAR, item_key VARCHAR, p_id INTEGER, " + "page INTEGER, source VARCHAR)" + ) + con.execute( + "INSERT INTO pfs.code_element VALUES " + "('99490', 2015, 'activity', 'consent', '', 'Consent;', " + "'DE2VH9PD', 1245, 67716, 'fr')" + ) + rows = read_elements(con, "99490") + assert len(rows) == 1 and rows[0].confirmed_by == "" + finally: + con.close() + class TestElements: def test_write_replaces_per_code(self, con): @@ -125,6 +227,65 @@ class TestElements: assert read_elements(con, "99490") == [] assert len(read_elements(con, "G0556")) == 1 + def test_confirmed_by_round_trips(self, con): + # #703: confirmed_by is a plain column — it round-trips like any + # other field, and doesn't change the row count. + row = ElementRow( + "99490", + 2015, + "activity", + "comprehensive-care-plan", + "", + "Comprehensive care plan;", + "DE2VH9PD", + 1245, + 67716, + "fr", + "cpt:GQGTPGYV", + ) + assert write_elements(con, "99490", [row], []) == 1 + got = read_elements(con, "99490") + assert len(got) == 1 + assert got[0].confirmed_by == "cpt:GQGTPGYV" + + def test_confirmed_by_defaults_to_empty_string(self, con): + # Ruling C1 pattern: last field, defaulted, so a caller that omits + # it (as `_el` above does) still works. + write_elements(con, "99490", [_el()], []) + assert read_elements(con, "99490")[0].confirmed_by == "" + + +class TestReviews: + def test_write_and_read_round_trip_source(self, con): + write_elements( + con, + "99490", + [], + [ + ReviewRow("99490", "Odd fr line;", "", "", "DE2VH9PD", 1246, "fr"), + ReviewRow("99490", "Odd cpt line;", "", "", "GQGTPGYV", 1247, "cpt"), + ], + ) + got = read_reviews(con, "99490") + assert [r.source for r in got] == ["fr", "cpt"] + + def test_source_defaults_to_empty_string(self, con): + write_elements( + con, "99490", [], [ReviewRow("99490", "Odd line;", "", "", "K", 9)] + ) + assert read_reviews(con, "99490")[0].source == "" + + def test_read_all_reviews_matches_per_code_reader(self, con): + write_elements( + con, "99490", [], [ReviewRow("99490", "Odd line;", "", "", "K", 9, "fr")] + ) + write_elements( + con, "G0556", [], [ReviewRow("G0556", "Other line;", "", "", "K", 3, "cpt")] + ) + out = read_all_reviews(con) + assert {r.code for r in out} == {"99490", "G0556"} + assert out == read_reviews(con, "99490") + read_reviews(con, "G0556") + class TestReadAll: def test_read_all_elements_matches_per_code_reader(self, con): diff --git a/tests/pfs/test_extract.py b/tests/pfs/test_extract.py index 64fd274..d926288 100644 --- a/tests/pfs/test_extract.py +++ b/tests/pfs/test_extract.py @@ -7,10 +7,12 @@ import sqlite3 import duckdb import pytest +from pfs.codetables import ElementRow from pfs.descriptors import DescriptorRun, Para, descriptor_runs from pfs.extract import ( Extraction, _cpt_elements, + _merge_or_confirm, extract_code, extract_run, extract_text, @@ -74,6 +76,12 @@ class TestDeterministic: assert [r.p_id for r in x.reviews] == [1199] assert x.reviews[0].proposed_value == "" + def test_review_rows_carry_the_fr_source(self): + # #703: ReviewRow.source distinguishes FR-sourced review lines + # from CPT-sourced ones. + x = extract_run(RUN) + assert x.reviews and all(r.source == "fr" for r in x.reviews) + def test_home_setting_from_element_line(self): assert "home" in _vals(extract_run(RUN), "setting") @@ -279,11 +287,12 @@ class TestExtractCodeCpt: stem_row = next(r for r in cpt_rows if r.value == "calendar-month") assert stem_row.text == "Chronic care management services" assert stem_row.p_id == 0 and stem_row.page == 0 - assert [(r.text, r.item_key, r.p_id) for r in x.reviews] == [ + assert [(r.text, r.item_key, r.p_id, r.source) for r in x.reviews] == [ ( "Something the vocabulary does not know about at all", "GQGTPGYV", 0, + "cpt", ) ] @@ -338,6 +347,10 @@ class TestExtractCodeCpt: consent_rows = [r for r in x.rows if r.value == "consent"] assert len(consent_rows) == 1 assert consent_rows[0].source == "fr" and consent_rows[0].p_id == 901 + # #703: the CPT manual's independent transcription of the same + # element cross-checks the winning FR row instead of being + # dropped outright — recorded as a source:item_key token. + assert consent_rows[0].confirmed_by == "cpt:GQGTPGYV" finally: store_con.close() @@ -359,6 +372,10 @@ class TestExtractCodeCpt: consent_rows = [r for r in x.rows if r.value == "consent"] assert len(consent_rows) == 1 assert consent_rows[0].source == "cpt" + # #703: HCPCS's independent transcription confirms the winning CPT + # row (HCPCS rows carry no item_key, so the token's second half is + # empty). + assert consent_rows[0].confirmed_by == "hcpcs:" def test_classifier_places_an_unmatched_cpt_element_line(self, store, con): # Mirrors TestClassifier's FR-side coverage, but for a required- @@ -463,3 +480,57 @@ class TestText: assert all( r.item_key == "" and r.p_id == 0 and r.source == "hcpcs" for r in x.rows ) + + +def _el(*, source, item_key="", value="consent"): + return ElementRow( + "99490", 2024, "activity", value, "", "Consent;", item_key, 0, 0, source + ) + + +class TestMergeOrConfirm: + """#703: a second source's independent transcription of the same + ``(type, value, detail)`` cross-checks the winning row instead of + being dropped — the row count doesn't grow.""" + + def test_first_sighting_is_kept_unconfirmed(self): + merged: dict = {} + _merge_or_confirm(merged, _el(source="fr")) + (row,) = merged.values() + assert row.source == "fr" and row.confirmed_by == "" + + def test_second_source_confirms_without_replacing_the_winner(self): + merged: dict = {} + _merge_or_confirm(merged, _el(source="fr", item_key="K1")) + _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV")) + (row,) = merged.values() + assert row.source == "fr" # the first source still wins + assert row.confirmed_by == "cpt:GQGTPGYV" + + def test_same_source_duplicate_does_not_confirm(self): + # Two rows from the same source (e.g. two FR rule years naming the + # same element) is a same-source duplicate, not a second, + # independent source — dropped exactly as before, uncounted. + merged: dict = {} + _merge_or_confirm(merged, _el(source="fr", item_key="K1")) + _merge_or_confirm(merged, _el(source="fr", item_key="K2")) + (row,) = merged.values() + assert row.item_key == "K1" and row.confirmed_by == "" + + def test_repeated_same_token_is_not_duplicated(self): + # Re-running the same source twice (or a code with more than one + # CPT hit for the same element) must not repeat a token. + merged: dict = {} + _merge_or_confirm(merged, _el(source="fr", item_key="K1")) + _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV")) + _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV")) + (row,) = merged.values() + assert row.confirmed_by == "cpt:GQGTPGYV" + + def test_a_third_source_appends_a_second_token(self): + merged: dict = {} + _merge_or_confirm(merged, _el(source="fr", item_key="K1")) + _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV")) + _merge_or_confirm(merged, _el(source="hcpcs")) + (row,) = merged.values() + assert row.confirmed_by == "cpt:GQGTPGYV,hcpcs:"