Agent Zero
355 lines
13 KiB
Python
355 lines
13 KiB
Python
#!/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 _router_prefixes(content: str) -> dict[str, str]:
|
|
"""Extract all router variable definitions with their APIRouter prefixes.
|
|
|
|
Modules may define multiple routers (e.g. ``router``, ``calendar_router``,
|
|
``resource_router``) each with its own prefix.
|
|
"""
|
|
prefixes: dict[str, str] = {}
|
|
for m in re.finditer(
|
|
r"(\w+)\s*=\s*APIRouter\([^)]*?prefix\s*=\s*[\"']([^\"']+)[\"']", content
|
|
):
|
|
prefixes[m.group(1)] = m.group(2).rstrip("/")
|
|
return prefixes
|
|
|
|
|
|
def find_backend_endpoints() -> list[dict]:
|
|
"""Find all backend endpoints from route decorators.
|
|
|
|
Includes the module's APIRouter prefix so paths are comparable with
|
|
frontend calls (which use full /api/v1/... URLs).
|
|
"""
|
|
endpoints = []
|
|
scan_dirs = [
|
|
BACKEND_APP / "routes",
|
|
BACKEND_APP / "plugins" / "builtins",
|
|
]
|
|
|
|
for scan_dir in scan_dirs:
|
|
if not scan_dir.exists():
|
|
continue
|
|
is_plugin_dir = scan_dir.name == "builtins"
|
|
for py_file in scan_dir.rglob("*.py"):
|
|
try:
|
|
content = py_file.read_text(encoding="utf-8")
|
|
except Exception:
|
|
continue
|
|
|
|
rel_path = str(py_file.relative_to(BACKEND_APP))
|
|
prefixes = _router_prefixes(content)
|
|
|
|
# Variable-agnostic with per-variable prefix mapping: modules may
|
|
# define multiple routers (e.g. ``calendar_router``) each with its
|
|
# own prefix. Path may be an empty string for collection routes
|
|
# (e.g. ``@router.get("")``).
|
|
for match in re.finditer(r"@(\w+)\.(get|post|put|patch|delete)\s*\(\s*[\"']([^\"']*)[\"']", content):
|
|
var_name = match.group(1)
|
|
method = match.group(2).upper()
|
|
prefix = prefixes.get(var_name, "")
|
|
path = prefix + match.group(3)
|
|
ep = {
|
|
"file": rel_path,
|
|
"method": method,
|
|
"path": path,
|
|
}
|
|
if is_plugin_dir:
|
|
ep["is_plugin"] = True
|
|
endpoints.append(ep)
|
|
|
|
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 JS template-literal interpolations (${qs}, ${id}, ...) like params;
|
|
# ensure a '/' separator so '/webhooks${params}' -> '/webhooks/{param}'
|
|
path = re.sub(r"([^/])\$\{[^}]*\}", r"\1/{param}", path)
|
|
path = re.sub(r"^\$\{[^}]*\}", "{param}", path)
|
|
# Replace {param} patterns
|
|
path = re.sub(r"\{[^}]+\}", "{param}", path)
|
|
# Collapse duplicate slashes and remove trailing slash
|
|
path = re.sub(r"//+", "/", path)
|
|
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())
|