fix(imports): 3 broken imports — ENTITY_MODELS path, Room→CommConversation, get_cached_mail_summary→MailService
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
This commit is contained in:
@@ -0,0 +1,347 @@
|
||||
#!/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*",
|
||||
r"apiPost(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"apiPut(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"apiPatch(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"apiDelete(?:<[^>]+>)?\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"\bM\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"\bj\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"\bSt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"\byt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
r"\bxt\s*\(\s*[`'\"]([^`'\"]+)[`'\"]\s*",
|
||||
]
|
||||
|
||||
for pattern 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]
|
||||
# Determine method from function name
|
||||
method = "GET"
|
||||
if "Post" in pattern or "\bj\s*" in pattern:
|
||||
method = "POST"
|
||||
elif "Put" in pattern:
|
||||
method = "PUT"
|
||||
elif "Patch" in pattern or "\byt\s*" in pattern:
|
||||
method = "PATCH"
|
||||
elif "Delete" in pattern or "\bxt\s*" in pattern:
|
||||
method = "DELETE"
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user