a53728e424
Ends the override leak: an SWO/PA save now applies to one device's line, not
every device the patient has. doc_status gains device_type ('' = legacy
patient-scoped row, applies until re-saved); the old unique key is dropped by
introspected name and replaced with the 4-col scope key. Staff overrides now
FEED THE VERDICT (device-scoped, staff wins over the CSV cell) instead of
display-only; visit overrides keep their patient-scoped fan-out via
confirmed_visits. PUT /api/doc-status and /api/confirm-visit accept an
optional echo of the patient's lines and return {lines, rollup_status}
recomputed through the same engine (stateless, no batch rebuild). Frontend
minimal fix: doc-status state and writes keyed patient:device. Independent
audit: SHIP (migration reproduced empirically on both old-key shapes; guard
scoped per its nit). 162 tests green.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""
|
|
Tests for device-keyed staff overrides (build plan 02, P4).
|
|
|
|
Locked semantics (phase-2 design Item 4 / brief section 6):
|
|
- visit overrides are patient-scoped and fan out to ALL the patient's lines
|
|
(they ride the confirmed_visits path by construction);
|
|
- swo/pa overrides apply only to lines whose device_type matches;
|
|
- device_type '' marks a legacy patient-scoped row and matches all devices;
|
|
- staff knowledge wins over the CSV cell;
|
|
- after a write, the recompute returns {lines, rollup} from the same engine.
|
|
|
|
PHI CONTRACT: synthetic patient_ids only; overrides carry hashes.
|
|
"""
|
|
|
|
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 # noqa: E402
|
|
from core.overrides import ( # noqa: E402
|
|
Override,
|
|
OverrideKey,
|
|
apply_overrides,
|
|
overrides_from_saved,
|
|
)
|
|
from core.worklist_readiness import build_readiness_index # noqa: E402
|
|
|
|
TODAY = date.today()
|
|
|
|
|
|
def _hash(pid: str) -> str:
|
|
return hashlib.sha256(pid.encode()).hexdigest()
|
|
|
|
|
|
def _record(**kwargs) -> ShipmentRecord:
|
|
defaults = dict(
|
|
patient_id="PT-OV-1",
|
|
device_type="dexcom_g7",
|
|
shipment_date=TODAY - timedelta(days=5),
|
|
quantity=3,
|
|
payer="Medicare Part B",
|
|
component="sensor",
|
|
plan_type="medicare",
|
|
csv_visit_date=TODAY - timedelta(days=30),
|
|
csv_pecos_verified="Yes",
|
|
csv_diagnosis_on_file="Yes",
|
|
)
|
|
defaults.update(kwargs)
|
|
return ShipmentRecord(**defaults)
|
|
|
|
|
|
def _swo_override(pid="PT-OV-1", device="dexcom_g7", status="on_file"):
|
|
return Override(
|
|
key=OverrideKey(
|
|
org_id="org", patient_hash=_hash(pid), device_type=device,
|
|
doc_type="swo",
|
|
),
|
|
status=status,
|
|
)
|
|
|
|
|
|
def test_override_swo_applies_only_to_one_device():
|
|
# Two devices, both missing SWO -> Action Needed. Override device A only.
|
|
g7 = _record(device_type="dexcom_g7")
|
|
libre = _record(device_type="freestyle_libre_2")
|
|
index = build_readiness_index(
|
|
[g7, libre], overrides=[_swo_override(device="dexcom_g7")]
|
|
)
|
|
assert index.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Clear to Ship"
|
|
assert index.by_line[("PT-OV-1", "freestyle_libre_2", "medicare")].line_status.value == "Action Needed"
|
|
|
|
|
|
def test_legacy_blank_device_row_applies_patient_wide():
|
|
g7 = _record(device_type="dexcom_g7")
|
|
libre = _record(device_type="freestyle_libre_2")
|
|
index = build_readiness_index(
|
|
[g7, libre], overrides=[_swo_override(device="")]
|
|
)
|
|
for key, verdict in index.by_line.items():
|
|
assert verdict.line_status.value == "Clear to Ship", key
|
|
|
|
|
|
def test_device_scoped_row_wins_over_legacy_row():
|
|
g7 = _record(device_type="dexcom_g7")
|
|
index = build_readiness_index(
|
|
[g7],
|
|
overrides=[
|
|
_swo_override(device="", status="on_file"),
|
|
_swo_override(device="dexcom_g7", status="pending"),
|
|
],
|
|
)
|
|
verdict = index.by_line[("PT-OV-1", "dexcom_g7", "medicare")]
|
|
swo = next(i for i in verdict.doc_items if i.doc_type == "swo")
|
|
assert swo.quality.value == "PENDING" # device-scoped Pending won
|
|
|
|
|
|
def test_override_triggers_recompute_to_better_status():
|
|
# SWO missing is the last open required item; the override flips the line.
|
|
rec = _record()
|
|
base = build_readiness_index([rec])
|
|
assert base.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Action Needed"
|
|
fixed = build_readiness_index([rec], overrides=[_swo_override()])
|
|
assert fixed.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Clear to Ship"
|
|
|
|
|
|
def test_override_visit_applies_to_all_patient_lines():
|
|
# Visit rides confirmed_visits: patient-scoped fan-out by construction.
|
|
g7 = _record(device_type="dexcom_g7", csv_visit_date=None,
|
|
csv_swo_status="On File")
|
|
libre = _record(device_type="freestyle_libre_2", csv_visit_date=None,
|
|
csv_swo_status="On File")
|
|
confirmed = {_hash("PT-OV-1"): TODAY - timedelta(days=10)}
|
|
index = build_readiness_index([g7, libre], confirmed_visits=confirmed)
|
|
for key, verdict in index.by_line.items():
|
|
visit = next(i for i in verdict.doc_items if i.doc_type == "visit")
|
|
assert visit.satisfied, key
|
|
|
|
|
|
def test_overrides_from_saved_shape():
|
|
saved = {
|
|
_hash("PT-OV-1"): {
|
|
"": {"swo": {"status": "on_file"}},
|
|
"dexcom_g7": {"pa": {"status": "approved"}},
|
|
}
|
|
}
|
|
overrides = overrides_from_saved(saved)
|
|
assert len(overrides) == 2
|
|
kinds = {(o.key.device_type, o.key.doc_type, o.status) for o in overrides}
|
|
assert kinds == {("", "swo", "on_file"), ("dexcom_g7", "pa", "approved")}
|
|
|
|
|
|
def test_apply_overrides_returns_touched_patients():
|
|
g7 = _record()
|
|
other = _record(patient_id="PT-OV-2")
|
|
index_lines = build_readiness_index([g7, other]).lines
|
|
touched = apply_overrides(index_lines, [_swo_override()])
|
|
assert touched == {"PT-OV-1"}
|
|
|
|
|
|
# --- 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)
|
|
|
|
|
|
def test_doc_status_put_accepts_device_and_returns_line_and_rollup(monkeypatch):
|
|
saved_writes = {}
|
|
|
|
def fake_upsert(org_id, patient_hash, doc_type, status,
|
|
status_date=None, expiry_date=None, device_type=""):
|
|
saved_writes[(patient_hash, doc_type, device_type)] = status
|
|
return True
|
|
|
|
def fake_load(org_id):
|
|
# Echo the persisted write back, as a fresh load would.
|
|
out = {}
|
|
for (ph, dt, dev), status in saved_writes.items():
|
|
out.setdefault(ph, {}).setdefault(dev, {})[dt] = {"status": status}
|
|
return out
|
|
|
|
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
|
|
monkeypatch.setattr(api_main, "upsert_doc_status", fake_upsert)
|
|
monkeypatch.setattr(api_main, "load_doc_statuses_for_org", fake_load)
|
|
monkeypatch.setattr(
|
|
api_main, "load_confirmed_visits_for_org", lambda org: {}
|
|
)
|
|
|
|
resp = client.put(
|
|
"/api/doc-status",
|
|
json={
|
|
"patient_id": "PT-API-1",
|
|
"doc_type": "swo",
|
|
"status": "on_file",
|
|
"device_type": "dexcom_g7",
|
|
"lines": [
|
|
{
|
|
"device_type": "dexcom_g7",
|
|
"plan_type": "medicare",
|
|
"csv_visit_date": (TODAY - timedelta(days=20)).isoformat(),
|
|
"csv_pecos_verified": "Yes",
|
|
"csv_diagnosis_on_file": "Yes",
|
|
},
|
|
{
|
|
"device_type": "freestyle_libre_2",
|
|
"plan_type": "medicare",
|
|
"csv_visit_date": (TODAY - timedelta(days=20)).isoformat(),
|
|
"csv_pecos_verified": "Yes",
|
|
"csv_diagnosis_on_file": "Yes",
|
|
},
|
|
],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["device_type"] == "dexcom_g7"
|
|
by_dev = {l["device_type"]: l for l in body["lines"]}
|
|
# The override satisfied G7's SWO; the Libre line is untouched (no leak).
|
|
assert by_dev["dexcom_g7"]["line_status"] == "Clear to Ship"
|
|
assert by_dev["freestyle_libre_2"]["line_status"] == "Action Needed"
|
|
assert body["rollup_status"] == "Action Needed" # worst line
|
|
|
|
|
|
def test_doc_status_put_without_lines_keeps_legacy_shape(monkeypatch):
|
|
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
|
|
monkeypatch.setattr(
|
|
api_main, "upsert_doc_status", lambda *a, **kw: True
|
|
)
|
|
resp = client.put(
|
|
"/api/doc-status",
|
|
json={"patient_id": "PT-API-2", "doc_type": "pa", "status": "approved"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["display"] == "Approved"
|
|
assert "lines" not in body
|
|
|
|
|
|
def test_confirm_visit_with_lines_returns_rerolled_patient(monkeypatch):
|
|
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
|
|
monkeypatch.setattr(
|
|
api_main, "upsert_confirmed_visit", lambda *a, **kw: True
|
|
)
|
|
monkeypatch.setattr(
|
|
api_main, "load_doc_statuses_for_org", lambda org: {}
|
|
)
|
|
ship = (TODAY - timedelta(days=5)).isoformat()
|
|
resp = client.post(
|
|
"/api/confirm-visit",
|
|
json={
|
|
"patient_id": "PT-API-3",
|
|
"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",
|
|
"lines": [
|
|
{
|
|
"device_type": "dexcom_g7",
|
|
"plan_type": "medicare",
|
|
"csv_swo_status": "On File",
|
|
"csv_pecos_verified": "Yes",
|
|
"csv_diagnosis_on_file": "Yes",
|
|
},
|
|
{
|
|
"device_type": "freestyle_libre_2",
|
|
"plan_type": "medicare",
|
|
"csv_swo_status": "On File",
|
|
"csv_pecos_verified": "Yes",
|
|
"csv_diagnosis_on_file": "Yes",
|
|
},
|
|
],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
# Legacy record fields still present
|
|
assert body["readiness_status"] == "Clear to Ship"
|
|
# The confirmed visit fanned out to BOTH echoed lines
|
|
assert all(l["line_status"] == "Clear to Ship" for l in body["lines"])
|
|
assert body["rollup_status"] == "Clear to Ship"
|
|
|
|
|
|
def test_upload_grades_with_saved_device_scoped_override(monkeypatch):
|
|
"""An SWO override saved for device A feeds the verdict for A's line only."""
|
|
pid_hash = hashlib.sha256("PT-UP-1".encode()).hexdigest()
|
|
monkeypatch.setattr(
|
|
api_main,
|
|
"load_doc_statuses_for_org",
|
|
lambda org: {
|
|
pid_hash: {"dexcom_g7": {"swo": {"status": "on_file"}}}
|
|
},
|
|
)
|
|
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
|
|
monkeypatch.setattr(
|
|
api_main, "load_confirmed_visits_for_org", lambda org: {}
|
|
)
|
|
ship = (TODAY - timedelta(days=5)).isoformat()
|
|
visit = (TODAY - timedelta(days=30)).isoformat()
|
|
header = (
|
|
"patient_id,device_type,shipment_date,quantity,payer,plan_type,"
|
|
"visit_date,pecos_verified,diagnosis_on_file"
|
|
)
|
|
csv_text = (
|
|
header
|
|
+ f"\nPT-UP-1,Dexcom G7,{ship},3,Medicare Part B,medicare,{visit},Yes,Yes"
|
|
+ f"\nPT-UP-1,Libre 2,{ship},1,Medicare Part B,medicare,{visit},Yes,Yes"
|
|
)
|
|
resp = client.post(
|
|
"/api/upload",
|
|
files={"file": ("f.csv", io.BytesIO(csv_text.encode()), "text/csv")},
|
|
)
|
|
assert resp.status_code == 200
|
|
recs = {r["device_type"]: r for r in resp.json()["records"]}
|
|
assert recs["dexcom_g7"]["readiness_status"] == "Clear to Ship"
|
|
assert recs["freestyle_libre_2"]["readiness_status"] == "Action Needed"
|
|
# Display follows the same device scope
|
|
assert recs["dexcom_g7"]["doc_state"]["swo"] == "On File"
|