""" Tests for the patient rollup (build plan 02, P3). Rollup rules locked in readiness-model-brief section 4 / phase-2 design Item 3: worst gradeable line; any gradeable Plan Type Needed line caps the patient no greener than Action Needed; non-gradeable lines never touch the light; a patient with no graded gradeable lines has rollup_status None (no label invented). PHI CONTRACT: synthetic patient_ids only. """ import io import sys from datetime import date, timedelta from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) from core.dedup import CoverageLine, MergedShipment # noqa: E402 from core.readiness import LineStatus # noqa: E402 from core.rollup import rollup_patient, rollup_patients # noqa: E402 TODAY = date.today() def make_line(patient_id="PT-1", device="dexcom_g7", plan="medicare", status=None, gradeable=True): shipment = MergedShipment( patient_id=patient_id, device_type=device, shipment_date=TODAY - timedelta(days=5), total_quantity=3, components=["sensor"], order_numbers=[], hcpcs_codes=[], payer="Medicare Part B", plan_type=None if plan == "unknown" else plan, ) return CoverageLine( patient_id=patient_id, device_type=device, plan_type=plan, gradeable=gradeable, current_shipment=shipment, prior_shipments=[], line_status=status, timing_flag=None, ) def test_patient_rollup_single_line(): ru = rollup_patient([make_line(status="On Track")]) assert ru.rollup_status == LineStatus.ON_TRACK assert ru.cgm_line_count == 1 assert ru.total_line_count == 1 def test_patient_rollup_mixed_statuses(): ru = rollup_patient([ make_line(device="dexcom_g7", status="Clear to Ship"), make_line(device="freestyle_libre_2", status="At Risk"), ]) assert ru.rollup_status == LineStatus.AT_RISK def test_patient_rollup_plan_type_needed_capping(): # PTN + Clear to Ship = Action Needed (capped, not PTN) ru = rollup_patient([ make_line(device="dexcom_g7", status="Clear to Ship"), make_line(device="freestyle_libre_2", plan="unknown", status="Plan Type Needed"), ]) assert ru.rollup_status == LineStatus.ACTION_NEEDED assert ru.plan_type_needed is True # PTN + At Risk = At Risk (the worse of the two) ru2 = rollup_patient([ make_line(device="dexcom_g7", status="At Risk"), make_line(device="freestyle_libre_2", plan="unknown", status="Plan Type Needed"), ]) assert ru2.rollup_status == LineStatus.AT_RISK def test_patient_rollup_all_lines_plan_type_needed(): ru = rollup_patient([ make_line(device="dexcom_g7", plan="unknown", status="Plan Type Needed"), make_line(device="freestyle_libre_2", plan="unknown", status="Plan Type Needed"), ]) assert ru.rollup_status == LineStatus.PLAN_TYPE_NEEDED def test_patient_rollup_non_cgm_excluded(): ru = rollup_patient([ make_line(device="dexcom_g7", status="Clear to Ship"), make_line(device="omnipod_5", gradeable=False, status=None), ]) assert ru.rollup_status == LineStatus.CLEAR_TO_SHIP # pod changes nothing assert ru.cgm_line_count == 1 assert ru.total_line_count == 2 def test_patient_rollup_no_gradeable_lines(): ru = rollup_patient([make_line(device="omnipod_5", gradeable=False)]) assert ru.rollup_status is None assert ru.total_line_count == 1 assert ru.cgm_line_count == 0 def test_rollup_patients_groups_by_patient(): lines = { ("A", "dexcom_g7", "medicare"): make_line("A", status="Clear to Ship"), ("A", "freestyle_libre_2", "medicare"): make_line( "A", device="freestyle_libre_2", status="Action Needed"), ("B", "dexcom_g7", "unknown"): make_line( "B", plan="unknown", status="Plan Type Needed"), } rollups = rollup_patients(lines) assert set(rollups) == {"A", "B"} assert rollups["A"].rollup_status == LineStatus.ACTION_NEEDED assert rollups["B"].rollup_status == LineStatus.PLAN_TYPE_NEEDED # --- API level --------------------------------------------------------------- httpx = pytest.importorskip("httpx", reason="httpx required for TestClient") from fastapi.testclient import TestClient # noqa: E402 import api.main as api_main # noqa: E402 client = TestClient(api_main.app) SHIP = (TODAY - timedelta(days=5)).isoformat() VISIT = (TODAY - timedelta(days=30)).isoformat() HEADER = ( "patient_id,device_type,shipment_date,quantity,payer,plan_type," "swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file" ) def _upload(csv_text: str): return client.post( "/api/upload", files={"file": ("fixture.csv", io.BytesIO(csv_text.encode()), "text/csv")}, ) def test_upload_response_contains_patients_nested(): csv_text = ( HEADER + f"\nPT-N-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + f"On File,{VISIT},Yes,,Yes" + f"\nPT-N-1,Libre 2,{SHIP},1,Medicare Part B,medicare," + "On File,,Yes,,Yes" ) resp = _upload(csv_text) assert resp.status_code == 200 body = resp.json() patients = body["patients"] assert len(patients) == 1 p = patients[0] assert p["patient_id"] == "PT-N-1" assert p["rollup_status"] == "Action Needed" # worst line (Libre 2 no visit) assert len(p["lines"]) == 2 by_dev = {l["device_type"]: l for l in p["lines"]} assert by_dev["dexcom_g7"]["line_status"] == "Clear to Ship" assert by_dev["freestyle_libre_2"]["line_status"] == "Action Needed" items = {i["doc_type"]: i for i in by_dev["freestyle_libre_2"]["doc_items"]} assert items["visit"]["satisfied"] is False assert items["visit"]["rule_id"] == "R101" # Worst line renders first assert p["lines"][0]["device_type"] == "freestyle_libre_2" def test_patient_stats_count_patients_not_rows(): csv_text = ( HEADER + f"\nPT-S-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + f"On File,{VISIT},Yes,,Yes" + f"\nPT-S-1,Libre 2,{SHIP},1,Medicare Part B,medicare," + f"On File,{VISIT},Yes,,Yes" + f"\nPT-S-2,Dexcom G6,{SHIP},3,Medicaid - GA,," + f"On File,{VISIT},,,Yes" ) resp = _upload(csv_text) assert resp.status_code == 200 stats = resp.json()["patient_stats"] assert stats["patients_total"] == 2 # 2-device patient counts once assert stats["Clear to Ship"] == 1 assert stats["Plan Type Needed"] == 1 def test_flat_records_untouched_by_nested_payload(): csv_text = ( HEADER + f"\nPT-F-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + f"On File,{VISIT},Yes,,Yes" ) resp = _upload(csv_text) body = resp.json() assert body["total"] == 1 assert len(body["records"]) == 1 # flat path byte-identical (P6 switches UI) assert body["records"][0]["readiness_status"] == "Clear to Ship" def test_non_gradeable_only_patient_has_null_rollup(): csv_text = ( HEADER + f"\nPT-P-1,Omnipod 5,{SHIP},1,Highmark BCBS,commercial," + f"On File,{VISIT},Yes,Approved,Yes" ) resp = _upload(csv_text) body = resp.json() p = body["patients"][0] assert p["rollup_status"] is None assert p["cgm_line_count"] == 0 assert p["lines"][0]["gradeable"] is False assert p["lines"][0]["line_status"] is None assert body["patient_stats"]["ungraded"] == 1 def test_rollup_patient_empty_raises(): with pytest.raises(ValueError): rollup_patient([]) def test_line_timing_flag_populated(): csv_text = ( HEADER + f"\nPT-T-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + f"On File,{VISIT},Yes,,Yes" ) resp = _upload(csv_text) line = resp.json()["patients"][0]["lines"][0] assert line["timing_flag"] == "RESUPPLY_READY"