new-site/scripts/tests/check_fulfillment_consistency.py

#!/usr/bin/env python3
"""
Compliance fulfillment consistency + intake-completeness checker.

For every DOT/FMCSA + state-trucking + hazmat/emissions service slug this asserts:
  1. It exists in COMPLIANCE_SERVICES  (checkout product/pricing)
  2. It has a REQUIRED_FIELDS spec      (intake validation)
  3. It has a SERVICE_META entry        (intake manifest pricing/name)
  4. It maps to intake steps in INTAKE_MANIFEST
  5. It resolves to a handler in SERVICE_HANDLERS (fulfillment)
  6. Every REQUIRED_FIELDS[slug].required field is actually collectible by the
     intake step(s) the manifest assigns to that slug (the core "we collect all
     the info we need" guarantee).

Run:  python3 scripts/tests/check_fulfillment_consistency.py
Exit 0 = all good; nonzero = gaps found (printed).
"""
from __future__ import annotations
import json
import os
import re
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
ORDERS_TS = os.path.join(ROOT, "api/src/routes/compliance-orders.ts")
MANIFEST_TS = os.path.join(ROOT, "site/src/lib/intake_manifest.ts")

# ── The set of slugs this checker is responsible for (DOT / state / hazmat). ──
TRUCKING_SLUGS = [
    # federal DOT/FMCSA
    "mcs150-update", "ucr-registration", "boc3-filing", "dot-registration",
    "mc-authority", "dot-drug-alcohol", "dot-audit-prep", "dot-full-compliance",
    "usdot-reactivation",
    # state-level trucking
    "irp-registration", "ifta-application", "ifta-quarterly", "or-weight-mile-tax",
    "ny-hut-registration", "ky-kyu-registration", "nm-weight-distance",
    "ct-highway-use-fee", "ca-mcp-carb", "state-dot-registration",
    "intrastate-authority", "osow-permit", "state-trucking-bundle",
    # hazmat / emissions
    "hazmat-phmsa", "state-emissions",
]

# Fields the StateTruckingIntakeStep + DOTIntakeStep emit into intake_data.
# These mirror the front-end PW.set() calls; if you add a field to a step, add it
# here. The checker fails if a REQUIRED_FIELDS entry references a field that no
# assigned step can produce.
STATE_TRUCKING_FIELDS = {
    "legal_name", "entity_name", "dot_number", "mc_number", "email",
    "base_state", "power_units", "fuel_type", "gross_weight_bracket",
    "operating_states", "ca_number", "engine_model_years", "authority_type",
    "boc3_on_file", "insurance_carrier", "insurance_policy", "load_dimensions",
    "load_weight", "hazmat_classes", "bulk_packaging", "small_business",
}
DOT_INTAKE_FIELDS = {
    "dot_number", "mc_number", "legal_name", "ein", "address_street",
    "address_city", "address_state", "address_zip", "phone", "email",
    "signer_name", "signer_title", "power_units", "drivers", "cdl_drivers",
    "carrier_operation", "interstate_intrastate", "hazmat", "annual_miles",
    "cargo_types", "fleet_size_bracket", "owner_operators", "der_name",
    "current_da_provider", "docket_type", "docket_number", "entity_type",
    "reporting_quarter",
}
STEP_FIELDS = {
    "state-trucking": STATE_TRUCKING_FIELDS,
    "dot-intake": DOT_INTAKE_FIELDS,
    # mcs150 step reuses the DOT intake field universe
    "mcs150": DOT_INTAKE_FIELDS,
}


def _extract_block(text: str, decl: str) -> str:
    """Return the text of the object literal assigned in `<decl> ... = { ... }`."""
    m = re.search(re.escape(decl), text)
    if not m:
        raise SystemExit(f"could not find `{decl}` in source")
    # Find the `=` that begins the assignment (skip past any `: Record<...>` type),
    # then the first `{` after it is the object literal.
    eq = text.find("=", m.end())
    if eq == -1:
        raise SystemExit(f"no `=` after {decl}")
    i = text.find("{", eq)
    depth = 0
    for j in range(i, len(text)):
        c = text[j]
        if c == "{":
            depth += 1
        elif c == "}":
            depth -= 1
            if depth == 0:
                return text[i : j + 1]
    raise SystemExit(f"unbalanced braces after {decl}")


def _top_level_keys(block: str) -> set[str]:
    """Keys quoted at brace-depth 1 inside an object literal."""
    keys, depth = set(), 0
    for mm in re.finditer(r'[{}]|"([^"]+)"\s*:', block):
        tok = mm.group(0)
        if tok == "{":
            depth += 1
        elif tok == "}":
            depth -= 1
        elif depth == 1 and mm.group(1) is not None:
            keys.add(mm.group(1))
    return keys


def main() -> int:
    orders = open(ORDERS_TS).read()
    manifest = open(MANIFEST_TS).read()

    services = _top_level_keys(_extract_block(orders, "const COMPLIANCE_SERVICES"))
    req_block = _extract_block(orders, "const REQUIRED_FIELDS")
    required_slugs = _top_level_keys(req_block)
    meta = _top_level_keys(_extract_block(manifest, "SERVICE_META"))
    manifest_map_block = _extract_block(manifest, "INTAKE_MANIFEST")
    manifest_slugs = _top_level_keys(manifest_map_block)

    # Handler registry (import the Python source statically — avoid importing the
    # module which pulls DB deps; parse the dict keys via regex instead).
    init_py = open(os.path.join(ROOT, "scripts/workers/services/__init__.py")).read()
    handler_block = init_py[init_py.find("SERVICE_HANDLERS"):]
    handler_block = handler_block[: handler_block.find("\n}\n") + 2]
    handler_slugs = set(re.findall(r'"([^"]+)"\s*:', handler_block))

    # Per-slug required-fields → which steps the manifest assigns.
    # Parse INTAKE_MANIFEST[slug] = [ "...", ... ]
    manifest_steps: dict[str, list[str]] = {}
    for mm in re.finditer(r'"([^"]+)"\s*:\s*\[([^\]]*)\]', manifest_map_block):
        slug = mm.group(1)
        steps = re.findall(r'"([^"]+)"', mm.group(2))
        manifest_steps[slug] = steps

    # Per-slug required fields list.
    required_fields: dict[str, list[str]] = {}
    for mm in re.finditer(r'"([^"]+)"\s*:\s*\{\s*required\s*:\s*\[([^\]]*)\]', req_block):
        required_fields[mm.group(1)] = re.findall(r'"([^"]+)"', mm.group(2))

    errors: list[str] = []
    for slug in TRUCKING_SLUGS:
        if slug not in services:
            errors.append(f"[{slug}] missing from COMPLIANCE_SERVICES")
        if slug not in required_slugs:
            errors.append(f"[{slug}] missing from REQUIRED_FIELDS")
        if slug not in meta:
            errors.append(f"[{slug}] missing from SERVICE_META")
        if slug not in manifest_slugs:
            errors.append(f"[{slug}] missing from INTAKE_MANIFEST")
        if slug not in handler_slugs:
            errors.append(f"[{slug}] missing from SERVICE_HANDLERS")

        # Intake-completeness: every required field must be collectible.
        steps = manifest_steps.get(slug, [])
        collectible: set[str] = set()
        for st in steps:
            collectible |= STEP_FIELDS.get(st, set())
        for field in required_fields.get(slug, []):
            base = field.split(".")[0]
            if base not in collectible:
                errors.append(
                    f"[{slug}] required field '{field}' is NOT collectible by "
                    f"assigned steps {steps}"
                )

    if errors:
        print("FULFILLMENT CONSISTENCY: FAIL")
        for e in errors:
            print("  -", e)
        return 1

    print(f"FULFILLMENT CONSISTENCY: OK ({len(TRUCKING_SLUGS)} slugs verified)")
    print("  All slugs present in: COMPLIANCE_SERVICES, REQUIRED_FIELDS,")
    print("  SERVICE_META, INTAKE_MANIFEST, SERVICE_HANDLERS.")
    print("  All required fields are collectible by their assigned intake steps.")
    return 0


if __name__ == "__main__":
    sys.exit(main())