bcb1a10fe6
- New CoverageFlag enum: SUPPLY_LAPSED, VISIT_REQUIRED, TRANSFER_PENDING, RENEWAL_CRITICAL/ELEVATED/SOON, RESUPPLY_READY, ACTIVE - Doc state machine: 5-item payer-dependent status per patient (SWO, Visit, PECOS, PA, Diagnosis) with cascade chain - Confirm Visit endpoint: staff enters prescriber-confirmed date, persisted in Supabase confirmed_visits table, survives all future CSV imports - Supabase migration: 001_add_confirmed_visits.sql (run manually in SQL editor) - Frontend: Badge rebuilt for 8 flags, DocStatusBar 5-dot display, ConfirmVisitModal, expandable WorklistTable rows - Legal: LOI, NDA, BAA drafts at pitch/legal/ for Nixon Law Group review - Compliance docs: privacy policy, incident response, data handling - CSV generator: market_data.json + PA/NJ generator scripts - 15/15 tests passing Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
import sys
|
||
from pathlib import Path
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
|
||
|
||
from datetime import date, timedelta
|
||
from core.coverage_calculator import ShipmentRecord, calculate_coverage, CoverageFlag
|
||
|
||
TODAY = date.today()
|
||
|
||
def make_record(**kwargs):
|
||
defaults = dict(
|
||
patient_id="PT001",
|
||
device_type="dexcom_g7",
|
||
shipment_date=TODAY - timedelta(days=10),
|
||
quantity=1,
|
||
payer="medicare",
|
||
component="sensor",
|
||
)
|
||
defaults.update(kwargs)
|
||
return ShipmentRecord(**defaults)
|
||
|
||
def test_shipment_record_accepts_csv_visit_date():
|
||
r = make_record(csv_visit_date=TODAY - timedelta(days=30))
|
||
assert r.csv_visit_date == TODAY - timedelta(days=30)
|
||
|
||
def test_shipment_record_accepts_doc_fields():
|
||
r = make_record(
|
||
csv_swo_status="On File",
|
||
csv_pecos_verified="Yes",
|
||
csv_pa_status="Not Required",
|
||
csv_diagnosis_on_file="Yes",
|
||
)
|
||
assert r.csv_swo_status == "On File"
|
||
assert r.csv_pecos_verified == "Yes"
|
||
|
||
def test_shipment_record_doc_fields_default_none():
|
||
r = make_record()
|
||
assert r.csv_visit_date is None
|
||
assert r.csv_swo_status is None
|
||
assert r.csv_transfer_from is None
|
||
|
||
|
||
def test_supply_lapsed_flag_when_coverage_ended():
|
||
# Shipment 200 days ago, 1 sensor (10 wear days) — coverage ended long ago
|
||
r = make_record(shipment_date=TODAY - timedelta(days=200), quantity=1)
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.SUPPLY_LAPSED
|
||
|
||
def test_renewal_critical_at_45_days():
|
||
# Last confirmed visit was 135 days ago — next visit due in 45 days
|
||
r = make_record(
|
||
shipment_date=TODAY - timedelta(days=10),
|
||
quantity=1,
|
||
csv_visit_date=TODAY - timedelta(days=135),
|
||
)
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.RENEWAL_CRITICAL
|
||
|
||
def test_renewal_elevated_at_60_days():
|
||
r = make_record(
|
||
shipment_date=TODAY - timedelta(days=10),
|
||
quantity=1,
|
||
csv_visit_date=TODAY - timedelta(days=120),
|
||
)
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.RENEWAL_ELEVATED
|
||
|
||
def test_renewal_soon_at_90_days():
|
||
r = make_record(
|
||
shipment_date=TODAY - timedelta(days=10),
|
||
quantity=1,
|
||
csv_visit_date=TODAY - timedelta(days=90),
|
||
)
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.RENEWAL_SOON
|
||
|
||
def test_active_when_visit_far_away():
|
||
# 6 sensors × 10 days = 60-day supply, shipped 10 days ago → 50 days left (outside refill window)
|
||
# visit 10 days ago → 170 days until next due → not in any RENEWAL tier
|
||
r = make_record(
|
||
shipment_date=TODAY - timedelta(days=10),
|
||
quantity=6,
|
||
csv_visit_date=TODAY - timedelta(days=10),
|
||
)
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.ACTIVE
|
||
|
||
def test_transfer_pending_flag():
|
||
r = make_record(csv_transfer_from="Prior Supplier LLC")
|
||
result = calculate_coverage(r)
|
||
assert result.flag == CoverageFlag.TRANSFER_PENDING
|
||
|
||
def test_priority_sort_order():
|
||
lapsed = make_record(shipment_date=TODAY - timedelta(days=200))
|
||
critical = make_record(patient_id="PT002", shipment_date=TODAY - timedelta(days=10),
|
||
csv_visit_date=TODAY - timedelta(days=135))
|
||
active = make_record(patient_id="PT003", shipment_date=TODAY - timedelta(days=10),
|
||
quantity=6, csv_visit_date=TODAY - timedelta(days=10))
|
||
from core.coverage_calculator import calculate_batch
|
||
results = calculate_batch([active, critical, lapsed])
|
||
assert results[0].flag == CoverageFlag.SUPPLY_LAPSED
|
||
assert results[1].flag == CoverageFlag.RENEWAL_CRITICAL
|
||
assert results[2].flag == CoverageFlag.ACTIVE
|