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>
324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""
|
|
Tests for the readiness verdict engine — the "kill the false green" logic.
|
|
|
|
Source spec: docs/readiness-model-brief-2026-06-23.md section 4 ("The verdict rule").
|
|
|
|
Two axes, never conflated into one string:
|
|
required_state: does this plan need this item? (from plan_type + rules)
|
|
input_state: did the client supply data? (SUPPLIED | ABSENT)
|
|
|
|
Per DocItem:
|
|
NOT_REQUIRED + any input -> satisfied (green-eligible)
|
|
REQUIRED + SUPPLIED + good -> satisfied
|
|
REQUIRED + SUPPLIED + bad -> gap
|
|
REQUIRED + ABSENT -> NOT_EVALUATED, blocks green, never satisfied
|
|
|
|
A line is "Clear to Ship" ONLY if EVERY required item is satisfied. Green is
|
|
earned by positive evidence, never granted by the absence of contradiction.
|
|
plan_type unknown -> "Plan Type Needed", structurally cannot be green.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
|
|
|
|
from datetime import date
|
|
|
|
from core.readiness import (
|
|
evaluate_readiness,
|
|
RequiredState,
|
|
InputState,
|
|
LineStatus,
|
|
)
|
|
|
|
VISIT = date(2026, 5, 1)
|
|
|
|
|
|
def items_by_type(verdict):
|
|
return {i.doc_type: i for i in verdict.doc_items}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The four-rule verdict table (brief section 4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_not_required_item_is_satisfied_even_when_absent():
|
|
"""Medicare FFS does not require PA -> PA is satisfied with no input at all."""
|
|
v = evaluate_readiness("medicare", csv_pa_status=None)
|
|
pa = items_by_type(v)["pa"]
|
|
assert pa.required_state == RequiredState.NOT_REQUIRED
|
|
assert pa.satisfied is True
|
|
|
|
|
|
def test_required_supplied_good_is_satisfied():
|
|
v = evaluate_readiness("medicare", csv_swo_status="On File")
|
|
swo = items_by_type(v)["swo"]
|
|
assert swo.required_state == RequiredState.REQUIRED
|
|
assert swo.input_state == InputState.SUPPLIED
|
|
assert swo.satisfied is True
|
|
|
|
|
|
def test_required_supplied_bad_is_gap():
|
|
v = evaluate_readiness("medicare", csv_swo_status="Expired")
|
|
swo = items_by_type(v)["swo"]
|
|
assert swo.satisfied is False
|
|
|
|
|
|
def test_required_absent_blocks_green():
|
|
"""THE false-green killer: a required item with no data is never satisfied."""
|
|
v = evaluate_readiness("medicare", csv_swo_status=None)
|
|
swo = items_by_type(v)["swo"]
|
|
assert swo.required_state == RequiredState.REQUIRED
|
|
assert swo.input_state == InputState.ABSENT
|
|
assert swo.satisfied is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Line status — clear to ship is earned, not defaulted
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _all_docs_good(**overrides):
|
|
base = dict(
|
|
csv_swo_status="On File",
|
|
csv_visit_date=VISIT,
|
|
csv_pecos_verified="Yes",
|
|
csv_pa_status="Not Required",
|
|
csv_diagnosis_on_file="Yes",
|
|
)
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def test_clear_to_ship_when_every_required_item_satisfied():
|
|
v = evaluate_readiness("medicare", **_all_docs_good())
|
|
assert v.line_status == LineStatus.CLEAR_TO_SHIP
|
|
|
|
|
|
def test_missing_required_item_is_not_clear_to_ship():
|
|
"""Old code greened this (absent column = no gap). New code must not."""
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status=None))
|
|
assert v.line_status != LineStatus.CLEAR_TO_SHIP
|
|
assert v.line_status == LineStatus.ACTION_NEEDED
|
|
|
|
|
|
def test_supplied_bad_required_item_is_at_risk():
|
|
v = evaluate_readiness("medicare_advantage", **_all_docs_good(csv_pa_status="Denied"))
|
|
assert v.line_status == LineStatus.AT_RISK
|
|
|
|
|
|
def test_pending_only_is_on_track():
|
|
"""No hard gap, no missing data, but an item still in progress -> On Track."""
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status="Pending"))
|
|
assert v.line_status == LineStatus.ON_TRACK
|
|
|
|
|
|
def test_bad_outranks_absent_in_severity():
|
|
"""A denied item (At Risk) outranks a missing item (Action Needed)."""
|
|
v = evaluate_readiness(
|
|
"medicare_advantage",
|
|
**_all_docs_good(csv_pa_status="Denied", csv_diagnosis_on_file=None),
|
|
)
|
|
assert v.line_status == LineStatus.AT_RISK
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan type unknown — never guess, cannot be green
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_plan_type_none_is_plan_type_needed():
|
|
v = evaluate_readiness(None, **_all_docs_good())
|
|
assert v.line_status == LineStatus.PLAN_TYPE_NEEDED
|
|
|
|
|
|
def test_plan_type_unknown_string_is_plan_type_needed():
|
|
"""An unmapped plan value is treated as unknown, never defaulted/guessed."""
|
|
v = evaluate_readiness("unknown", **_all_docs_good())
|
|
assert v.line_status == LineStatus.PLAN_TYPE_NEEDED
|
|
|
|
|
|
def test_plan_unknown_leaves_pecos_and_pa_not_evaluated():
|
|
v = evaluate_readiness(None, **_all_docs_good())
|
|
items = items_by_type(v)
|
|
assert items["pecos"].required_state == RequiredState.NOT_EVALUATED
|
|
assert items["pa"].required_state == RequiredState.NOT_EVALUATED
|
|
|
|
|
|
def test_plan_unknown_still_grades_universal_items():
|
|
"""SWO, visit, diagnosis are universal (LCD L33822) -> still graded when plan unknown."""
|
|
v = evaluate_readiness(None, csv_swo_status="On File")
|
|
swo = items_by_type(v)["swo"]
|
|
assert swo.required_state == RequiredState.REQUIRED
|
|
assert swo.satisfied is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Required-ness comes from payer_rules.json, not duplicated Python sets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_pecos_required_for_medicare():
|
|
v = evaluate_readiness("medicare", csv_pecos_verified=None)
|
|
assert items_by_type(v)["pecos"].required_state == RequiredState.REQUIRED
|
|
|
|
|
|
def test_pecos_not_required_for_medicaid():
|
|
"""Medicaid uses state enrollment, not PECOS (deferred) -> PECOS not required."""
|
|
v = evaluate_readiness("medicaid", csv_pecos_verified=None)
|
|
assert items_by_type(v)["pecos"].required_state == RequiredState.NOT_REQUIRED
|
|
|
|
|
|
def test_pa_required_for_medicaid():
|
|
v = evaluate_readiness("medicaid", csv_pa_status=None)
|
|
assert items_by_type(v)["pa"].required_state == RequiredState.REQUIRED
|
|
|
|
|
|
def test_pa_not_required_for_medicare_ffs():
|
|
v = evaluate_readiness("medicare", csv_pa_status=None)
|
|
assert items_by_type(v)["pa"].required_state == RequiredState.NOT_REQUIRED
|
|
|
|
|
|
def test_all_five_items_present():
|
|
v = evaluate_readiness("medicare", **_all_docs_good())
|
|
assert set(items_by_type(v).keys()) == {"swo", "visit", "pecos", "pa", "diagnosis"}
|
|
|
|
|
|
def test_diagnosis_supplied_negative_is_gap():
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="No"))
|
|
dx = items_by_type(v)["diagnosis"]
|
|
assert dx.input_state == InputState.SUPPLIED
|
|
assert dx.satisfied is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hardening (code-auditor findings H1, H2, M1, M2, M3, M4, L1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_known_plan_missing_config_does_not_false_green():
|
|
"""H1/H2: a known plan absent from payer_rules.json must NOT grade green.
|
|
|
|
PECOS/PA cannot be evaluated without their rule config -> NOT_EVALUATED,
|
|
which must block 'Clear to Ship'. Fails safe, never open.
|
|
"""
|
|
v = evaluate_readiness(
|
|
"medicaid", rules={"payer_rules": {}},
|
|
**_all_docs_good(csv_pa_status="Denied"),
|
|
)
|
|
assert v.line_status != LineStatus.CLEAR_TO_SHIP
|
|
assert items_by_type(v)["pa"].required_state == RequiredState.NOT_EVALUATED
|
|
assert items_by_type(v)["pecos"].required_state == RequiredState.NOT_EVALUATED
|
|
|
|
|
|
def test_diagnosis_icd10_code_counts_as_on_file():
|
|
"""M2: a real ICD-10 code in the diagnosis field is positive evidence, not a gap."""
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="E11.9"))
|
|
assert items_by_type(v)["diagnosis"].satisfied is True
|
|
|
|
|
|
def test_diagnosis_explicit_negative_is_bad():
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="Missing"))
|
|
assert items_by_type(v)["diagnosis"].satisfied is False
|
|
|
|
|
|
def test_pecos_enrolled_synonym_is_good():
|
|
"""M3: standard PECOS portal statuses should read as verified."""
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_pecos_verified="Enrolled"))
|
|
assert items_by_type(v)["pecos"].satisfied is True
|
|
|
|
|
|
def test_swo_active_synonym_is_good():
|
|
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status="Active"))
|
|
assert items_by_type(v)["swo"].satisfied is True
|
|
|
|
|
|
def test_plan_type_case_and_whitespace_normalized():
|
|
"""M4: a correctly-mapped plan with stray case/space must still grade."""
|
|
v = evaluate_readiness(" Medicare ", **_all_docs_good())
|
|
assert v.line_status == LineStatus.CLEAR_TO_SHIP
|
|
|
|
|
|
def test_staff_confirmed_visit_carries_value_for_citation():
|
|
"""L1: a staff-confirmed visit (no CSV date) still records its date for the citation."""
|
|
v = evaluate_readiness(
|
|
"medicare", confirmed_visit_date=VISIT,
|
|
csv_swo_status="On File", csv_pa_status="Not Required",
|
|
csv_diagnosis_on_file="Yes", csv_pecos_verified="Yes",
|
|
)
|
|
assert items_by_type(v)["visit"].value == VISIT.isoformat()
|
|
|
|
|
|
def test_known_plan_types_match_payer_rules_config():
|
|
"""M1 drift guard: the gradeable allow-list must equal the configured plans."""
|
|
import json
|
|
rules_path = (
|
|
Path(__file__).parent.parent
|
|
/ "python-backend" / "config" / "payer_rules.json"
|
|
)
|
|
configured = set(json.loads(rules_path.read_text())["payer_rules"].keys()) - {"default"}
|
|
from core.readiness import KNOWN_PLAN_TYPES
|
|
assert set(KNOWN_PLAN_TYPES) == configured
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P2: citation rule IDs (build plan 02, readiness-model-design-phase2 §3.1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_doc_items_carry_rule_ids():
|
|
v = evaluate_readiness(
|
|
"medicare",
|
|
csv_swo_status="On File",
|
|
csv_visit_date=VISIT,
|
|
csv_pecos_verified="Yes",
|
|
csv_diagnosis_on_file="Yes",
|
|
)
|
|
items = items_by_type(v)
|
|
assert items["swo"].rule_id == "R100"
|
|
assert items["visit"].rule_id == "R101"
|
|
assert items["diagnosis"].rule_id == "R102"
|
|
assert items["pecos"].rule_id == "R004"
|
|
assert items["pa"].rule_id == "R003"
|
|
|
|
|
|
def test_flat_and_wrapped_config_produce_identical_verdicts():
|
|
flat = {"payer_rules": {"medicare": {
|
|
"pa_required": False, "pecos_required": True,
|
|
}}}
|
|
wrapped = {"payer_rules": {"medicare": {
|
|
"pa_required": {"value": False, "rule_id": "R003"},
|
|
"pecos_required": {"value": True, "rule_id": "R004"},
|
|
}}}
|
|
kwargs = dict(
|
|
csv_swo_status="On File", csv_visit_date=VISIT,
|
|
csv_pecos_verified="Yes", csv_diagnosis_on_file="Yes",
|
|
)
|
|
v_flat = evaluate_readiness("medicare", rules=flat, **kwargs)
|
|
v_wrapped = evaluate_readiness("medicare", rules=wrapped, **kwargs)
|
|
assert v_flat.line_status == v_wrapped.line_status
|
|
for a, b in zip(v_flat.doc_items, v_wrapped.doc_items):
|
|
assert (a.doc_type, a.required_state, a.satisfied) == (
|
|
b.doc_type, b.required_state, b.satisfied
|
|
)
|
|
|
|
|
|
def test_not_evaluated_items_have_no_rule_id():
|
|
v = evaluate_readiness(None)
|
|
items = items_by_type(v)
|
|
assert items["pecos"].required_state == RequiredState.NOT_EVALUATED
|
|
assert items["pecos"].rule_id is None
|
|
assert items["pa"].rule_id is None
|
|
|
|
|
|
def test_on_disk_rule_ids_match_registry():
|
|
"""Guards citation drift: every wrapped rule_id in payer_rules.json must
|
|
equal the registry ID for its key (auditor finding L2, 2026-07-07)."""
|
|
import json
|
|
from pathlib import Path as _P
|
|
from core.rules import PLAN_KEY_RULE_IDS
|
|
|
|
cfg = json.loads(
|
|
(_P(__file__).parent.parent / "python-backend" / "config" / "payer_rules.json").read_text()
|
|
)
|
|
for plan, entry in cfg["payer_rules"].items():
|
|
for key, expected_id in PLAN_KEY_RULE_IDS.items():
|
|
raw = entry.get(key)
|
|
assert isinstance(raw, dict) and raw.get("rule_id") == expected_id, (
|
|
f"{plan}.{key} must be wrapped with rule_id {expected_id}"
|
|
)
|