695ace207c
Copy: drop paper/electronic/fax framing across the revalidation + enrollment marketing pages and the order-confirmation email; present two service tiers: - Standard filing (no CMS account; we prepare CMS-855, you sign, we submit to MAC) - Expedited filing (CMS I&A surrogate access; same-day PECOS filing + tracking) Internal worker todos + the _STANDARD_FILING_SLUGS identifier updated to match. New scripts/test_healthcare_e2e.py validates the whole order line (slug consistency x6 places, price agreement, intake field collection+enforcement, worker dispatch, handler execution producing CMS-855 PDF+anchor, free-tool action_urls). 45 checks. Bugs found + fixed by the test: - medicare-enrollment requires practice_state server-side but the wizard never enforced it -> orders could be paid then stall. Wizard now requires it. - determine_form_type defaulted org NPIs to the individual 855I because enumeration_type is never collected -> wrong form, CMS rejection. Now does a live NPPES lookup (safe 855I fallback).
284 lines
12 KiB
Python
284 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""End-to-end consistency + flow test for the Healthcare / NPI ordering line.
|
|
|
|
Validates the full order path across all wiring points and runs the worker
|
|
handlers for real (PDF generation, todo creation), catching logical errors.
|
|
|
|
Checks:
|
|
1. Slug consistency across all 5 registration points.
|
|
2. Pricing agreement (catalog vs intake manifest vs ERPNext www/orders).
|
|
3. Intake manifest steps + the wizard collects every server-required field.
|
|
4. Worker dispatch maps every slug to a handler.
|
|
5. Each handler runs end-to-end (mocked DB/MinIO/esign) and produces the
|
|
right artifacts: CMS-855 PDF + signature anchor for filing slugs, todo
|
|
for all.
|
|
6. The free-tool action_urls point at real order/service slugs.
|
|
|
|
Run: python3 scripts/test_healthcare_e2e.py
|
|
Exit code 0 = all pass, 1 = failures.
|
|
"""
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
HEALTHCARE_SLUGS = [
|
|
"npi-revalidation", "npi-reactivation", "nppes-update",
|
|
"medicare-enrollment", "oig-sam-screening", "provider-compliance-bundle",
|
|
]
|
|
# Slugs that should generate a CMS-855 PDF + e-sign anchor.
|
|
FILING_SLUGS = {"npi-revalidation", "npi-reactivation", "medicare-enrollment"}
|
|
|
|
failures: list[str] = []
|
|
passes: list[str] = []
|
|
|
|
|
|
def ok(msg):
|
|
passes.append(msg)
|
|
|
|
|
|
def fail(msg):
|
|
failures.append(msg)
|
|
|
|
|
|
def read(p: Path) -> str:
|
|
return (ROOT / p).read_text()
|
|
|
|
|
|
# ── 1. Slug consistency across the 5 registration points ──────────────────
|
|
def check_slug_consistency():
|
|
sources = {
|
|
"compliance-orders.ts (catalog)": ("api/src/routes/compliance-orders.ts",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":\s*\{'),
|
|
"compliance-orders.ts (intake reqs)": ("api/src/routes/compliance-orders.ts",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":\s*\{ required'),
|
|
"intake_manifest.ts (steps)": ("site/src/lib/intake_manifest.ts",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":\s*\['),
|
|
"intake_manifest.ts (meta)": ("site/src/lib/intake_manifest.ts",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":\s*\{ name'),
|
|
"www/orders.py": ("performancewest_erpnext/performancewest_erpnext/www/orders.py",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":'),
|
|
"services/__init__.py (dispatch)": ("scripts/workers/services/__init__.py",
|
|
r'"(npi-[a-z-]+|nppes-update|medicare-enrollment|oig-sam-screening|provider-compliance-bundle)":\s*\w+Handler'),
|
|
}
|
|
want = set(HEALTHCARE_SLUGS)
|
|
for label, (path, pat) in sources.items():
|
|
try:
|
|
txt = read(Path(path))
|
|
except FileNotFoundError:
|
|
fail(f"[slugs] file missing: {path}")
|
|
continue
|
|
found = set(re.findall(pat, txt))
|
|
missing = want - found
|
|
extra = found - want
|
|
if missing:
|
|
fail(f"[slugs] {label}: MISSING {sorted(missing)}")
|
|
elif extra:
|
|
fail(f"[slugs] {label}: UNEXPECTED {sorted(extra)}")
|
|
else:
|
|
ok(f"[slugs] {label}: all 6 present")
|
|
|
|
|
|
# ── 2. Pricing agreement ──────────────────────────────────────────────────
|
|
def check_pricing():
|
|
cat = read(Path("api/src/routes/compliance-orders.ts"))
|
|
man = read(Path("site/src/lib/intake_manifest.ts"))
|
|
|
|
def prices(txt):
|
|
out = {}
|
|
for slug in HEALTHCARE_SLUGS:
|
|
m = re.search(rf'"{re.escape(slug)}":\s*\{{[^}}]*?price_cents:\s*(\d+)', txt, re.S)
|
|
if m:
|
|
out[slug] = int(m.group(1))
|
|
return out
|
|
|
|
cprices, mprices = prices(cat), prices(man)
|
|
for slug in HEALTHCARE_SLUGS:
|
|
c, m = cprices.get(slug), mprices.get(slug)
|
|
if c is None:
|
|
fail(f"[price] {slug}: no price in catalog")
|
|
elif m is None:
|
|
fail(f"[price] {slug}: no price in intake manifest")
|
|
elif c != m:
|
|
fail(f"[price] {slug}: catalog {c} != manifest {m}")
|
|
else:
|
|
ok(f"[price] {slug}: ${c/100:.0f} consistent")
|
|
|
|
|
|
# ── 3. Intake: wizard collects every server-required field ────────────────
|
|
def check_intake_fields():
|
|
cat = read(Path("api/src/routes/compliance-orders.ts"))
|
|
step = read(Path("site/src/components/intake/steps/NpiIntakeStep.astro"))
|
|
|
|
# Fields the wizard writes into intake_data (from PW.set({...}) block).
|
|
set_block = re.search(r"PW\.set\(\{[^}]*intake_data:\s*\{(.+?)\}\}\)", step, re.S)
|
|
collected = set(re.findall(r"(\w+):", set_block.group(1))) if set_block else set()
|
|
|
|
# Fields the wizard *enforces* as required (the `missing.push` validations).
|
|
enforced = set()
|
|
if "npi-provider-name" in step and "missing.push" in step:
|
|
if 'val("npi-provider-name")' in step: enforced.add("provider_name")
|
|
if 'val("npi-number")' in step or "val(\"npi-number\")" in step: enforced.add("npi")
|
|
if 'val("npi-email")' in step: enforced.add("email")
|
|
|
|
# Slug-conditional enforcement: e.g. practice_state is only required when
|
|
# the active service is medicare-enrollment. Map field -> {slugs}.
|
|
conditional_enforced = {}
|
|
if 'activeSlugs.includes("medicare-enrollment")' in step and 'val("npi-practice-state")' in step:
|
|
conditional_enforced["practice_state"] = {"medicare-enrollment"}
|
|
|
|
for slug in HEALTHCARE_SLUGS:
|
|
m = re.search(rf'"{re.escape(slug)}":\s*\{{ required:\s*\[([^\]]*)\]', cat)
|
|
if not m:
|
|
fail(f"[intake] {slug}: no required-fields entry in catalog")
|
|
continue
|
|
required = set(re.findall(r'"([a-z_]+)"', m.group(1)))
|
|
# every required field must be collectable by the wizard
|
|
not_collected = required - collected
|
|
if not_collected:
|
|
fail(f"[intake] {slug}: required {sorted(not_collected)} NOT collected by wizard")
|
|
# every required field must be *enforced* — either always, or
|
|
# conditionally for this slug.
|
|
not_enforced = set()
|
|
for fld in required - enforced:
|
|
if slug not in conditional_enforced.get(fld, set()):
|
|
not_enforced.add(fld)
|
|
if not_enforced:
|
|
fail(f"[intake] {slug}: required {sorted(not_enforced)} collected but NOT validated as required in wizard")
|
|
if not not_collected and not not_enforced:
|
|
ok(f"[intake] {slug}: all required fields collected + enforced")
|
|
|
|
|
|
# ── 4 & 5. Worker dispatch + handler execution ────────────────────────────
|
|
def check_handlers():
|
|
import asyncio
|
|
import workers.services.npi_provider as npi
|
|
from workers.services import SERVICE_HANDLERS
|
|
|
|
for slug in HEALTHCARE_SLUGS:
|
|
if slug not in SERVICE_HANDLERS:
|
|
fail(f"[dispatch] {slug}: no handler registered")
|
|
continue
|
|
ok(f"[dispatch] {slug}: -> {SERVICE_HANDLERS[slug].__name__}")
|
|
|
|
# Run each handler with mocked IO and verify artifacts.
|
|
captured = {}
|
|
|
|
def fake_todo(self, order_number, intake, title, description, priority="normal"):
|
|
captured.setdefault(self.SERVICE_SLUG, {})["todo"] = {"title": title, "desc": description}
|
|
|
|
esign_calls = {}
|
|
|
|
def fake_esign(self, order_number, intake, provider, customer_email, form_type, document_key, anchors):
|
|
esign_calls[self.SERVICE_SLUG] = {"form_type": form_type, "anchors": anchors, "key": document_key}
|
|
return True # pretend esign + email succeeded
|
|
|
|
# Stub MinIO upload inside the filler path by stubbing _generate_855... upload
|
|
orig_gen = npi._BaseNPIHandler._generate_855_for_signing
|
|
|
|
def gen_no_upload(self, order_number, intake, provider, customer_email):
|
|
from document_gen.templates.cms855_pdf_filler import determine_form_type, fill_cms855
|
|
ft = determine_form_type(self.SERVICE_SLUG, intake)
|
|
pdf, anchors, missing = fill_cms855(ft, intake, order_number)
|
|
captured.setdefault(self.SERVICE_SLUG, {})["pdf"] = {"len": len(pdf), "anchors": anchors, "missing": missing, "form_type": ft}
|
|
# exercise the esign record path (stubbed)
|
|
self._create_855_esign_record(order_number, intake, provider, customer_email, ft, f"compliance/{order_number}/cms{ft}.pdf", anchors)
|
|
return f"CMS-{ft.upper()} generated ({len(pdf)} bytes)"
|
|
|
|
npi._BaseNPIHandler._create_todo = fake_todo
|
|
npi._BaseNPIHandler._create_855_esign_record = fake_esign
|
|
npi._BaseNPIHandler._generate_855_for_signing = gen_no_upload
|
|
|
|
intake = {
|
|
"npi": "1234567893", "provider_name": "Jane Q Smith", "email": "jane@example.com",
|
|
"dob": "01011980", "practice_state": "CA", "enumeration_type": "NPI-1",
|
|
}
|
|
for slug in HEALTHCARE_SLUGS:
|
|
h = SERVICE_HANDLERS[slug]()
|
|
order = {"order_number": f"CO-T-{slug}", "customer_name": "Jane Q Smith",
|
|
"customer_email": "jane@example.com", "intake_data": dict(intake)}
|
|
try:
|
|
asyncio.run(h.process(order))
|
|
except Exception as e:
|
|
fail(f"[handler] {slug}: raised {type(e).__name__}: {e}")
|
|
continue
|
|
|
|
c = captured.get(slug, {})
|
|
if "todo" not in c:
|
|
fail(f"[handler] {slug}: no admin todo created")
|
|
else:
|
|
ok(f"[handler] {slug}: admin todo created")
|
|
|
|
if slug in FILING_SLUGS:
|
|
if "pdf" not in c:
|
|
fail(f"[handler] {slug}: expected CMS-855 PDF, none generated")
|
|
else:
|
|
pdf = c["pdf"]
|
|
if pdf["len"] < 10000:
|
|
fail(f"[handler] {slug}: PDF suspiciously small ({pdf['len']} bytes)")
|
|
elif not pdf["anchors"]:
|
|
fail(f"[handler] {slug}: PDF has NO signature anchor (form={pdf['form_type']})")
|
|
else:
|
|
a = pdf["anchors"][0]
|
|
need = {"field", "page", "x", "y", "w", "h", "page_w", "page_h"}
|
|
if not need.issubset(a):
|
|
fail(f"[handler] {slug}: anchor missing keys {need - set(a)}")
|
|
else:
|
|
ok(f"[handler] {slug}: CMS-{pdf['form_type'].upper()} {pdf['len']}B + anchor on page {a['page']}")
|
|
if slug not in esign_calls:
|
|
fail(f"[handler] {slug}: esign record was not requested")
|
|
else:
|
|
ok(f"[handler] {slug}: esign record requested (form {esign_calls[slug]['form_type']})")
|
|
else:
|
|
if "pdf" in c:
|
|
fail(f"[handler] {slug}: unexpectedly generated a CMS-855 PDF (should not)")
|
|
else:
|
|
ok(f"[handler] {slug}: correctly skips CMS-855 generation")
|
|
|
|
|
|
# ── 6. Free-tool action_urls point at real slugs ──────────────────────────
|
|
def check_free_tool_links():
|
|
try:
|
|
lookup = read(Path("api/src/routes/npi-lookup.ts"))
|
|
except FileNotFoundError:
|
|
fail("[tool] npi-lookup.ts missing")
|
|
return
|
|
urls = set(re.findall(r'action_url:\s*"(/(?:order|services/healthcare)[^"]*)"', lookup))
|
|
order_pages = {p.stem for p in (ROOT / "site/src/pages/order").glob("*.astro")}
|
|
for u in sorted(urls):
|
|
if u.startswith("/order/"):
|
|
slug = u.split("/order/")[1].strip("/")
|
|
if slug not in order_pages:
|
|
fail(f"[tool] action_url {u} -> no order page /order/{slug}.astro")
|
|
else:
|
|
ok(f"[tool] action_url {u} resolves")
|
|
else:
|
|
# /services/healthcare[/x]
|
|
tail = u.replace("/services/healthcare", "").strip("/")
|
|
target = ROOT / "site/src/pages/services/healthcare" / (f"{tail}.astro" if tail else "index.astro")
|
|
if not target.exists():
|
|
fail(f"[tool] action_url {u} -> no page {target.relative_to(ROOT)}")
|
|
else:
|
|
ok(f"[tool] action_url {u} resolves")
|
|
|
|
|
|
def main():
|
|
check_slug_consistency()
|
|
check_pricing()
|
|
check_intake_fields()
|
|
check_handlers()
|
|
check_free_tool_links()
|
|
|
|
print("\n".join(f" PASS {p}" for p in passes))
|
|
if failures:
|
|
print("\n".join(f" FAIL {f}" for f in failures))
|
|
print(f"\n{len(passes)} passed, {len(failures)} FAILED")
|
|
sys.exit(1)
|
|
print(f"\nALL {len(passes)} CHECKS PASSED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|