leocrm/scripts/test_suite/trace_api_contracts.py

#!/usr/bin/env python3
"""Trace API Contracts — Find Frontend↔Backend Verkabelungsfehler.

Compares every frontend API call with the backend response fields.
Mismatches = Verkabelungsfehler.

Usage: python3 scripts/test_suite/trace_api_contracts.py
"""
from __future__ import annotations

import ast
import json
import os
import re
import sys
from collections import defaultdict
from pathlib import Path

# Project root
ROOT = Path(__file__).resolve().parent.parent.parent
FRONTEND_SRC = ROOT / "frontend" / "src"
BACKEND_APP = ROOT / "app"


def find_frontend_api_calls() -> list[dict]:
    """Find all API calls in frontend source."""
    calls = []
    api_dir = FRONTEND_SRC / "api"
    if not api_dir.exists():
        return calls

    for ts_file in api_dir.rglob("*.ts"):
        try:
            content = ts_file.read_text(encoding="utf-8")
        except Exception:
            continue

        rel_path = str(ts_file.relative_to(FRONTEND_SRC))

        # Find apiGet/apiPost/apiPut/apiPatch/apiDelete calls with URL patterns
        patterns = [
            (r"apiGet(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "GET"),
            (r"apiPost(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "POST"),
            (r"apiPut(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "PUT"),
            (r"apiPatch(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "PATCH"),
            (r"apiDelete(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "DELETE"),
            (r"\bM\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "GET"),
            (r"\bj\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "POST"),
            (r"\bSt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "PUT"),
            (r"\byt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "PATCH"),
            (r"\bxt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*", "DELETE"),
        ]

        for pattern, method in patterns:
            for match in re.finditer(pattern, content):
                url = match.group(1)
                # Normalize URL — remove template literals, query params
                url_clean = re.sub(r"\$\{[^}]+\}", "{param}", url)
                url_clean = url_clean.split("?")[0].split("#")[0]

                calls.append({
                    "file": rel_path,
                    "method": method,
                    "url": url_clean,
                    "raw_url": url,
                })

    return calls


def find_frontend_expected_fields() -> dict[str, list[str]]:
    """Find TypeScript interfaces/types that define expected API response fields."""
    fields_by_url = defaultdict(list)
    api_dir = FRONTEND_SRC / "api"
    if not api_dir.exists():
        return fields_by_url

    for ts_file in api_dir.rglob("*.ts"):
        try:
            content = ts_file.read_text(encoding="utf-8")
        except Exception:
            continue

        # Find interface definitions
        for match in re.finditer(r"interface\s+(\w+)\s*\{([^}]+)\}", content):
            iface_name = match.group(1)
            body = match.group(2)
            field_names = re.findall(r"(\w+)\s*[?:]", body)
            fields_by_url[iface_name] = field_names

    return fields_by_url


def find_backend_response_fields() -> list[dict]:
    """Find backend response fields from routes."""
    responses = []
    routes_dir = BACKEND_APP / "routes"
    if not routes_dir.exists():
        return responses

    for py_file in routes_dir.rglob("*.py"):
        try:
            content = py_file.read_text(encoding="utf-8")
        except Exception:
            continue

        rel_path = str(py_file.relative_to(BACKEND_APP))

        # Find response_model definitions
        for match in re.finditer(r"response_model\s*=\s*(\w+)", content):
            responses.append({
                "file": rel_path,
                "response_model": match.group(1),
            })

        # Find content= dict definitions (inline responses)
        for match in re.finditer(r"content\s*=\s*\{([^}]+)\}", content, re.DOTALL):
            body = match.group(1)
            fields = re.findall(r"['\"](\w+)['\"]\s*:", body)
            if fields:
                responses.append({
                    "file": rel_path,
                    "fields": fields,
                    "type": "inline",
                })

        # Find return dict patterns
        for match in re.finditer(r"return\s*\{([^}]+)\}", content, re.DOTALL):
            body = match.group(1)
            fields = re.findall(r"['\"](\w+)['\"]\s*:", body)
            if fields:
                responses.append({
                    "file": rel_path,
                    "fields": fields,
                    "type": "return_dict",
                })

    return responses


def find_backend_endpoints() -> list[dict]:
    """Find all backend endpoints from route decorators."""
    endpoints = []
    routes_dir = BACKEND_APP / "routes"
    if not routes_dir.exists():
        return endpoints

    for py_file in routes_dir.rglob("*.py"):
        try:
            content = py_file.read_text(encoding="utf-8")
        except Exception:
            continue

        rel_path = str(py_file.relative_to(BACKEND_APP))

        # Find router decorators
        for match in re.finditer(r"@router\.(get|post|put|patch|delete)\s*\(\s*[\"']([^\"']+)[\"']", content):
            method = match.group(1).upper()
            path = match.group(2)
            endpoints.append({
                "file": rel_path,
                "method": method,
                "path": path,
            })

    # Also find plugin routes
    plugins_dir = BACKEND_APP / "plugins" / "builtins"
    if plugins_dir.exists():
        for py_file in plugins_dir.rglob("*.py"):
            try:
                content = py_file.read_text(encoding="utf-8")
            except Exception:
                continue

            rel_path = str(py_file.relative_to(BACKEND_APP))
            for match in re.finditer(r"@router\.(get|post|put|patch|delete)\s*\(\s*[\"']([^\"']+)[\"']", content):
                method = match.group(1).upper()
                path = match.group(2)
                endpoints.append({
                    "file": rel_path,
                    "method": method,
                    "path": path,
                    "is_plugin": True,
                })

    return endpoints


def normalize_path(path: str) -> str:
    """Normalize a path for comparison."""
    # Remove leading /api/v1 if present
    path = re.sub(r"^/api/v1/", "/", path)
    # Replace {param} patterns
    path = re.sub(r"\{[^}]+\}", "{param}", path)
    # Remove trailing slash
    path = path.rstrip("/")
    return path


def check_frontend_calls_have_backend(frontend_calls, backend_endpoints) -> list[dict]:
    """Check if every frontend API call has a matching backend endpoint."""
    backend_paths = set()
    for ep in backend_endpoints:
        normalized = normalize_path(ep["path"])
        backend_paths.add((ep["method"], normalized))

    mismatches = []
    for call in frontend_calls:
        url = call["url"]
        # Remove /api/v1 prefix if present
        url_normalized = normalize_path(url)
        key = (call["method"], url_normalized)

        if key not in backend_paths:
            # Try without method (some endpoints might have different method mapping)
            found = False
            for bm, bp in backend_paths:
                if bp == url_normalized:
                    found = True
                    break
            if not found:
                mismatches.append({
                    "type": "frontend_call_no_backend",
                    "method": call["method"],
                    "url": url,
                    "file": call["file"],
                    "severity": "HIGH",
                    "message": f"Frontend calls {call['method']} {url} but no matching backend endpoint found",
                })

    return mismatches


def check_backend_endpoints_have_frontend(frontend_calls, backend_endpoints) -> list[dict]:
    """Check if every backend endpoint is called by the frontend."""
    frontend_paths = set()
    for call in frontend_calls:
        url_normalized = normalize_path(call["url"])
        frontend_paths.add((call["method"], url_normalized))

    mismatches = []
    for ep in backend_endpoints:
        normalized = normalize_path(ep["path"])
        key = (ep["method"], normalized)

        if key not in frontend_paths:
            mismatches.append({
                "type": "backend_endpoint_no_frontend",
                "method": ep["method"],
                "path": ep["path"],
                "file": ep["file"],
                "severity": "LOW",
                "message": f"Backend endpoint {ep['method']} {ep['path']} is never called by frontend",
            })

    return mismatches


def main():
    print("=" * 70)
    print("TRACE API CONTRACTS — Frontend ↔ Backend Verkabelungsfehler")
    print("=" * 70)

    print("\n[1] Scanning frontend API calls...")
    frontend_calls = find_frontend_api_calls()
    print(f"    Found {len(frontend_calls)} API calls in frontend")

    print("\n[2] Scanning backend endpoints...")
    backend_endpoints = find_backend_endpoints()
    print(f"    Found {len(backend_endpoints)} endpoints in backend")

    print("\n[3] Scanning backend response fields...")
    backend_responses = find_backend_response_fields()
    print(f"    Found {len(backend_responses)} response definitions")

    print("\n[4] Scanning frontend expected fields...")
    frontend_fields = find_frontend_expected_fields()
    print(f"    Found {len(frontend_fields)} TypeScript interfaces")

    print("\n[5] Checking frontend calls have matching backend...")
    missing_backend = check_frontend_calls_have_backend(frontend_calls, backend_endpoints)
    print(f"    Found {len(missing_backend)} frontend calls without backend endpoint")

    print("\n[6] Checking backend endpoints are called by frontend...")
    missing_frontend = check_backend_endpoints_have_frontend(frontend_calls, backend_endpoints)
    print(f"    Found {len(missing_frontend)} backend endpoints never called by frontend")

    # Summary
    all_issues = missing_backend + missing_frontend

    print("\n" + "=" * 70)
    print("SUMMARY")
    print("=" * 70)
    print(f"Frontend API calls:     {len(frontend_calls)}")
    print(f"Backend endpoints:      {len(backend_endpoints)}")
    print(f"Response definitions:   {len(backend_responses)}")
    print(f"TS interfaces:          {len(frontend_fields)}")
    print(f"\nISSUES FOUND: {len(all_issues)}")
    print(f"  HIGH (frontend calls missing backend):  {len(missing_backend)}")
    print(f"  LOW (backend endpoints not called):     {len(missing_frontend)}")

    if missing_backend:
        print("\n--- HIGH SEVERITY: Frontend calls without backend ---")
        for m in missing_backend[:30]:
            print(f"  [{m['method']}] {m['url']}  (in {m['file']})")
        if len(missing_backend) > 30:
            print(f"  ... and {len(missing_backend) - 30} more")

    if missing_frontend:
        print("\n--- LOW SEVERITY: Backend endpoints never called by frontend ---")
        for m in missing_frontend[:30]:
            print(f"  [{m['method']}] {m['path']}  (in {m['file']})")
        if len(missing_frontend) > 30:
            print(f"  ... and {len(missing_frontend) - 30} more")

    # Write results to JSON
    results_file = ROOT / "scripts" / "test_suite" / "results_trace_api_contracts.json"
    with open(results_file, "w") as f:
        json.dump({
            "frontend_calls": frontend_calls,
            "backend_endpoints": backend_endpoints,
            "issues": all_issues,
            "summary": {
                "total_frontend_calls": len(frontend_calls),
                "total_backend_endpoints": len(backend_endpoints),
                "total_issues": len(all_issues),
                "high_severity": len(missing_backend),
                "low_severity": len(missing_frontend),
            },
        }, f, indent=2, default=str)
    print(f"\nResults written to {results_file}")

    return len(all_issues)


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