new-site/scripts/tests/test_cpni_calea_variants.py

"""Smoke tests for CPNI + CALEA variant generator selection.

Ops 16: confirms every (line_105_primary, infra_type) pair we claim to
support maps to a real importable generator, and that the generic
fallback engages when the pair isn't in the mapping.

Why this is a smoke test, not a full E2E: the full filing flow uploads to
FCC ECFS and requires live creds. Here we just assert:

  1. Each mapped module actually imports without raising.
  2. Each mapped module exposes the expected ``generate_<name>`` function.
  3. Unmapped (primary, infra) pairs fall back to generic (return None).
  4. _pick_template_generator() handles missing line_105_primary gracefully.

Run:
    python -m scripts.tests.test_cpni_calea_variants
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# Make the repo importable (no sys.path surprises under cron / pytest).
REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

# Expected variants for each service. Keep this in sync with the mapping
# dicts inside _pick_template_generator in the two handlers.
CPNI_MAPPING = {
    ("clec", "facilities"):              "cpni_clec_generator",
    ("clec", "reseller"):                "cpni_clec_reseller_generator",
    ("ixc",  "facilities"):              "cpni_ixc_generator",
    ("ixc",  "reseller"):                "cpni_ixc_reseller_generator",
    ("wireless", "facilities"):          "cpni_wireless_generator",
    ("wireless", "mvno"):                "cpni_wireless_mvno_generator",
    ("satellite", "facilities"):         "cpni_satellite_generator",
    ("audio_bridging", "facilities"):    "cpni_audio_bridge_generator",
    ("private_line", "facilities"):      "cpni_private_line_generator",
}

CALEA_MAPPING = {
    ("clec", "facilities"):              "calea_clec_ss7_generator",
    ("clec", "reseller"):                "calea_clec_ss7_generator",
    ("ixc",  "facilities"):              "calea_ixc_ss7_generator",
    ("ixc",  "reseller"):                "calea_ixc_ss7_generator",
    ("wireless", "facilities"):          "calea_wireless_generator",
    ("wireless", "mvno"):                "calea_wireless_mvno_generator",
    ("satellite", "facilities"):         "calea_satellite_generator",
    ("audio_bridging", "facilities"):    "calea_audio_bridge_generator",
}


def _import_check(module_name: str) -> tuple[bool, str]:
    """Import scripts.document_gen.templates.<module_name> and confirm
    it has ``generate_<module_name sans _generator>``. Returns (ok, detail)."""
    try:
        mod = importlib.import_module(
            f"scripts.document_gen.templates.{module_name}"
        )
    except Exception as exc:
        return False, f"import failed: {exc!r}"

    fn_name = "generate_" + module_name.replace("_generator", "")
    fn = getattr(mod, fn_name, None)
    if fn is None:
        return False, f"module loaded but missing callable {fn_name}"
    if not callable(fn):
        return False, f"{fn_name} is not callable"
    return True, "ok"


def _variant_selection_check(
    handler_cls,
    mapping: dict[tuple[str, str], str],
    label: str,
) -> tuple[int, int, list[str]]:
    """Run the handler's _pick_template_generator for each mapped pair
    plus an unmapped one; count passes/fails."""
    passed = 0
    failed = 0
    messages: list[str] = []
    handler = handler_cls()

    # Happy path — every mapped pair returns a callable.
    for (primary, infra), expected_module in mapping.items():
        entity = {
            "line_105_primary": primary,
            "line_105_categories": [{"id": primary, "infra_type": infra}],
        }
        fn = handler._pick_template_generator(entity)
        if callable(fn):
            expected_fn = "generate_" + expected_module.replace("_generator", "")
            actual_fn = getattr(fn, "__name__", "<unknown>")
            if actual_fn == expected_fn:
                passed += 1
            else:
                failed += 1
                messages.append(
                    f"  [{label}] ({primary},{infra}) → {actual_fn} "
                    f"(expected {expected_fn})"
                )
        else:
            failed += 1
            messages.append(
                f"  [{label}] ({primary},{infra}) returned None "
                f"(expected {expected_module})"
            )

    # Fallback — an unmapped pair returns None so caller uses generic.
    entity = {
        "line_105_primary": "paging",
        "line_105_categories": [{"id": "paging", "infra_type": "facilities"}],
    }
    fn = handler._pick_template_generator(entity)
    if fn is None:
        passed += 1
    else:
        failed += 1
        messages.append(
            f"  [{label}] (paging,facilities) returned {fn} — expected None"
        )

    # Missing line_105_primary — must not blow up; falls back to
    # voip_interconnected which isn't in the mapping, so returns None.
    entity = {}
    try:
        fn = handler._pick_template_generator(entity)
        if fn is None:
            passed += 1
        else:
            failed += 1
            messages.append(
                f"  [{label}] empty entity → returned {fn} — expected None"
            )
    except Exception as exc:
        failed += 1
        messages.append(f"  [{label}] empty entity → raised {exc!r}")

    return passed, failed, messages


def main() -> int:
    print("=" * 72)
    print("CPNI + CALEA variant smoke test")
    print("=" * 72)

    total_passed = 0
    total_failed = 0

    # Phase 1: every mapped generator module imports and exports its fn.
    print("\nPhase 1 — generator modules import and expose their callable:")
    for label, mapping in (("CPNI", CPNI_MAPPING), ("CALEA", CALEA_MAPPING)):
        seen: set[str] = set()
        for module in mapping.values():
            if module in seen:
                continue
            seen.add(module)
            ok, detail = _import_check(module)
            status = "OK" if ok else "FAIL"
            print(f"  [{label}] {module:40s} {status} — {detail}")
            if ok:
                total_passed += 1
            else:
                total_failed += 1

    # Phase 2: handlers wire the mapping correctly.
    print("\nPhase 2 — handler._pick_template_generator returns the right fn:")
    try:
        from scripts.workers.services.cpni_certification import CPNIFilingHandler
        p, f, msgs = _variant_selection_check(
            CPNIFilingHandler, CPNI_MAPPING, "CPNI",
        )
        total_passed += p
        total_failed += f
        for m in msgs:
            print(m)
        print(f"  CPNI selection: {p} passed, {f} failed")
    except Exception as exc:
        total_failed += 1
        print(f"  CPNI handler failed to load: {exc!r}")

    try:
        from scripts.workers.services.calea_ssi import CALEASSIHandler
        p, f, msgs = _variant_selection_check(
            CALEASSIHandler, CALEA_MAPPING, "CALEA",
        )
        total_passed += p
        total_failed += f
        for m in msgs:
            print(m)
        print(f"  CALEA selection: {p} passed, {f} failed")
    except Exception as exc:
        total_failed += 1
        print(f"  CALEA handler failed to load: {exc!r}")

    print()
    print("=" * 72)
    print(f"Total: {total_passed} passed, {total_failed} failed")
    print("=" * 72)
    return 0 if total_failed == 0 else 1


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