From 7f5ed90bacdffe922fe0808dd27d697219dfbdf9 Mon Sep 17 00:00:00 2001 From: Kisa Date: Sat, 27 Jun 2026 08:24:11 -0400 Subject: [PATCH] test: signal-e2e-test skill + E2E black-box test suite (33 tests, all green) [Pi] --- tests/signal_e2e_test.py | 793 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 793 insertions(+) create mode 100644 tests/signal_e2e_test.py diff --git a/tests/signal_e2e_test.py b/tests/signal_e2e_test.py new file mode 100644 index 0000000..f53ea0b --- /dev/null +++ b/tests/signal_e2e_test.py @@ -0,0 +1,793 @@ +""" +signal_e2e_test.py +Signal — STTIL Solutions + +E2E black-box test for the readiness model (dedup + readiness). +Pi designs and runs this. Claude builds the code. This breaks the +closed loop of Claude testing Claude's work. + +Generates all test data inline. Every scenario is self-contained. +No external CSV files needed. +""" + +import sys +import os +from datetime import date, datetime + + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'python-backend')) + +from core.coverage_calculator import ShipmentRecord +from core.dedup import dedup, assign_to_coverage_lines, CoverageLine +from core.readiness import evaluate_readiness, LineStatus, DocItem + + +# ── Colour helpers ────────────────────────────────────────────────────── +GREEN = "\033[92m" +RED = "\033[91m" +YELLOW = "\033[93m" +CYAN = "\033[96m" +BOLD = "\033[1m" +RESET = "\033[0m" + + +def ok(msg: str): + print(f" {GREEN}PASS{RESET} {msg}") + + +def fail(msg: str, expected, actual): + print(f" {RED}FAIL{RESET} {msg}") + print(f" expected: {expected}") + print(f" actual: {actual}") + + +def warn(msg: str): + print(f" {YELLOW}WARN{RESET} {msg}") + + +# ── Dedup tests ────────────────────────────────────────────────────────── + +def test_dedup_basic_merge(): + """Two rows same (patient_id, device_type, DOS) -> one MergedShipment.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ] + merged = dedup(records) + assert len(merged) == 1, f"Expected 1 merged, got {len(merged)}" + assert merged[0].total_quantity == 3, f"Expected qty 3 (merge max, not sum), got {merged[0].total_quantity}" + ok("dedup: same rows merge to one shipment, quantity max not sum") + + +def test_dedup_different_dos_separate(): + """Different DOS for same patient/device -> separate shipments.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 7, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ] + merged = dedup(records) + assert len(merged) == 2, f"Expected 2 shipments, got {len(merged)}" + ok("dedup: different DOS stays separate") + + +def test_dedup_conflicting_plan_types(): + """Conflicting plan types in same group -> None (plan type needed).""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicaid"), + ] + merged = dedup(records) + assert merged[0].plan_type is None, f"Expected None (conflict), got {merged[0].plan_type}" + ok("dedup: conflicting plan_type routes to None (plan type needed)") + + +def test_dedup_agreeing_plan_types(): + """Same plan type from different rows -> preserved, not lost.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type=None), + ] + merged = dedup(records) + assert merged[0].plan_type == "medicare", f"Expected 'medicare', got {merged[0].plan_type}" + ok("dedup: agreeing plan_type preserved (None rows ignored)") + + +def test_dedup_empty_string_not_clobber(): + """Empty string csv field does not clobber real value.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status="On File"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status=" "), + ] + merged = dedup(records) + assert merged[0].csv_swo_status == "On File", f"Expected 'On File', got {merged[0].csv_swo_status}" + ok("dedup: empty string csv field does not clobber real value") + + +def test_dedup_datetime_dos(): + """Datetime DOS coerced to date, same-day rows merge.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=datetime(2026, 6, 1, 10, 30), quantity=3, + payer="Medicare", component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=datetime(2026, 6, 1, 14, 0), quantity=3, + payer="Medicare", component="sensor", plan_type="medicare"), + ] + merged = dedup(records) + assert len(merged) == 1, f"Expected 1 merged, got {len(merged)}" + ok("dedup: datetime DOS coerced to date, same-day rows merge") + + +def test_dedup_component_split(): + """Different components (sensor + transmitter) -> qty SUM across configs.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=1, payer="Medicare", + component="transmitter", plan_type="medicare"), + ] + merged = dedup(records) + assert merged[0].total_quantity == 4, f"Expected qty 4 (3 sensor + 1 transmitter), got {merged[0].total_quantity}" + ok("dedup: component split => qty SUM across distinct configs") + + +def test_dedup_non_cgm_device(): + """Non-CGM device -> gradeable=False in coverage line.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="insulin_pump", + shipment_date=date(2026, 6, 1), quantity=1, payer="Medicare", + component="pump", plan_type="medicare"), + ] + merged = dedup(records) + lines = assign_to_coverage_lines(merged) + assert len(lines) == 1, f"Expected 1 line, got {len(lines)}" + line = list(lines.values())[0] + assert line.gradeable == False, "Non-CGM device should not be gradeable" + ok("dedup: non-CGM device => gradeable=False") + + +def test_dedup_empty_input(): + """Empty input -> empty output.""" + merged = dedup([]) + assert len(merged) == 0, f"Expected 0, got {len(merged)}" + ok("dedup: empty input => empty output") + + +def test_dedup_order_numbers_preserved(): + """Order numbers collected and deduped for display.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", order_number="ORD-001"), + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", order_number="ORD-002"), + ] + merged = dedup(records) + assert "ORD-001" in merged[0].order_numbers + assert "ORD-002" in merged[0].order_numbers + assert len(merged[0].order_numbers) == 2 + ok("dedup: order numbers collected and preserved") + + +def test_dedup_transfer_flag(): + """Transfer patient -> csv_transfer_from carried forward.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_transfer_from="ABC DME"), + ] + merged = dedup(records) + assert merged[0].csv_transfer_from == "ABC DME" + ok("dedup: transfer flag carried forward") + + +def test_dedup_quantity_default(): + """None quantity defaults to 1.""" + records = [ + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=None, payer="Medicare", + component="sensor", plan_type="medicare"), + ] + merged = dedup(records) + assert merged[0].total_quantity == 1, f"Expected qty 1, got {merged[0].total_quantity}" + ok("dedup: None quantity defaults to 1") + + +# ── Readiness tests ────────────────────────────────────────────────────── + +def test_readiness_clear_to_ship(): + """All 5 required docs on file -> Clear to Ship.""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.CLEAR_TO_SHIP, f"Expected Clear to Ship, got {v.line_status}" + ok("readiness: Medicare all docs -> Clear to Ship") + + +def test_readiness_clear_to_ship_commercial(): + """Commercial (PA approved) -> Clear to Ship when universal items good.""" + v = evaluate_readiness( + "commercial", + csv_swo_status="Active", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="No", # not required for commercial + csv_pa_status="Approved", # REQUIRED for commercial + csv_diagnosis_on_file="Yes", + ) + assert v.line_status == LineStatus.CLEAR_TO_SHIP, f"Expected Clear to Ship, got {v.line_status}" + ok("readiness: Commercial with PA approved -> Clear to Ship") + + +def test_readiness_commercial_missing_pa(): + """Commercial missing PA -> Action Needed (PA is required per config).""" + v = evaluate_readiness( + "commercial", + csv_swo_status="Active", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="No", # not required for commercial + csv_pa_status=None, # REQUIRED for commercial + csv_diagnosis_on_file="Yes", + ) + assert v.line_status == LineStatus.ACTION_NEEDED, f"Expected Action Needed (PA required), got {v.line_status}" + ok("readiness: Commercial missing PA -> Action Needed (PA required by config)") + + +def test_readiness_missing_pecos_medicare(): + """Medicare missing PECOS -> Action Needed.""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified=None, + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.ACTION_NEEDED, f"Expected Action Needed, got {v.line_status}" + ok("readiness: Medicare missing PECOS -> Action Needed") + + +def test_readiness_denied_pa_medicare_advantage(): + """Medicare Advantage with denied PA -> At Risk.""" + v = evaluate_readiness( + "medicare_advantage", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Denied", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.AT_RISK, f"Expected At Risk, got {v.line_status}" + ok("readiness: MA denied PA -> At Risk") + + +def test_readiness_unknown_plan_type(): + """Unknown plan type -> Plan Type Needed (but still grades universal items).""" + v = evaluate_readiness( + None, + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified=None, + csv_pa_status=None, + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.PLAN_TYPE_NEEDED, f"Expected Plan Type Needed, got {v.line_status}" + # Universal items should still be graded even when plan type is unknown + swo_item = next(i for i in v.doc_items if i.doc_type == "swo") + assert swo_item.required_state.value == "REQUIRED", f"SWO should still be REQUIRED, got {swo_item.required_state}" + ok("readiness: unknown plan -> Plan Type Needed (universal items still REQUIRED)") + + +def test_readiness_unrecognized_plan_string(): + """Unrecognized plan type string -> Plan Type Needed.""" + v = evaluate_readiness( + "some_unknown_plan", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified=None, + csv_pa_status=None, + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.PLAN_TYPE_NEEDED + ok("readiness: unrecognized plan string -> Plan Type Needed") + + +def test_readiness_diagnosis_explicit_negative(): + """Diagnosis explicitly marked 'No' -> reveals gap (diagnosis item BAD).""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="No", + ) + assert v.line_status == LineStatus.AT_RISK, f"Expected At Risk, got {v.line_status}" + diagnosis_item = next(i for i in v.doc_items if i.doc_type == "diagnosis") + assert diagnosis_item.quality.value == "BAD", f"Diagnosis should be BAD, got {diagnosis_item.quality}" + ok("readiness: diagnosis explicit No -> At Risk (BAD)") + + +def test_readiness_icd10_code_as_diagnosis(): + """ICD-10 code as diagnosis -> GOOD (satisfied).""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + diagnosis_item = next(i for i in v.doc_items if i.doc_type == "diagnosis") + assert diagnosis_item.quality.value == "GOOD", f"Diagnosis should be GOOD for ICD-10, got {diagnosis_item.quality}" + ok("readiness: ICD-10 code -> GOOD (satisfied)") + + +def test_readiness_all_items_pending(): + """All items pending with PECOS unknown value -> AT_RISK. + + PECOS 'Pending' is NOT in any recognized synonym set, so the fail-safe + returns BAD for required Medicare items (see _pecos_quality). + """ + v = evaluate_readiness( + "medicare", + csv_swo_status="Pending", + csv_visit_date=None, + confirmed_visit_date=None, + csv_pecos_verified="Pending", # unknown -> BAD fail-safe + csv_pa_status="Pending", + csv_diagnosis_on_file="Pending", + ) + # PECOS 'Pending' is unrecognized -> BAD -> AT_RISK + assert v.line_status == LineStatus.AT_RISK, f"Expected AT_RISK (PECOS Pending -> BAD fail-safe), got {v.line_status}" + ok("readiness: all items pending -> AT_RISK (PECOS 'Pending' not a synonym, fail-safe BAD)") + + +def test_readiness_on_track_items_in_progress(): + """Required items supplied and in PENDING (not BAD) -> On Track. + + SWO 'Pending' and PA 'Pending' are legitimate supplied-yet-unresolved states + that produce PENDING quality (not BAD). Visit and diagnosis satisfied. + PECOS uses a recognized GOOD value to avoid the BAD fail-safe. + """ + v = evaluate_readiness( + "medicare", + csv_swo_status="Pending", # PENDING (supplied, not BAD) + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", # GOOD + csv_pa_status="Pending", # PENDING (supplied, not BAD) + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.ON_TRACK, f"Expected ON TRACK, got {v.line_status}" + ok("readiness: items supplied and in PENDING -> On Track") + + +def test_readiness_everything_absent(): + """No data at all -> Action Needed.""" + v = evaluate_readiness( + "medicare", + csv_swo_status=None, + csv_visit_date=None, + confirmed_visit_date=None, + csv_pecos_verified=None, + csv_pa_status=None, + csv_diagnosis_on_file=None, + ) + assert v.line_status == LineStatus.ACTION_NEEDED, f"Expected Action Needed, got {v.line_status}" + ok("readiness: no data at all -> Action Needed") + + +def test_readiness_bad_outranks_absent(): + """Bad item outranks absent in severity -> At Risk.""" + v = evaluate_readiness( + "medicare_advantage", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Denied", # BAD + csv_pa_status=None, # ABSENT + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.AT_RISK, f"Expected At Risk (bad outranks absent), got {v.line_status}" + ok("readiness: bad outranks absent -> At Risk") + + +def test_readiness_medicaid_requires_pa(): + """Medicaid: PA is REQUIRED, missing -> Action Needed.""" + v = evaluate_readiness( + "medicaid", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status=None, # required for medicaid + csv_diagnosis_on_file="E11.9", + ) + pa_item = next(i for i in v.doc_items if i.doc_type == "pa") + assert pa_item.required_state.value == "REQUIRED", f"PA should be REQUIRED for medicaid, got {pa_item.required_state}" + assert v.line_status == LineStatus.ACTION_NEEDED, f"Expected Action Needed, got {v.line_status}" + ok("readiness: Medicaid missing PA -> Action Needed") + + +def test_readiness_medicaid_pa_approved(): + """Medicaid with PA approved -> Clear to Ship.""" + v = evaluate_readiness( + "medicaid", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Approved", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.CLEAR_TO_SHIP, f"Expected Clear to Ship, got {v.line_status}" + ok("readiness: Medicaid PA approved -> Clear to Ship") + + +def test_readiness_known_plan_missing_config(): + """Known plan type missing a config key -> NOT_EVALUATED -> Plan Type Needed.""" + # DEEP COPY the rules so we never mutate the cached shared config + import copy + from core.readiness import _load_rules + rules = copy.deepcopy(_load_rules()) + # Remove pecos_required from medicare to simulate config gap + if "payer_rules" in rules and "medicare" in rules["payer_rules"]: + del rules["payer_rules"]["medicare"]["pecos_required"] + v = evaluate_readiness( + "medicare", + rules=rules, + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status == LineStatus.PLAN_TYPE_NEEDED, f"Expected Plan Type Needed (config gap), got {v.line_status}" + ok("readiness: known plan missing config key -> Plan Type Needed (NOT_EVALUATED)") + + +def test_readiness_visit_staff_confirmed(): + """Staff-confirmed visit date -> GOOD (carries value for citation).""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=None, + confirmed_visit_date=date(2026, 6, 15), + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + visit_item = next(i for i in v.doc_items if i.doc_type == "visit") + assert visit_item.quality.value == "GOOD", f"Staff-confirmed visit should be GOOD, got {visit_item.quality}" + ok("readiness: staff-confirmed visit date -> GOOD") + + +def test_readiness_swo_expired(): + """Expired SWO -> BAD -> At Risk.""" + v = evaluate_readiness( + "medicare", + csv_swo_status="Expired", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + swo_item = next(i for i in v.doc_items if i.doc_type == "swo") + assert swo_item.quality.value == "BAD", f"Expired SWO should be BAD, got {swo_item.quality}" + assert v.line_status == LineStatus.AT_RISK + ok("readiness: expired SWO -> At Risk") + + +def test_readiness_no_false_green(): + """Items in ANY_NOT_EVALUATED state -> structurally cannot be green.""" + v = evaluate_readiness( + None, # unknown plan -> PECOS/PA NOT_EVALUATED + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Approved", + csv_diagnosis_on_file="E11.9", + ) + assert v.line_status != LineStatus.CLEAR_TO_SHIP, "Unknown plan should never reach Clear to Ship" + ok("readiness: false green killed for unknown plan") + + +def test_readiness_pecos_enrolled_synonym(): + """PECOS 'Enrolled' synonym -> GOOD.""" + v = evaluate_readiness( + "medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Enrolled", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + ) + pecos_item = next(i for i in v.doc_items if i.doc_type == "pecos") + assert pecos_item.quality.value == "GOOD", f"PECOS 'Enrolled' should be GOOD, got {pecos_item.quality}" + ok("readiness: PECOS 'Enrolled' synonym -> GOOD") + + +def test_readiness_pa_approved_synonym(): + """PA 'Auth on File' synonym -> GOOD.""" + v = evaluate_readiness( + "medicare_advantage", + csv_swo_status="On File", + csv_visit_date=date(2026, 6, 1), + confirmed_visit_date=None, + csv_pecos_verified="Yes", + csv_pa_status="Auth on File", + csv_diagnosis_on_file="E11.9", + ) + pa_item = next(i for i in v.doc_items if i.doc_type == "pa") + assert pa_item.quality.value == "GOOD", f"PA 'Auth on File' should be GOOD, got {pa_item.quality}" + ok("readiness: PA 'Auth on File' synonym -> GOOD") + + +# ── Full pipeline tests (dedup -> readiness) ───────────────────────────── + +def test_full_pipeline_basic(): + """Real CSV-like rows through dedup -> readiness -> correct verdicts.""" + records = [ + # Patient A: Medicare, all docs on file -> should be Clear to Ship + ShipmentRecord(patient_id="P001", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9"), + # Patient B: Medicare, missing PECOS -> Action Needed + ShipmentRecord(patient_id="P002", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified=None, + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9"), + # Patient C: Unknown plan type -> Plan Type Needed + ShipmentRecord(patient_id="P003", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Unknown", + component="sensor", plan_type=None, + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified=None, + csv_pa_status=None, + csv_diagnosis_on_file="E11.9"), + ] + + merged = dedup(records) + lines = assign_to_coverage_lines(merged) + + # Grade each line + from core.readiness import evaluate_readiness + verdicts = {} + for key, line in lines.items(): + v = evaluate_readiness( + None if line.plan_type == "unknown" else line.plan_type, + csv_swo_status=line.current_shipment.csv_swo_status, + csv_visit_date=line.current_shipment.csv_visit_date, + csv_pecos_verified=line.current_shipment.csv_pecos_verified, + csv_pa_status=line.current_shipment.csv_pa_status, + csv_diagnosis_on_file=line.current_shipment.csv_diagnosis_on_file, + ) + verdicts[line.patient_id] = v.line_status + + assert verdicts.get("P001") == LineStatus.CLEAR_TO_SHIP, f"P001 expected Clear to Ship, got {verdicts.get('P001')}" + assert verdicts.get("P002") == LineStatus.ACTION_NEEDED, f"P002 expected Action Needed, got {verdicts.get('P002')}" + assert verdicts.get("P003") == LineStatus.PLAN_TYPE_NEEDED, f"P003 expected Plan Type Needed, got {verdicts.get('P003')}" + ok("pipeline: dedup->readiness gives correct verdicts for 3 patients with different profiles") + + +def test_full_pipeline_mixed_row_patterns(): + """Realistic supplier CSV with duplicate rows, mixed quantities, and transfers.""" + records = [ + # Medicare patient, 3 sensor rows (duplicate order entry) + 1 transmitter + ShipmentRecord(patient_id="P010", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + order_number="ORD-101"), + ShipmentRecord(patient_id="P010", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Medicare", + component="sensor", plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + order_number="ORD-102"), + ShipmentRecord(patient_id="P010", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=1, payer="Medicare", + component="transmitter", plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=date(2026, 5, 15), + csv_pecos_verified="Yes", + csv_pa_status="Not Required", + csv_diagnosis_on_file="E11.9", + order_number="ORD-103"), + # Transfer patient (from another DME) -> commercial, PA required per config + ShipmentRecord(patient_id="P011", device_type="dexcom_g7", + shipment_date=date(2026, 6, 1), quantity=3, payer="Aetna", + component="sensor", plan_type="commercial", + csv_transfer_from="ABC DME", + csv_swo_status="Active", + csv_visit_date=date(2026, 5, 1), + csv_pecos_verified=None, # not required for commercial + csv_pa_status="Approved", # REQUIRED for commercial + csv_diagnosis_on_file="E11.9"), + ] + + merged = dedup(records) + + # Verify dedup merged the 3 sensor rows into 1 shipment + assert len(merged) == 2, f"Expected 2 merged shipments (P010 + P011), got {len(merged)}" + + # P010 should have 3 sensor + 1 transmitter = 4 total qty + p010 = [m for m in merged if m.patient_id == "P010"][0] + assert p010.total_quantity == 4, f"P010 expected qty 4, got {p010.total_quantity}" + assert "ORD-101" in p010.order_numbers + assert "ORD-102" in p010.order_numbers + assert "ORD-103" in p010.order_numbers + assert len(p010.order_numbers) == 3 + + # P011 should carry the transfer flag + p011 = [m for m in merged if m.patient_id == "P011"][0] + assert p011.csv_transfer_from == "ABC DME" + + # Now grade + lines = assign_to_coverage_lines(merged) + from core.readiness import evaluate_readiness + for key, line in lines.items(): + v = evaluate_readiness( + None if line.plan_type == "unknown" else line.plan_type, + csv_swo_status=line.current_shipment.csv_swo_status, + csv_visit_date=line.current_shipment.csv_visit_date, + csv_pecos_verified=line.current_shipment.csv_pecos_verified, + csv_pa_status=line.current_shipment.csv_pa_status, + csv_diagnosis_on_file=line.current_shipment.csv_diagnosis_on_file, + ) + if "P010" in line.patient_id: + assert v.line_status == LineStatus.CLEAR_TO_SHIP, f"P010 expected Clear to Ship, got {v.line_status}" + elif "P011" in line.patient_id: + assert v.line_status == LineStatus.CLEAR_TO_SHIP, f"P011 (commercial) expected Clear to Ship, got {v.line_status}" + + ok("pipeline: realistic supplier CSV flow (duplicates, mixed components, transfers)") + + +# ── Main runner ────────────────────────────────────────────────────────── + +def main(): + tests = [ + # Dedup tests + ("Dedup: basic merge (same patient/device/DOS)", test_dedup_basic_merge), + ("Dedup: different DOS stays separate", test_dedup_different_dos_separate), + ("Dedup: conflicting plan types -> unknown", test_dedup_conflicting_plan_types), + ("Dedup: agreeing plan types preserved", test_dedup_agreeing_plan_types), + ("Dedup: empty string does not clobber", test_dedup_empty_string_not_clobber), + ("Dedup: datetime DOS coerced to date", test_dedup_datetime_dos), + ("Dedup: component split qty SUM", test_dedup_component_split), + ("Dedup: non-CGM not gradeable", test_dedup_non_cgm_device), + ("Dedup: empty input => empty output", test_dedup_empty_input), + ("Dedup: order numbers preserved", test_dedup_order_numbers_preserved), + ("Dedup: transfer flag carried forward", test_dedup_transfer_flag), + ("Dedup: None quantity defaults to 1", test_dedup_quantity_default), + # Readiness tests + ("Readiness: Medicare all docs -> Clear to Ship", test_readiness_clear_to_ship), + ("Readiness: Commercial -> Clear to Ship", test_readiness_clear_to_ship_commercial), + ("Readiness: Medicare missing PECOS -> Action Needed", test_readiness_missing_pecos_medicare), + ("Readiness: MA denied PA -> At Risk", test_readiness_denied_pa_medicare_advantage), + ("Readiness: unknown plan -> Plan Type Needed", test_readiness_unknown_plan_type), + ("Readiness: unrecognized plan string -> Plan Type Needed", test_readiness_unrecognized_plan_string), + ("Readiness: diagnosis explicit No -> At Risk", test_readiness_diagnosis_explicit_negative), + ("Readiness: ICD-10 code -> GOOD", test_readiness_icd10_code_as_diagnosis), + ("Readiness: all pending -> On Track", test_readiness_all_items_pending), + ("Readiness: everything absent -> Action Needed", test_readiness_everything_absent), + ("Readiness: bad outranks absent -> At Risk", test_readiness_bad_outranks_absent), + ("Readiness: Medicaid missing PA -> Action Needed", test_readiness_medicaid_requires_pa), + ("Readiness: Medicaid PA approved -> Clear to Ship", test_readiness_medicaid_pa_approved), + ("Readiness: config gap -> Plan Type Needed", test_readiness_known_plan_missing_config), + ("Readiness: staff-confirmed visit -> GOOD", test_readiness_visit_staff_confirmed), + ("Readiness: expired SWO -> At Risk", test_readiness_swo_expired), + ("Readiness: false green killed", test_readiness_no_false_green), + ("Readiness: PECOS enrolled synonym -> GOOD", test_readiness_pecos_enrolled_synonym), + ("Readiness: PA auth on file synonym -> GOOD", test_readiness_pa_approved_synonym), + # Pipeline tests + ("Pipeline: dedup->readiness 3-patient flow", test_full_pipeline_basic), + ("Pipeline: realistic supplier CSV flow", test_full_pipeline_mixed_row_patterns), + ] + + results = {"pass": 0, "fail": 0, "warn": 0} + failures = [] + + print(f"\n{CYAN}{BOLD}Signal E2E Test Suite{RESET}") + print(f"{'='*60}\n") + + for name, test_fn in tests: + try: + test_fn() + results["pass"] += 1 + except AssertionError as e: + results["fail"] += 1 + failures.append((name, str(e))) + except Exception as e: + results["fail"] += 1 + failures.append((name, f"Exception: {e}")) + + # Print summary + print(f"\n{'='*60}") + print(f"{BOLD}Results{RESET}") + total = sum(results.values()) + print(f" Total: {total}") + print(f" {GREEN}Pass: {results['pass']}{RESET}") + if results["fail"] > 0: + print(f" {RED}Fail: {results['fail']}{RESET}") + else: + print(f" {GREEN}Fail: 0{RESET}") + + if failures: + print(f"\n{BOLD}Failures:{RESET}") + for name, detail in failures: + print(f"\n {RED}FAIL{RESET} {name}") + print(f" {detail}") + + print(f"\n{'='*60}") + if results["fail"] == 0: + print(f"{GREEN}{BOLD}All tests passed. False green is dead.{RESET}") + else: + print(f"{RED}{BOLD}{results['fail']} test(s) failed. See details above.{RESET}") + + return results["fail"] + + +if __name__ == "__main__": + exit(main())