0320dc17ba
Make the free NPI compliance check the catch-all for ALL verified institutional providers, but route anyone with a more important/time-sensitive issue to THAT email instead -- each provider gets exactly one email, their most urgent. - SEGMENTS gain a 'priority' (lower=more urgent): reactivation 10, revalidation overdue 20, due-soon 30, bundle 45, free-NPI-check 100 (catch-all). - assign_segment()/assign_all(): route each provider to the single highest-priority active segment whose selector matches; warm_segment() takes the assignment map and only claims its assigned providers (disjoint pools, no double-mailing). main() now splits the daily slice by priority order, serving urgent segments fully before the broad free-check consumes the remainder. - nppes_outdated selector -> 'institutional_default' (every verified, non- deactivated row), since the free check's value no longer depends on staleness; list/campaign renamed 'HC Warmup - Free NPI Check'. - FIX latent bug: reactivation selector treated 'not on CMS reval list' as deactivated -- false for org NPIs (would mis-tell active practices they're deactivated). Now uses the REAL nppes_deactivated flag (or OIG/SAM exclusion). - Drop blanket oig_screening from the default rotation: it matched every row and would starve the catch-all, and the free check already screens OIG/SAM and routes to the paid fix on a hit. Still runnable via --segments. - Add scripts/test_segment_assignment.py (10 cases incl. 'overdue AND stale -> overdue wins'); all pass.
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Regression test for one-email-per-provider segment assignment.
|
|
|
|
Verifies assign_segment() routes each provider to their single MOST-URGENT
|
|
matching segment (lowest priority number), so a provider who qualifies for
|
|
several segments gets exactly one email -- the most important one -- and everyone
|
|
else falls through to the free NPI compliance check (the catch-all default).
|
|
|
|
Run: python3 scripts/test_segment_assignment.py (exit 0 = pass)
|
|
"""
|
|
import sys, os
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import build_healthcare_campaigns_cron as cron # noqa: E402
|
|
|
|
ACTIVE = ["npi_reactivation", "revalidation_overdue", "revalidation_due_soon",
|
|
"nppes_outdated", "compliance_bundle"]
|
|
|
|
|
|
def row(**kw):
|
|
base = {"email": "x@p.com", "verify_ok": "Y", "reval_status": "",
|
|
"days_overdue": "", "leie_excluded": "", "optout_ending": "",
|
|
"nppes_deactivated": "", "nppes_years_stale": ""}
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
CASES = [
|
|
("plain verified institutional -> free check",
|
|
row(reval_status="no_reval_flag"), "nppes_outdated"),
|
|
("overdue 30d -> revalidation_overdue beats free check",
|
|
row(reval_status="overdue", days_overdue="30"), "revalidation_overdue"),
|
|
("due soon 30d -> revalidation_due_soon",
|
|
row(reval_status="upcoming", days_overdue="-30"), "revalidation_due_soon"),
|
|
("OIG excluded -> reactivation (most urgent), beats overdue",
|
|
row(leie_excluded="Y", reval_status="overdue", days_overdue="20"),
|
|
"npi_reactivation"),
|
|
("NPPES deactivated -> reactivation",
|
|
row(nppes_deactivated="Y", reval_status="no_reval_flag"), "npi_reactivation"),
|
|
("overdue AND stale -> overdue wins, NOT free check",
|
|
row(reval_status="overdue", days_overdue="30", nppes_years_stale="9"),
|
|
"revalidation_overdue"),
|
|
("deactivated -> NOT free check even though verified",
|
|
row(nppes_deactivated="Y"), "npi_reactivation"),
|
|
("optout ending + nothing urgent -> bundle beats free check",
|
|
row(optout_ending="2025-12-01", reval_status="no_reval_flag"),
|
|
"compliance_bundle"),
|
|
("not verified + nothing -> no assignment",
|
|
row(verify_ok="N", reval_status="no_reval_flag"), None),
|
|
("heavily overdue 400d (out of window) -> free check",
|
|
row(reval_status="overdue", days_overdue="400"), "nppes_outdated"),
|
|
]
|
|
|
|
|
|
def main() -> int:
|
|
failures = 0
|
|
for desc, r, exp in CASES:
|
|
got = cron.assign_segment(r, ACTIVE)
|
|
ok = got == exp
|
|
if not ok:
|
|
failures += 1
|
|
print(f" [{'ok' if ok else 'FAIL'}] {desc:58} -> {got} (exp {exp})")
|
|
# Sanity: every provider gets AT MOST one segment (assign_all is a dict).
|
|
rows = [r for _, r, _ in CASES]
|
|
a = cron.assign_all(rows, ACTIVE)
|
|
assert all(isinstance(v, str) for v in a.values()), "assignment must be single-valued"
|
|
print(f"\n{'PASS' if not failures else f'{failures} FAILED'}")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|