Reframe healthcare filing as standard vs expedited; e2e test + bug fixes

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).
This commit is contained in:
justin 2026-06-05 03:58:46 -05:00
parent 5cfe9702e2
commit 695ace207c
7 changed files with 381 additions and 54 deletions

View file

@ -23,6 +23,7 @@ from __future__ import annotations
import io
import logging
import re
from pathlib import Path
LOG = logging.getLogger("cms855_pdf_filler")
@ -93,15 +94,45 @@ def _split_name(full: str) -> tuple[str, str, str]:
return parts[0], parts[1][0], " ".join(parts[2:])
def _lookup_enumeration_type(npi: str) -> str:
"""Best-effort NPPES lookup of an NPI's enumeration_type (NPI-1 / NPI-2).
Returns "" if the NPI is missing, malformed, or the lookup fails callers
fall back to the individual form (855I), the safe default.
"""
npi = (npi or "").strip()
if not re.fullmatch(r"\d{10}", npi):
return ""
try:
import json as _json
from urllib.request import urlopen
url = f"https://npiregistry.cms.hhs.gov/api/?version=2.1&number={npi}"
with urlopen(url, timeout=8) as resp: # nosec - public, read-only
data = _json.loads(resp.read().decode("utf-8"))
results = data.get("results") or []
if results:
return (results[0].get("enumeration_type") or "").upper()
except Exception:
LOG.warning("NPPES enumeration_type lookup failed for %s", npi)
return ""
def determine_form_type(slug: str, intake: dict) -> str:
"""Pick which 855 form applies for the order."""
"""Pick which 855 form applies for the order.
Org NPIs (Type 2 / NPI-2) revalidate/enroll on 855B; individuals on 855I.
The wizard does not collect enumeration_type, so when it is absent we look
it up live from NPPES rather than silently defaulting an organization to the
wrong (individual) form, which CMS would reject.
"""
if slug not in ("npi-revalidation", "medicare-enrollment"):
return "855i"
enum_type = (intake.get("enumeration_type") or "").upper()
if slug == "npi-revalidation":
# Org NPIs (Type 2) revalidate on 855B; individuals on 855I.
return "855b" if enum_type in ("NPI-2", "2", "ORGANIZATION") else "855i"
if slug == "medicare-enrollment":
return "855b" if enum_type in ("NPI-2", "2", "ORGANIZATION") else "855i"
return "855i"
if not enum_type:
enum_type = _lookup_enumeration_type(intake.get("npi", ""))
return "855b" if enum_type in ("NPI-2", "2", "ORGANIZATION") else "855i"
def fill_cms855(form_type: str, intake: dict, order_number: str = "") -> tuple[bytes, list[dict], list[str]]:

View file

@ -0,0 +1,284 @@
#!/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()

View file

@ -41,7 +41,7 @@ _SLUG_META = {
"enrollment record, and submit. Capture the PECOS tracking ID."
),
"access": (
"PECOS via CMS I&A surrogacy (preferred). Fallback: paper CMS-855I/B/R, provider wet-signs cert page, mail to provider's MAC."
"Standard: prepare CMS-855I/B/R, provider signs cert, submit to MAC. Expedited: file in PECOS via CMS I&A surrogate access."
),
"priority": "high",
},
@ -53,7 +53,7 @@ _SLUG_META = {
"reason, correct any stale data, and re-certify the record."
),
"access": (
"NPPES via CMS I&A surrogacy. No paper option (NPPES is web-only)."
"NPPES via CMS I&A surrogate access (online-only)."
),
"priority": "high",
},
@ -66,7 +66,7 @@ _SLUG_META = {
"changes and certify."
),
"access": (
"NPPES via CMS I&A surrogacy. No paper option (NPPES is web-only)."
"NPPES via CMS I&A surrogate access (online-only)."
),
"priority": "normal",
},
@ -78,7 +78,7 @@ _SLUG_META = {
"Confirm taxonomy, practice location, and authorized official."
),
"access": (
"PECOS via CMS I&A surrogacy (preferred). Fallback: paper CMS-855, provider wet-signs, mail to MAC."
"Standard: prepare CMS-855, provider signs cert, submit to MAC. Expedited: file in PECOS via CMS I&A surrogate access."
),
"priority": "high",
},
@ -104,17 +104,17 @@ _SLUG_META = {
"record. Set the next revalidation reminder."
),
"access": (
"PECOS/NPPES via CMS I&A surrogacy; screening is public. Paper CMS-855 fallback for the enrollment/revalidation piece."
"Standard CMS-855 filing for the enrollment/revalidation piece; NPPES + PECOS via CMS I&A surrogate access; screening is public."
),
"priority": "high",
},
}
# Slugs whose fulfilment includes a paper CMS-855 (auto-filled official form,
# e-signed, then printed + USPS Priority-mailed to the provider's MAC). The
# Slugs whose fulfilment includes a CMS-855 (auto-filled official form, signed
# via the secure e-sign link, then submitted to the provider's MAC). The
# bundle's revalidation piece is handled by the dedicated revalidation order it
# spawns, so it is not listed here.
_PAPER_855_SLUGS = {
_STANDARD_FILING_SLUGS = {
"npi-revalidation",
"npi-reactivation",
"medicare-enrollment",
@ -151,16 +151,16 @@ class _BaseNPIHandler:
)
# For PECOS enrollment/revalidation we generate the official CMS-855,
# send it for e-signature, then a human prints + USPS-mails it to the MAC.
paper_note = ""
if self.SERVICE_SLUG in _PAPER_855_SLUGS:
# send it for e-signature, then a human submits it to the MAC.
filing_note = ""
if self.SERVICE_SLUG in _STANDARD_FILING_SLUGS:
try:
paper_note = self._generate_855_for_signing(
filing_note = self._generate_855_for_signing(
order_number, intake, provider, customer_email
)
except Exception as exc: # never block the admin todo on PDF issues
LOG.error("[%s] CMS-855 generation failed: %s", order_number, exc)
paper_note = f"CMS-855 auto-generation FAILED ({exc}); prepare the form manually."
filing_note = f"CMS-855 auto-generation FAILED ({exc}); prepare the form manually."
description = (
f"{meta['action']}\n\n"
@ -171,9 +171,9 @@ class _BaseNPIHandler:
f"Practice state: {practice_state or 'not provided'}\n"
f"Portal: {meta['portal']}\n"
f"Access model: {meta['access']}\n"
+ (f"\n{paper_note}\n" if paper_note else "")
+ (f"\n{filing_note}\n" if filing_note else "")
+ "\nReview-staged: complete/verify the form, get it signed, then "
"print and USPS Priority Mail it to the provider's MAC (or file in "
"submit it to the provider's MAC (standard), or file in "
"PECOS if surrogate access was granted). Mark this order complete."
)
@ -191,7 +191,7 @@ class _BaseNPIHandler:
Returns a human-readable note for the admin todo describing what was
generated and what still needs manual completion. The signed PDF is
printed and USPS Priority-mailed to the MAC by the fulfilment team.
submitted to the MAC by the fulfilment team.
"""
try:
from scripts.document_gen.templates.cms855_pdf_filler import (
@ -231,13 +231,13 @@ class _BaseNPIHandler:
)
note_lines = [
f"PAPER CMS-{form_type.upper()} generated (official form, auto-filled where possible).",
f"CMS-{form_type.upper()} generated (official form, auto-filled where possible).",
f"Unsigned PDF: {document_key}",
]
if signed and customer_email:
note_lines.append(f"E-sign link emailed to {customer_email}. After signing, print + USPS Priority Mail to the MAC.")
note_lines.append(f"E-sign link emailed to {customer_email}. After signing, submit to the MAC (standard) or file via PECOS surrogate access (expedited).")
else:
note_lines.append("No customer email or esign infra — send the form for wet signature manually.")
note_lines.append("No customer email or esign infra — send the form for signature manually.")
if missing:
note_lines.append("MANUAL COMPLETION NEEDED:")
note_lines.extend(f" - {m}" for m in missing)