sttil/Signal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Signal/tests/test_worklist_readiness.py
Kisa 6181de74ed feat: patient-nested worklist — the true-green display (plan 02, P6) 
The worklist now shows one row per patient carrying the rollup verdict,
expanding into device coverage lines with doc checklists and rule citations.
Green is earned only from the readiness verdict. Kisa-approved design
(mockup v2, decisions A/B/C/D/E locked 2026-07-07): purple Plan Type Needed
badge + Map-the-plan-type CTA; ◇ column-not-mapped chips; not-processed
strip with download; Data-gaps filter (honest labels, no sixth status);
extended legend; per-line work-queue export (backend lines[] support,
audit-logged). Tabs count patients. SWO/PA cycling is device-scoped and
returns the re-rolled patient through the P4 echo contract; Confirm Visit
fans out across the patient's echoed lines. Legacy table remains at
?legacy=1 (its DOCS_REQUIRED stopgap intentionally kept while it exists —
literal deviation from the plan's remove-in-same-change line, logged).
Independent audit: SHIP; its MEDIUM (line-identity merge collision) and two
LOWs fixed in-commit. 171 backend tests green; frontend builds clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 06:37:04 -04:00

767 lines
28 KiB
Python
Raw Blame History

"""
Tests for the readiness-model wiring (build plan 01).
Engine behavior is covered by test_dedup.py / test_readiness.py; these tests
target the SEAMS the wiring adds: the orchestration module, the /api/upload
augmentation, the /api/confirm-visit recompute, and back-compat invariants.
PHI CONTRACT: fixtures use synthetic patient_ids only. No names, DOBs, SSNs.
"""
import hashlib
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.coverage_calculator import ShipmentRecord, calculate_batch # noqa: E402
from core.worklist_readiness import ( # noqa: E402
build_readiness_index,
merged_to_record,
readiness_by_line,
readiness_for_record,
readiness_stats,
resolve_group,
)
TODAY = date.today()
def _record(**kwargs) -> ShipmentRecord:
defaults = dict(
patient_id="PT-1001",
device_type="dexcom_g7",
shipment_date=TODAY - timedelta(days=5),
quantity=3,
payer="Medicare Part B",
component="sensor",
)
defaults.update(kwargs)
return ShipmentRecord(**defaults)
def _hash(pid: str) -> str:
return hashlib.sha256(pid.encode()).hexdigest()
# --- orchestration module (unit level) --------------------------------------
def test_readiness_by_line_full_evidence_is_clear_to_ship():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([rec])
key = ("PT-1001", "dexcom_g7", "medicare")
assert key in verdicts
assert verdicts[key].line_status.value == "Clear to Ship"
def test_readiness_by_line_skips_non_gradeable_lines():
rec = _record(device_type="omnipod_5", plan_type="medicare")
verdicts = readiness_by_line([rec])
assert verdicts == {}
def test_no_plan_type_reads_plan_type_needed():
rec = _record(
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([rec])
key = ("PT-1001", "dexcom_g7", "unknown")
assert verdicts[key].line_status.value == "Plan Type Needed"
by_type = {i.doc_type: i for i in verdicts[key].doc_items}
assert by_type["pecos"].required_state.value == "NOT_EVALUATED"
assert by_type["pa"].required_state.value == "NOT_EVALUATED"
def test_same_dos_different_order_numbers_is_one_line_verdict():
a = _record(order_number="ORD-1", plan_type="medicare")
b = _record(order_number="ORD-2", plan_type="medicare")
verdicts = readiness_by_line([a, b])
assert len(verdicts) == 1
def test_verdict_graded_from_latest_dos():
older = _record(
shipment_date=TODAY - timedelta(days=40),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=45),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
# Latest DOS is missing PECOS -> the line verdict must reflect THIS row.
newer = _record(
shipment_date=TODAY - timedelta(days=5),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=45),
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([older, newer])
assert len(verdicts) == 1
verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")]
assert verdict.line_status.value == "Action Needed"
def test_conflicting_plan_types_route_to_unknown_line():
a = _record(order_number="ORD-1", plan_type="medicare")
b = _record(order_number="ORD-2", plan_type="medicaid")
index = build_readiness_index([a, b])
assert set(index.by_line) == {("PT-1001", "dexcom_g7", "unknown")}
verdict, line_key, merged = resolve_group(
index, "PT-1001", "dexcom_g7", a.shipment_date
)
assert verdict is not None
assert verdict.line_status.value == "Plan Type Needed"
assert line_key == ("PT-1001", "dexcom_g7", "unknown")
assert merged is not None
def test_mixed_plan_population_resolves_to_merged_line():
"""Review finding (HIGH family): a group where only SOME rows carry
plan_type merges to the typed line; row resolution must follow the MERGE,
not any single row's own plan_type."""
blank = _record(
order_number="ORD-1",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
typed = _record(
order_number="ORD-2",
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
index = build_readiness_index([blank, typed])
assert set(index.by_line) == {("PT-1001", "dexcom_g7", "medicare")}
verdict, line_key, _ = resolve_group(
index, "PT-1001", "dexcom_g7", blank.shipment_date
)
assert verdict is not None, "blank-plan row must inherit its merged line's verdict"
assert verdict.line_status.value == "Clear to Ship"
assert line_key == ("PT-1001", "dexcom_g7", "medicare")
def test_conflicted_dos_never_borrows_another_lines_verdict():
"""Review finding (HIGH): a conflicted newer DOS must read its own
unknown-line verdict, never the older clean line's green."""
old_clean = _record(
shipment_date=TODAY - timedelta(days=45),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=50),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
new_a = _record(order_number="ORD-1", plan_type="medicare")
new_b = _record(order_number="ORD-2", plan_type="medicaid")
index = build_readiness_index([old_clean, new_a, new_b])
assert set(index.by_line) == {
("PT-1001", "dexcom_g7", "medicare"),
("PT-1001", "dexcom_g7", "unknown"),
}
old_verdict, _, _ = resolve_group(
index, "PT-1001", "dexcom_g7", old_clean.shipment_date
)
new_verdict, _, _ = resolve_group(
index, "PT-1001", "dexcom_g7", new_a.shipment_date
)
assert old_verdict.line_status.value == "Clear to Ship"
assert new_verdict.line_status.value == "Plan Type Needed"
def test_merged_to_record_matches_dedup_field_policy():
"""Review finding (MEDIUM): fallback record fields must come from the
dedup merge (latest-non-null, collected order numbers), so the legacy
doc_state cannot contradict the readiness verdict."""
first = _record(order_number="ORD-1", csv_pecos_verified="Yes")
second = _record(order_number="ORD-2", plan_type="medicare")
index = build_readiness_index([first, second])
_, _, merged = resolve_group(
index, "PT-1001", "dexcom_g7", first.shipment_date
)
rec = merged_to_record(merged)
assert rec.csv_pecos_verified == "Yes" # latest-non-null keeps it
assert rec.order_number == "ORD-1, ORD-2" # collected, not arbitrary
assert rec.plan_type == "medicare"
def test_confirmed_visit_counts_as_positive_evidence():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
confirmed = {_hash("PT-1001"): TODAY - timedelta(days=20)}
verdicts = readiness_by_line([rec], confirmed_visits=confirmed)
verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")]
assert verdict.line_status.value == "Clear to Ship"
visit = next(i for i in verdict.doc_items if i.doc_type == "visit")
assert visit.satisfied
def test_readiness_stats_counts_lines():
good = _record(
patient_id="PT-A",
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
unknown = _record(patient_id="PT-B")
stats = readiness_stats(readiness_by_line([good, unknown]))
assert stats["Clear to Ship"] == 1
assert stats["Plan Type Needed"] == 1
assert stats["lines_total"] == 2
def test_no_raw_patient_id_in_logs(caplog):
# Conflicting plan types force the dedup warning path.
a = _record(patient_id="RAW-SECRET-ID-42", order_number="1", plan_type="medicare")
b = _record(patient_id="RAW-SECRET-ID-42", order_number="2", plan_type="medicaid")
with caplog.at_level("WARNING"):
readiness_by_line([a, b])
assert caplog.text # the warning fired
assert "RAW-SECRET-ID-42" not in caplog.text
def test_readiness_for_record_single_path():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdict = readiness_for_record(
rec, confirmed_visit_date=TODAY - timedelta(days=10)
)
assert verdict.line_status.value == "Clear to Ship"
# --- API level (FastAPI TestClient) ------------------------------------------
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)
def _upload(csv_text: str):
return client.post(
"/api/upload",
files={"file": ("fixture.csv", io.BytesIO(csv_text.encode()), "text/csv")},
)
# Dates engineered so the legacy timing flag is RESUPPLY_READY (coverage ends
# within the 30-day window) and no renewal flag fires (visit recent).
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 test_upload_false_green_blocked_at_api_layer():
"""Acceptance case 1: legacy label green-ish, readiness verdict honest."""
csv_text = (
HEADER
+ f"\nPT-FG-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["flag"] == "RESUPPLY_READY"
assert rec["status_label"] == "Clear to Ship" # legacy timing label
assert rec["readiness_status"] == "Action Needed" # PECOS absent blocks green
def test_upload_without_plan_type_reads_plan_type_needed():
"""Acceptance case 2: today's default for every record — honest, never guessed."""
header = (
"patient_id,device_type,shipment_date,quantity,payer,"
"swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file"
)
csv_text = (
header
+ f"\nPT-PTN-1,Dexcom G7,{SHIP},3,Medicare Part B,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
graded = [r for r in body["records"] if r["readiness_status"] is not None]
assert graded, "expected at least one graded record"
assert all(r["readiness_status"] == "Plan Type Needed" for r in graded)
assert body["readiness_stats"]["Plan Type Needed"] == body[
"readiness_stats"
]["lines_total"]
def test_upload_clear_to_ship_reachable_end_to_end():
"""Acceptance case 3: full positive evidence + mapped plan type = true green."""
csv_text = (
HEADER
+ f"\nPT-GREEN-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["plan_type"] == "medicare"
assert rec["readiness_status"] == "Clear to Ship"
items = {i["doc_type"]: i for i in rec["readiness_items"]}
assert items["pa"]["required_state"] == "NOT_REQUIRED" # Medicare FFS
assert all(i["satisfied"] for i in items.values())
def test_upload_with_order_numbers_still_resolves_records():
"""Regression for the record_lookup miss: CoverageResult has no
order_number, so exact-key lookups miss for order-numbered CSVs; the
fallback must recover doc_state AND readiness fields."""
header = HEADER + ",order_number"
csv_text = (
header
+ f"\nPT-ORD-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-778899"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["order_number"] == "ORD-778899"
assert rec["doc_state"] is not None
assert rec["readiness_status"] == "Clear to Ship"
def test_upload_backcompat_legacy_fields_unchanged():
"""Acceptance case 6: the wiring is additive — every legacy field matches
an independent reconstruction from the engines for an order-less CSV
(where the exact record lookup must hit and byte-identity must hold)."""
csv_text = (
HEADER
+ f"\nPT-BC-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
+ f"\nPT-BC-2,Libre 3,{(TODAY - timedelta(days=90)).isoformat()},1,"
+ "Aetna,commercial,Pending,,,Requested,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
from api.normalizer import normalize_csv
from core.doc_state_machine import compute_doc_state
records, _, _ = normalize_csv(csv_text)
by_pid = {r.patient_id: r for r in records}
expected = calculate_batch(records, as_of=TODAY, confirmed_visits={})
assert len(body["records"]) == len(expected)
for got, want in zip(body["records"], expected):
rec = by_pid[want.patient_id]
assert got["patient_id"] == want.patient_id
assert got["flag"] == want.flag.value
assert got["priority_score"] == want.priority_score
assert got["status_label"] == api_main.FLAG_LABELS.get(
want.flag.value, want.flag.value
)
assert got["payer"] == want.payer
assert got["days_until_coverage_end"] == want.days_until_coverage_end
assert got["coverage_end_date"] == want.coverage_end_date.isoformat()
assert got["rule_version"] == want.rule_version
assert got["action"] == api_main.FLAG_ACTIONS.get(
want.flag.value, "Review"
)
# Record-echo fields resolve through the exact lookup (order-less CSV)
assert got["quantity"] == rec.quantity
assert got["order_number"] == rec.order_number
assert got["csv_swo_status"] == rec.csv_swo_status
assert got["csv_pecos_verified"] == rec.csv_pecos_verified
assert got["csv_pa_status"] == rec.csv_pa_status
assert got["csv_diagnosis_on_file"] == rec.csv_diagnosis_on_file
# doc_state reconstructed independently from the same inputs
doc = compute_doc_state(
payer_type=api_main._normalize_payer_type(want.payer),
csv_swo_status=rec.csv_swo_status,
csv_visit_date=rec.csv_visit_date,
confirmed_visit_date=None,
csv_pecos_verified=rec.csv_pecos_verified,
csv_pa_status=rec.csv_pa_status,
csv_diagnosis_on_file=rec.csv_diagnosis_on_file,
is_transfer=bool(rec.csv_transfer_from),
)
assert got["doc_state"]["visit"] == doc.visit
assert got["doc_state"]["pecos"] == doc.pecos
assert got["doc_state"]["diagnosis"] == doc.diagnosis
assert got["cascade"] == doc.cascade
# Full legacy stats dict reconstructed independently from the flags
flags = [w.flag.value for w in expected]
assert body["stats"] == {
"total": len(expected),
"supply_lapsed": flags.count("SUPPLY_LAPSED"),
"visit_required": flags.count("VISIT_REQUIRED"),
"transfer_pending": flags.count("TRANSFER_PENDING"),
"renewal_critical": flags.count("RENEWAL_CRITICAL"),
"renewal_elevated": flags.count("RENEWAL_ELEVATED"),
"renewal_soon": flags.count("RENEWAL_SOON"),
"resupply_ready": flags.count("RESUPPLY_READY"),
"active": flags.count("ACTIVE"),
"no_recent_shipment": flags.count("NO_RECENT_SHIPMENT"),
"prescriber_action": (
flags.count("SUPPLY_LAPSED")
+ flags.count("VISIT_REQUIRED")
+ flags.count("RENEWAL_CRITICAL")
+ flags.count("RENEWAL_ELEVATED")
),
}
def test_upload_mixed_plan_group_both_rows_carry_merged_verdict():
"""Review finding (HIGH family), API level: duplicate order-numbered rows
where only one carries plan_type — both output rows must carry the merged
line's verdict, and stats must reconcile with the rows."""
header = HEADER + ",order_number"
csv_text = (
header
+ f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,,"
+ f"On File,{VISIT},Yes,,Yes,ORD-1"
+ f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-2"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
assert len(body["records"]) == 2
for rec in body["records"]:
assert rec["readiness_status"] == "Clear to Ship"
assert rec["plan_type"] == "medicare"
assert rec["order_number"] == "ORD-1, ORD-2"
assert body["readiness_stats"] == {"Clear to Ship": 1, "lines_total": 1}
def test_upload_conflicted_dos_reads_own_line_not_older_green():
"""Review finding (HIGH), API level: the false-green repro. A conflicted
newer DOS must read Plan Type Needed even when an older clean DOS on the
same patient+device graded Clear to Ship."""
header = HEADER + ",order_number"
old_ship = (TODAY - timedelta(days=45)).isoformat()
old_visit = (TODAY - timedelta(days=50)).isoformat()
csv_text = (
header
+ f"\nPT-CONF-1,Dexcom G7,{old_ship},3,Medicare Part B,medicare,"
+ f"On File,{old_visit},Yes,,Yes,ORD-OLD"
+ f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-N1"
+ f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicaid - GA,medicaid,"
+ f"On File,{VISIT},Yes,,Yes,ORD-N2"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
by_ship = {}
for rec in body["records"]:
by_ship.setdefault(rec["last_shipment_date"], []).append(rec)
for rec in by_ship[old_ship]:
assert rec["readiness_status"] == "Clear to Ship"
for rec in by_ship[SHIP]:
assert rec["readiness_status"] == "Plan Type Needed"
assert body["readiness_stats"] == {
"Clear to Ship": 1,
"Plan Type Needed": 1,
"lines_total": 2,
}
def test_upload_skipped_record_leaves_no_phantom_line_in_stats():
"""Review finding (LOW): a record calculate_batch skips (G7 transmitter
has no wear-day rule) must not leave a phantom line in readiness_stats."""
header = HEADER + ",component"
csv_text = (
header
+ f"\nPT-OK-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,sensor"
+ f"\nPT-TX-1,Dexcom G7,{SHIP},1,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,transmitter"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
pids = {r["patient_id"] for r in body["records"]}
assert pids == {"PT-OK-1"} # transmitter row skipped by the engine
assert body["readiness_stats"]["lines_total"] == 1
# Failed-data honesty: the dropped row is visible in the skipped list,
# never silently vanished (Kisa ruling 2026-07-07).
assert body["skipped"] >= 1
assert any(
"PT-TX-1" in r and "not scored" in r for r in body["skipped_reasons"]
)
def test_upload_non_gradeable_rows_carry_no_verdict():
csv_text = (
HEADER
+ f"\nPT-POD-1,Omnipod 5,{SHIP},1,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["readiness_status"] is None
assert rec["readiness_items"] is None
def test_confirm_visit_recompute_flips_to_clear(monkeypatch):
"""Acceptance case 5: confirming the visit satisfies the last open
required item in the same response."""
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
monkeypatch.setattr(
api_main, "upsert_confirmed_visit", lambda *a, **kw: True
)
# Upload path first: everything good except visit evidence.
csv_text = (
HEADER
+ f"\nPT-CV-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ "On File,,Yes,,Yes"
)
up = _upload(csv_text)
assert up.status_code == 200
assert up.json()["records"][0]["readiness_status"] == "Action Needed"
resp = client.post(
"/api/confirm-visit",
json={
"patient_id": "PT-CV-1",
"confirmed_date": (TODAY - timedelta(days=15)).isoformat(),
"shipment_date": SHIP,
"payer": "Medicare Part B",
"device_type": "dexcom_g7",
"quantity": 3,
"csv_swo_status": "On File",
"csv_pecos_verified": "Yes",
"csv_diagnosis_on_file": "Yes",
"plan_type": "medicare",
},
)
assert resp.status_code == 200
rec = resp.json()
assert rec["readiness_status"] == "Clear to Ship"
visit = next(i for i in rec["readiness_items"] if i["doc_type"] == "visit")
assert visit["satisfied"]
def test_confirm_visit_normalizes_plan_type(monkeypatch):
"""Review finding (LOW): both paths must return the same normalized
plan_type representation."""
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
monkeypatch.setattr(
api_main, "upsert_confirmed_visit", lambda *a, **kw: True
)
resp = client.post(
"/api/confirm-visit",
json={
"patient_id": "PT-NORM-1",
"confirmed_date": (TODAY - timedelta(days=15)).isoformat(),
"shipment_date": SHIP,
"payer": "Medicare Part B",
"device_type": "dexcom_g7",
"quantity": 3,
"csv_swo_status": "On File",
"csv_pecos_verified": "Yes",
"csv_diagnosis_on_file": "Yes",
"plan_type": " Medicare ",
},
)
assert resp.status_code == 200
rec = resp.json()
assert rec["plan_type"] == "medicare"
assert rec["readiness_status"] == "Clear to Ship"
def test_upload_readiness_items_carry_rule_ids():
"""P2 contract W2: the serialized doc_items cite their rules."""
csv_text = (
HEADER
+ f"\nPT-RID-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
items = {i["doc_type"]: i for i in resp.json()["records"][0]["readiness_items"]}
assert items["pecos"]["rule_id"] == "R004"
assert items["pa"]["rule_id"] == "R003"
assert items["swo"]["rule_id"] == "R100"
# ---------------------------------------------------------------------------
# P5: plan-type enforcement in the live path (plan 02)
# ---------------------------------------------------------------------------
def test_plan_type_value_normalization():
"""Synonyms canonicalize; the same string in a PAYER column changes
nothing about readiness."""
from api.normalizer import _normalize_plan_type
for raw, want in [
("Medicare Part B", "medicare"), ("FFS", "medicare"),
("Original Medicare", "medicare"), ("MA", "medicare_advantage"),
("Part C", "medicare_advantage"), ("state medicaid", "medicaid"),
("Private", "commercial"), ("Employer", "commercial"),
("", None), (" ", None), ("Tricare Select", None),
]:
assert _normalize_plan_type(raw) == want, raw
def test_payer_name_never_sets_plan_type():
"""A payer of 'Medicare Part B' with NO plan_type column yields
Plan Type Needed — never a guess from the payer name."""
header = (
"patient_id,device_type,shipment_date,quantity,payer,"
"swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file"
)
csv_text = (
header
+ f"\nPT-NG-1,Dexcom G7,{SHIP},3,Medicare Part B,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
rec = resp.json()["records"][0]
assert rec["plan_type"] is None
assert rec["readiness_status"] == "Plan Type Needed"
def test_plan_type_synonym_flows_end_to_end():
csv_text = (
HEADER
+ f"\nPT-SYN-1,Dexcom G7,{SHIP},3,Medicare Part B,Medicare Part B,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
rec = resp.json()["records"][0]
assert rec["plan_type"] == "medicare" # value map canonicalized it
assert rec["readiness_status"] == "Clear to Ship"
def test_unknown_plan_type_value_yields_plan_type_needed():
csv_text = (
HEADER
+ f"\nPT-UNK-1,Dexcom G7,{SHIP},3,Tricare,Tricare Select,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
rec = resp.json()["records"][0]
assert rec["plan_type"] is None
assert rec["readiness_status"] == "Plan Type Needed"
def test_default_config_never_reaches_readiness():
"""'default' is not a known plan type; it must grade Plan Type Needed,
never through the default payer_rules entry."""
from core.readiness import evaluate_readiness
v = evaluate_readiness(
"default",
csv_swo_status="On File",
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
assert v.line_status.value == "Plan Type Needed"
def test_deprecation_warning_logged_once_per_batch(caplog):
header = (
"patient_id,device_type,shipment_date,quantity,payer,"
"swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file"
)
csv_text = (
header
+ f"\nPT-DW-1,Dexcom G7,{SHIP},3,Medicare Part B,On File,{VISIT},Yes,,Yes"
+ f"\nPT-DW-2,Libre 3,{SHIP},1,Aetna,On File,{VISIT},,,Yes"
)
with caplog.at_level("WARNING"):
resp = _upload(csv_text)
assert resp.status_code == 200
hits = [r for r in caplog.records if "plan_type not provided" in r.getMessage()]
assert len(hits) == 1 # once per batch, not per record
assert "2/2 records" in hits[0].getMessage()
def test_unrecognized_plan_value_warning_hashes_patient(caplog):
csv_text = (
HEADER
+ f"\nPT-RAW-SECRET,Dexcom G7,{SHIP},3,Tricare,Tricare Select,"
+ f"On File,{VISIT},Yes,,Yes"
)
with caplog.at_level("WARNING"):
_upload(csv_text)
hits = [r for r in caplog.records if "Unrecognized plan_type" in r.getMessage()]
assert len(hits) == 1
assert "PT-RAW-SECRET" not in hits[0].getMessage()
def test_export_per_line_rows():
"""P6 (Kisa-locked): the work-queue export writes one row per coverage
line when lines are provided; legacy record export unchanged otherwise."""
resp = client.post(
"/api/export",
json={
"lines": [
{
"patient_id": "PT-EXP-1",
"device_display": "Dexcom G7",
"payer": "Medicare Part B",
"plan_type": "medicare",
"rollup_status": "Action Needed",
"line_status": "Clear to Ship",
"timing": "In resupply window",
"order_numbers": "ORD-1, ORD-2",
"hcpcs_codes": "A4239",
"total_quantity": 3,
"shipment_date": "2026-07-02",
"swo": "On File",
"visit": "2026-06-14",
"pecos": "Enrolled",
"pa": "Not required",
"diagnosis": "E11.65",
}
]
},
)
assert resp.status_code == 200
body = resp.text
lines = body.strip().splitlines()
assert lines[0].startswith("Patient ID,Device,Payer,Plan Type")
assert len(lines) == 2
assert '"ORD-1, ORD-2"' in lines[1] # joined order numbers stay one cell
assert "Clear to Ship" in lines[1]
def test_export_legacy_records_shape_unchanged():
resp = client.post("/api/export", json={"records": [], "batch_id": None})
assert resp.status_code == 200
assert resp.text.strip().splitlines()[0].startswith("Order Number,HCPCS,Patient ID")
Reference in a new issue View git blame Copy permalink