138fec17e9
Standard (no-login) CMS filings are mailed in one Priority Mail envelope per destination agency, batched each postal working-day morning to save postage. - migration 089: paper_filing_batches table + esign_records.paper_batch_id / filing_destination_key (idempotent: a filing is batched at most once). - batch_cover_sheet.py: per-agency cover sheet (sender/dest/date/manifest) + merged print-job PDF (cover + all enclosed signed filings). - daily_paper_batch.py worker: gather signed+unbatched cms855/cms10114 filings, group by destination (MAC by state via mac_routing; Fargo for CMS-10114), build cover+merged PDF per agency, persist batch, mark filings batched. Self-gates on postal working days (skips weekends + federal/USPS holidays). Phase 1 = human prints+mails; phase 2 = wire print-mail API. - worker-crons: pw-paper-batch systemd timer (Mon-Fri 13:30 UTC, self-gated). - test_paper_batch.py: 15/15 pass (working-day gating, routing, cover+merge).
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""Tests for the daily paper-filing batch (Standard no-login CMS filing path).
|
|
|
|
Covers the pure logic that doesn't need a DB/MinIO:
|
|
- postal working-day gating (weekends + federal/USPS holidays)
|
|
- destination routing (state -> MAC; CMS-10114 -> NPI Enumerator)
|
|
- cover-sheet rendering + multi-page pagination
|
|
- merge of cover sheet + filing PDFs
|
|
|
|
Run: python scripts/tests/test_paper_batch.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import io
|
|
import sys
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "workers"))
|
|
|
|
|
|
def _load(name: str, rel: str):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / rel)
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
return m
|
|
|
|
|
|
dpb = _load("dpb", "scripts/workers/daily_paper_batch.py")
|
|
bcs = _load("bcs", "scripts/document_gen/templates/batch_cover_sheet.py")
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
|
|
def check(label, cond):
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" PASS {label}")
|
|
else:
|
|
FAIL += 1
|
|
print(f" FAIL {label}")
|
|
|
|
|
|
def test_working_days():
|
|
print("working-day gating")
|
|
check("Saturday is not a working day", not dpb._is_postal_working_day(date(2026, 6, 6)))
|
|
check("Sunday is not a working day", not dpb._is_postal_working_day(date(2026, 6, 7)))
|
|
check("Monday is a working day", dpb._is_postal_working_day(date(2026, 6, 8)))
|
|
check("Memorial Day (holiday) is not a working day", not dpb._is_postal_working_day(date(2026, 5, 25)))
|
|
check("Christmas (holiday) is not a working day", not dpb._is_postal_working_day(date(2026, 12, 25)))
|
|
|
|
|
|
def test_routing():
|
|
print("destination routing")
|
|
ca = dpb._destination_for("CA", "cms855i")
|
|
check("CA 855 -> Noridian JE", ca and ca[0] == "noridian_je")
|
|
ny = dpb._destination_for("NY", "cms855i")
|
|
check("NY 855 -> NGS JK", ny and ny[0] == "ngs_jk")
|
|
tx = dpb._destination_for("TX", "cms855b")
|
|
check("TX 855B -> Novitas JH", tx and tx[0] == "novitas_jh")
|
|
enum = dpb._destination_for("TX", "cms10114")
|
|
check("any-state CMS-10114 -> NPI Enumerator (Fargo)", enum and enum[0] == "npi_enumerator")
|
|
check("CMS-10114 ignores state for routing", dpb._destination_for("CA", "cms10114")[0] == "npi_enumerator")
|
|
check("unknown state -> None (human review)", dpb._destination_for("ZZ", "cms855i") is None)
|
|
check("blank state -> None", dpb._destination_for("", "cms855i") is None)
|
|
|
|
|
|
def test_cover_sheet():
|
|
print("cover sheet + merge")
|
|
items = [
|
|
{"order_number": f"CO-X{i:03d}", "provider": f"Provider {i} MD",
|
|
"npi": f"1{i:09d}", "form": "855i"}
|
|
for i in range(40)
|
|
]
|
|
cover = bcs.build_cover_sheet(
|
|
destination_name="Noridian JE",
|
|
destination_address_lines=("Provider Enrollment (JE)", "P.O. Box VERIFY", "Fargo, ND"),
|
|
batch_date=date(2026, 6, 8),
|
|
items=items,
|
|
)
|
|
from pypdf import PdfReader
|
|
n_cover = len(PdfReader(io.BytesIO(cover)).pages)
|
|
check("40-item cover sheet paginates to >1 page", n_cover > 1)
|
|
merged = bcs.merge_batch_pdf(cover, [cover, cover])
|
|
n_merged = len(PdfReader(io.BytesIO(merged)).pages)
|
|
check("merge concatenates cover + 2 filings", n_merged == n_cover * 3)
|
|
# empty-items cover still renders one page
|
|
empty = bcs.build_cover_sheet(
|
|
destination_name="X", destination_address_lines=("a",),
|
|
batch_date=date(2026, 6, 8), items=[],
|
|
)
|
|
check("empty batch cover sheet renders", len(PdfReader(io.BytesIO(empty)).pages) >= 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_working_days()
|
|
test_routing()
|
|
test_cover_sheet()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
sys.exit(1 if FAIL else 0)
|