"""Unit tests for the CMS-1500 text-layer reader. Words are synthetic (made-up codes and numbers placed at real CMS-1500 positions), so no patient data lives in this file. """ from decimal import Decimal from cms1500_text import Word, parse_diagnosis_codes, parse_service_lines def W(text, x, y): return Word(text=text, x=x, y=y) # Slot labels as Waystar prints them in box 21 (x, y are normalized 0-1). DX_LABELS = [ W("A.", 0.047, 0.602), W("B.", 0.199, 0.603), W("C.", 0.352, 0.602), W("D.", 0.505, 0.604), W("E.", 0.047, 0.617), W("F.", 0.200, 0.618), W("G.", 0.352, 0.618), W("H.", 0.505, 0.619), W("I.", 0.045, 0.632), W("J.", 0.199, 0.632), W("K.", 0.352, 0.631), W("L.", 0.504, 0.632), W("21.", 0.046, 0.581), W("DIAGNOSIS", 0.082, 0.581), ] def test_dx_codes_are_keyed_by_slot_letter_in_slot_order(): words = DX_LABELS + [ # Printed slightly above their labels, with jittery y — the old OCR # sort by (y, x) scrambled exactly this kind of row. W("Z9999", 0.248, 0.5975), # B W("M1111", 0.095, 0.5985), # A W("R2222", 0.400, 0.5980), # C W("S81812A", 0.095, 0.6130), # E ] assert parse_diagnosis_codes(words) == [ ("A", "M1111"), ("B", "Z9999"), ("C", "R2222"), ("E", "S81812A"), ] def line_words(y, mm="01", dd="05", yy="26", pos="12", cpt="97110", mods=("GP",), ptr="AB", charge="9000", units="2", npi="1234567890", row_no=None): """One box-24 service line at Waystar's x positions. None = not printed.""" ws = [W("NPI", 0.801, y)] if row_no: ws.append(W(row_no, 0.026, y - 0.009)) # gutter number sits in the shaded row above for text, x in ((mm, 0.048), (dd, 0.083), (yy, 0.118), (pos, 0.259), (cpt, 0.348), (ptr, 0.565), (charge, 0.683), (units, 0.724), (npi, 0.883)): if text is not None: ws.append(W(text, x, y)) for i, m in enumerate(mods): ws.append(W(m, 0.424 + 0.035 * i, y)) return ws BOX_24_HEADER = [W("24.", 0.046, 0.642), W("UNITS", 0.736, 0.661)] BOX_25_ROW = [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869), W("14439", 0.395, 0.869), W("63000", 0.701, 0.869)] def test_service_line_fields_are_read_from_their_columns(): words = BOX_24_HEADER + BOX_25_ROW + line_words(0.689, mods=("GP", "CQ", "59")) [line] = parse_service_lines(words) assert line.line_no == 1 assert line.dos_from == "01/05/2026" assert line.dos_to == "" assert line.pos == "12" assert line.cpt == "97110" assert line.modifiers == ("GP", "CQ", "59") assert line.dx_pointer == "AB" assert line.charge == Decimal("90.00") assert line.units == "2" assert line.npi == "1234567890" def test_lines_are_numbered_top_to_bottom_and_ignore_gutter_and_box_26(): words = (BOX_24_HEADER + BOX_25_ROW + line_words(0.719, cpt="97140", row_no="2") + line_words(0.689, cpt="97110", row_no="1")) lines = parse_service_lines(words) assert [(l.line_no, l.cpt) for l in lines] == [(1, "97110"), (2, "97140")] assert all(l.dos_from == "01/05/2026" for l in lines) # gutter "1"/"2" never read as MM def test_a_missing_value_stays_blank_on_its_own_line(): words = (BOX_24_HEADER + line_words(0.689, units="2") + line_words(0.719, units=None) + line_words(0.750, units="1")) assert [l.units for l in parse_service_lines(words)] == ["2", "", "1"] def test_charge_without_decimal_point_has_implied_cents(): words = (BOX_24_HEADER + line_words(0.689, charge="12000") + line_words(0.719, charge="45.50")) assert [l.charge for l in parse_service_lines(words)] == [Decimal("120.00"), Decimal("45.50")] def test_hcpcs_line_with_dos_to_and_large_units(): words = BOX_24_HEADER + line_words(0.689, cpt="Q4250", mods=(), ptr="A", units="120") words += [W("01", 0.155, 0.689), W("07", 0.186, 0.689), W("26", 0.224, 0.689)] [line] = parse_service_lines(words) assert (line.cpt, line.units, line.modifiers) == ("Q4250", "120", ()) assert line.dos_to == "01/07/2026" from cms1500_text import Claim, build_claim, parse_header, to_report_row # noqa: E402 HEADER_WORDS = [ W("1a.", 0.619, 0.128), W("NUMBER", 0.723, 0.128), W("1AB2CD3EF45", 0.677, 0.145), W("3.", 0.381, 0.158), W("MM", 0.402, 0.165), W("DD", 0.439, 0.165), W("YY", 0.482, 0.165), W("07", 0.401, 0.174), W("04", 0.436, 0.174), W("1950", 0.483, 0.174), W("M", 0.524, 0.174), ] def test_header_fields_come_from_their_boxes(): h = parse_header(HEADER_WORDS + BOX_25_ROW) assert h == { "insured_id": "1AB2CD3EF45", "dob": "07/04/1950", "patient_account": "44680 14439", "total_charge": Decimal("630.00"), } def page(*line_specs, total="18000"): """A full synthetic page: header + dx + given lines + box 25/28 row.""" ws = HEADER_WORDS + DX_LABELS + [W("M1111", 0.095, 0.598), W("Z9999", 0.248, 0.598)] ws += BOX_24_HEADER + [W("25.", 0.046, 0.854), W("44680", 0.324, 0.869), W("14439", 0.395, 0.869), W(total, 0.701, 0.869)] for i, kw in enumerate(line_specs): ws += line_words(0.689 + 0.030 * i, **kw) return ws def test_clean_claim_has_no_review_notes(): claim = build_claim([page({}, {"cpt": "97140"})]) assert isinstance(claim, Claim) assert [l.cpt for l in claim.lines] == ["97110", "97140"] assert claim.issues == [] def test_line_charges_must_add_up_to_box_28(): claim = build_claim([page({}, {}, total="20000")]) assert claim.issues == ["Line charges 180.00 != box 28 total 200.00"] def test_missing_line_values_and_bad_pointer_are_flagged(): claim = build_claim([page({"units": None, "ptr": "AC"}, {"npi": None})]) assert claim.issues == [ "Line 1: no Units", "Line 1: Dx pointer C has no diagnosis code", "Line 2: no Rendering NPI", ] def test_claim_with_no_service_lines_or_no_text_is_flagged(): assert build_claim([page()]).issues == ["No service lines found", "Line charges 0.00 != box 28 total 180.00"] assert build_claim([[]]).issues == ["No text layer on PDF (scanned?) - check by hand"] def test_lines_continue_across_pages_and_totals_add_up(): claim = build_claim([page({}, {}, {}, {}, {}, {}, total="54000"), page({}, total="9000")]) assert [l.line_no for l in claim.lines] == [1, 2, 3, 4, 5, 6, 7] assert claim.total_charge == Decimal("630.00") assert claim.issues == [] def test_report_row_keeps_legacy_columns_and_alignment(): claim = build_claim([page({"units": "2"}, {"cpt": "97140", "units": None, "mods": ("GP", "59")})]) row = to_report_row(claim, claim_id="999", insurance="PAYER", patient_name="DOE, J", signature_physician="Jo Doe", fallback_account="") assert list(row)[:15] == [ "Claim ID", "Rendering Provider #", "Signature of Physician", "Patient's Account No", "Date of Birth", "Insurance Name", "Insured ID Number", "Patient's Name", "Date of Service", "Diagnosis Codes", "CPT", "POS", "Modifier", "Units", "Charges", ] assert row["CPT"] == "CPT 1: 97110; CPT 2: 97140" assert row["Units"] == "Units 1: 2; Units 2: " # blank stays on line 2 assert row["Modifier"] == "Modifier 1: GP; Modifier 2: GP 59" assert row["Diagnosis Codes"] == "M1111, Z9999 (Total: 2)" assert row["Patient's Account No"] == "44680 14439" assert row["Charges"] == "180.00" assert row["Line Charges"] == "Charge 1: 90.00; Charge 2: 90.00" assert row["Review Notes"] == "Line 2: no Units" def test_report_row_for_single_line_claim_uses_plain_values(): row = to_report_row(build_claim([page({}, total="9000")]), claim_id="1") assert (row["CPT"], row["Units"], row["Rendering Provider #"]) == ("97110", "2", "1234567890") assert row["Review Notes"] == "" def test_report_row_falls_back_to_grid_charges_when_box_28_is_blank(): claim = build_claim([page({})]) claim.total_charge = None row = to_report_row(claim, claim_id="1", fallback_charges="90") assert row["Charges"] == "90"