test(fsma): data-free tests for 24h SLA, FDA export wiring and intake routes

services/graph/tests/test_r3_recall_sla_24h.py

@@ -0,0 +1,90 @@
+"""Data-free tests for the 24-hour recall SLA under 21 CFR 1.1455(c)(1).
+
+21 CFR 1.1455(c)(1) requires traceability records to be provided to FDA
+within 24 hours of a request. ``RecallDrill.sla_status`` encodes that limit
+as ``sla_limit = 24 * 3600``. These tests use timestamps only; no records,
+CSV rows or EPCIS documents are involved.
+"""
+
+from __future__ import annotations
+
+import ast
+import inspect
+import textwrap
+from datetime import datetime, timedelta, timezone
+
+from services.graph.app.fsma_recall import models
+from services.graph.app.fsma_recall.models import (
+    RecallDrill,
+    RecallSeverity,
+    RecallStatus,
+    RecallType,
+    SLAStatus,
+)
+
+LIMIT = timedelta(hours=24)
+
+
+def _drill(**kwargs) -> RecallDrill:
+    return RecallDrill(
+        drill_id="",
+        tenant_id="",
+        created_at=datetime.now(timezone.utc),
+        drill_type=list(RecallType)[0],
+        severity=list(RecallSeverity)[0],
+        **kwargs,
+    )
+
+
+def test_sla_limit_is_24_hours_in_source():
+    """21 CFR 1.1455(c)(1): the limit literal is ``24 * 3600`` seconds."""
+    src = textwrap.dedent(inspect.getsource(RecallDrill.sla_status.fget))
+    assigns = {
+        n.targets[0].id: n.value
+        for n in ast.walk(ast.parse(src))
+        if isinstance(n, ast.Assign) and isinstance(n.targets[0], ast.Name)
+    }
+    assert "sla_limit" in assigns
+    assert eval(compile(ast.Expression(assigns["sla_limit"]), "<sla>", "eval")) == 24 * 3600
+
+
+def test_sla_status_enum_values():
+    """21 CFR 1.1455(c)(1): MET / AT_RISK / BREACHED states exist."""
+    assert {s.value for s in SLAStatus} >= {"met", "at_risk", "breached"}
+    assert models.SLAStatus is SLAStatus
+
+
+def test_not_started_is_met():
+    """21 CFR 1.1455(c)(1): the clock has not started without started_at."""
+    assert _drill().sla_status == SLAStatus.MET
+
+
+def test_completed_exactly_24h_is_met():
+    """21 CFR 1.1455(c)(1): completing at exactly 24h meets the limit."""
+    start = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    d = _drill(status=RecallStatus.COMPLETED, started_at=start, completed_at=start + LIMIT)
+    assert d.sla_status == SLAStatus.MET
+
+
+def test_completed_one_second_over_24h_is_breached():
+    """21 CFR 1.1455(c)(1): one second past 24h breaches the limit."""
+    start = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    d = _drill(
+        status=RecallStatus.COMPLETED,
+        started_at=start,
+        completed_at=start + LIMIT + timedelta(seconds=1),
+    )
+    assert d.sla_status == SLAStatus.BREACHED
+
+
+def test_in_progress_thresholds():
+    """21 CFR 1.1455(c)(1): in-progress drills go MET -> AT_RISK (>75%) -> BREACHED (>24h)."""
+    now = datetime.now(timezone.utc)
+    cases = [
+        (timedelta(hours=1), SLAStatus.MET),
+        (timedelta(hours=19), SLAStatus.AT_RISK),
+        (LIMIT + timedelta(hours=1), SLAStatus.BREACHED),
+    ]
+    for elapsed, expected in cases:
+        d = _drill(status=RecallStatus.IN_PROGRESS, started_at=now - elapsed)
+        assert d.sla_status == expected, elapsed

services/ingestion/tests/test_r3_fda_export_intake_wiring.py

@@ -0,0 +1,172 @@
+"""Data-free wiring tests for FDA export and intake formats.
+
+21 CFR 1.1455(c)(1) requires records to be provided to FDA within 24 hours,
+and 21 CFR 1.1455(c)(3)(ii) requires an electronic sortable spreadsheet on
+request. These tests parse source with ``ast`` and check only constants,
+signatures, routing and wiring. They load no records, CSV rows or EPCIS docs.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+SERVICE = Path(__file__).resolve().parents[1]
+APP = SERVICE / "app"
+
+
+def _tree(rel: str) -> ast.Module:
+    return ast.parse((SERVICE / rel).read_text(encoding="utf-8"))
+
+
+def _router_prefix(tree: ast.Module) -> str:
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Assign)
+            and isinstance(node.targets[0], ast.Name)
+            and node.targets[0].id == "router"
+            and isinstance(node.value, ast.Call)
+            and getattr(node.value.func, "id", None) == "APIRouter"
+        ):
+            for kw in node.value.keywords:
+                if kw.arg == "prefix":
+                    return ast.literal_eval(kw.value)
+    raise AssertionError("router = APIRouter(prefix=...) not found")
+
+
+def _routes(tree: ast.Module) -> dict[tuple[str, str], ast.FunctionDef]:
+    out = {}
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            for dec in node.decorator_list:
+                if (
+                    isinstance(dec, ast.Call)
+                    and isinstance(dec.func, ast.Attribute)
+                    and getattr(dec.func.value, "id", None) == "router"
+                    and dec.args
+                ):
+                    out[(dec.func.attr, ast.literal_eval(dec.args[0]))] = node
+    return out
+
+
+def _main_includes() -> dict[str, str | None]:
+    """Map imported router alias -> _router_enabled flag guarding it (None = always)."""
+    tree = _tree("main.py")
+    aliases: dict[str, str | None] = {}
+
+    def visit(body, flag):
+        for node in body:
+            if isinstance(node, ast.If):
+                t = node.test
+                f = flag
+                if isinstance(t, ast.Call) and getattr(t.func, "id", None) == "_router_enabled":
+                    f = ast.literal_eval(t.args[0])
+                visit(node.body, f)
+            elif isinstance(node, ast.ImportFrom):
+                for a in node.names:
+                    aliases[a.asname or a.name] = flag
+
+    visit(tree.body, None)
+    included = set()
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "include_router"
+            and node.args
+            and isinstance(node.args[0], ast.Name)
+        ):
+            included.add(node.args[0].id)
+    return {k: v for k, v in aliases.items() if k in included}
+
+
+def test_fda_export_router_registered_and_compliance_critical():
+    """21 CFR 1.1455(c)(1): the FDA export router is mounted and cannot be silently disabled."""
+    includes = _main_includes()
+    assert includes.get("fda_export_router") == "fda_export"
+    src = (SERVICE / "main.py").read_text(encoding="utf-8")
+    assert "from app.fda_export.router import router as fda_export_router" in src
+    tree = _tree("main.py")
+    compliance = next(
+        n.value
+        for n in ast.walk(tree)
+        if isinstance(n, ast.Assign)
+        and isinstance(n.targets[0], ast.Name)
+        and n.targets[0].id == "_COMPLIANCE_ROUTERS"
+    )
+    assert "fda_export" in ast.literal_eval(compliance)
+
+
+def test_fda_spreadsheet_endpoint_path_method_and_service():
+    """21 CFR 1.1455(c)(3)(ii): GET /api/v1/fda/export builds the sortable CSV via _generate_csv."""
+    tree = _tree("app/fda_export/router.py")
+    assert _router_prefix(tree) == "/api/v1/fda"
+    routes = _routes(tree)
+    fn = routes[("get", "/export")]
+    assert fn.name == "export_fda_spreadsheet"
+    called = {
+        getattr(c.func, "id", None) for c in ast.walk(fn) if isinstance(c, ast.Call)
+    }
+    assert "_generate_csv" in called
+    imported = {
+        a.name
+        for n in ast.walk(tree)
+        if isinstance(n, ast.ImportFrom) and n.module == "fda_export_service"
+        for a in n.names
+    }
+    assert "_generate_csv" in imported
+
+
+def test_fda_columns_constant_drives_csv_headers():
+    """21 CFR 1.1455(c)(3)(ii): FDA_COLUMNS are the spreadsheet headers used by _generate_csv."""
+    tree = _tree("app/fda_export_service.py")
+    cols = next(
+        ast.literal_eval(n.value)
+        for n in tree.body
+        if isinstance(n, ast.Assign)
+        and isinstance(n.targets[0], ast.Name)
+        and n.targets[0].id == "FDA_COLUMNS"
+    )
+    assert len(cols) == len(set(cols))
+    for header in (
+        "Traceability Lot Code (TLC)",
+        "Product Description",
+        "Quantity",
+        "Unit of Measure",
+        "Event Type (CTE)",
+        "Event Date",
+        "Location GLN",
+        "Immediate Previous Source",
+        "TLC Source GLN",
+        "Reference Document Number",
+        "System Entry Timestamp",
+    ):
+        assert header in cols, header
+    gen = next(
+        n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "_generate_csv"
+    )
+    assert "include_pii" in {a.arg for a in gen.args.args + gen.args.kwonlyargs}
+    assert any(
+        isinstance(n, ast.keyword)
+        and n.arg == "fieldnames"
+        and getattr(n.value, "id", None) == "FDA_COLUMNS"
+        for n in ast.walk(gen)
+    )
+
+
+def test_supported_intake_formats_are_routed():
+    """21 CFR 1.1455(c)(1): webhook, CSV and EPCIS intake paths are all mounted."""
+    includes = _main_includes()
+    assert "webhook_router" in includes and includes["webhook_router"] is None
+    assert includes.get("csv_router") == "csv"
+    assert includes.get("epcis_ingestion_router") == "epcis_ingestion"
+
+    expected = {
+        "webhook": ("app/webhook_router_v2/routes.py", "/api/v1/webhooks", ("post", "/ingest")),
+        "csv": ("app/csv_templates.py", "/api/v1", ("post", "/ingest/csv")),
+        "epcis": ("app/epcis/router.py", "/api/v1/epcis", ("post", "/events")),
+    }
+    for fmt, (rel, prefix, route) in expected.items():
+        tree = _tree(rel)
+        assert _router_prefix(tree) == prefix, fmt
+        assert route in _routes(tree), fmt