d24340c47a
core/rules.py format-tolerant readers + stable registry (R001-R004 plan-scoped, R100-R102 universal per LCD L33822). payer_rules.json wrapped in the same commit as both engine readers (coupling hazard closed). Every DocItem now cites the rule that decided it; NOT_EVALUATED cites nothing. Independent audit: SHIP (registry-drift test added per its L2 finding). 138 tests green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
112 lines
4.1 KiB
Python
112 lines
4.1 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
|
||
|
||
|
||
def test_wrapped_config_drives_timing_rules():
|
||
"""P2: with the wrapped citation format on disk, the timing calculator
|
||
still resolves visit_renewal_days correctly (R001 = 180 days)."""
|
||
visit = TODAY - timedelta(days=30)
|
||
r = make_record(csv_visit_date=visit)
|
||
result = calculate_coverage(r)
|
||
assert result.next_visit_due_date == visit + timedelta(days=180)
|