793 lines
34 KiB
Python
793 lines
34 KiB
Python
"""
|
|
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())
|