Signal/tests/test_rollup.py

"""
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"