docqube_backend/scripts/endpoint_inventory.py

"""
B0.1.3 — classify every route as freeze / skip / websocket.

Introspects the live FastAPI application rather than grepping for decorators,
so the count cannot drift from what the app actually serves and nothing hidden
behind a conditional include is missed.

Writes tests/characterization/inventory.json, which is committed and read by
test_inventory.py. When the inventory changes, that test fails and the diff
shows exactly which endpoint appeared or vanished.

    APP_ENV=rework_test python scripts/endpoint_inventory.py
"""

import json
import os
import sys
from pathlib import Path

os.environ.setdefault("APP_ENV", "rework_test")
os.environ.setdefault("PYTHONIOENCODING", "utf-8")

sys.path.insert(0, str(Path(__file__).parent.parent))

OUTPUT = Path(__file__).parent.parent / "tests" / "characterization" / "inventory.json"

# Reasons an endpoint is not worth freezing. Order matters: the first match wins.
SKIP_RULES = [
    ("webhook", "Inbound webhook — the contract belongs to Zoho/DocuSeal, not us"),
    ("/api/health", "Liveness probe, no data"),
    ("/docs", "Generated documentation"),
    ("/redoc", "Generated documentation"),
    ("/openapi.json", "Generated schema — covered by the B0.3 contract gate"),
]

# Paths that stream bytes rather than return JSON. Worth an existence and auth
# test, but recording a byte-for-byte body is not characterization, it is a
# fixture with extra steps.
BINARY_HINTS = ("/download", "/preview", "/thumbnail", "/image", "/stream", "/file/")


# Modules that own at least one of the 15 tenant-scoped tables. A mutation in
# one of these can move data across a tenant boundary, so it is worth freezing
# before B1 rewrites how scoping works. A mutation elsewhere cannot, so it is
# deferred rather than skipped — see D9 in the plan.
TENANT_OWNING_MODULES = {
    "activity_logs",
    "auth",
    "chat",
    "documents",
    "drive",
    "notifications",
    "signing",
    "storage",
    "tenant",
}


def classify(path: str, method: str, module: str) -> tuple[str, str]:
    lowered = path.lower()

    for needle, reason in SKIP_RULES:
        if needle in lowered:
            return "skip", reason

    if any(hint in lowered for hint in BINARY_HINTS):
        return "freeze-shallow", "Streams bytes — freeze status and headers, not the body"

    if method == "GET":
        return "freeze", "Read endpoint"

    if module in TENANT_OWNING_MODULES:
        return "freeze", f"Mutation in {module}, which owns tenant-scoped tables"

    return "defer", (
        f"Mutation in {module}, which owns no tenant-scoped table — "
        "covered after B1, not before it"
    )


def module_of(path: str, endpoint) -> str:
    mod = getattr(endpoint, "__module__", "") or ""
    if ".modules." in mod:
        return mod.split(".modules.")[1].split(".")[0]
    if mod.startswith("app.api"):
        return "_api"
    if mod.startswith("app.tasks"):
        return "_tasks"
    if mod.startswith("app.ai_service"):
        return "_ai_service"
    if mod.startswith("app.main"):
        return "_main"
    return "_other"


def main() -> int:
    from fastapi.routing import APIRoute, APIWebSocketRoute

    from app.main import app

    records = []
    for route in app.routes:
        path = getattr(route, "path", None)
        if not path:
            continue

        endpoint = getattr(route, "endpoint", None)

        if isinstance(route, APIWebSocketRoute):
            records.append(
                {
                    "path": path,
                    "method": "WEBSOCKET",
                    "module": module_of(path, endpoint),
                    "classification": "websocket",
                    "reason": "Bidirectional — probed by B0.2.6, not frozen",
                    "response_model": False,
                }
            )
            continue

        if not isinstance(route, APIRoute):
            continue

        has_model = getattr(route, "response_model", None) is not None
        module = module_of(path, endpoint)
        for method in sorted(route.methods or []):
            if method in {"HEAD", "OPTIONS"}:
                continue
            classification, reason = classify(path, method, module)
            records.append(
                {
                    "path": path,
                    "method": method,
                    "module": module_of(path, endpoint),
                    "classification": classification,
                    "reason": reason,
                    "response_model": has_model,
                }
            )

    records.sort(key=lambda r: (r["module"], r["path"], r["method"]))

    totals = {}
    for r in records:
        totals[r["classification"]] = totals.get(r["classification"], 0) + 1

    by_module = {}
    for r in records:
        m = by_module.setdefault(r["module"], {"total": 0, "response_model": 0})
        m["total"] += 1
        m["response_model"] += 1 if r["response_model"] else 0

    payload = {
        "_comment": (
            "Generated by scripts/endpoint_inventory.py (B0.1.3). Do not hand-edit. "
            "Regenerate and commit the diff when routes change."
        ),
        "totals": {
            "endpoints": len(records),
            "by_classification": dict(sorted(totals.items())),
            "with_response_model": sum(1 for r in records if r["response_model"]),
            "missing_response_model": sum(1 for r in records if not r["response_model"]),
        },
        "by_module": dict(sorted(by_module.items())),
        "endpoints": records,
    }

    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")

    print(f"wrote {OUTPUT.relative_to(Path.cwd())}")
    print(f"  endpoints           {len(records)}")
    for k, v in sorted(totals.items()):
        print(f"  {k:<20}{v}")
    print(f"  response_model      {payload['totals']['with_response_model']}")
    print(f"  missing             {payload['totals']['missing_response_model']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())