mz_chatbot / scripts /check_contract.py
Shurem's picture
initial commit
ff3c2e9
Raw History Blame Contribute Delete
5.18 kB
"""
check_contract.py β€” static conformance check against the frontend contract.
Compares every response model in app/schemas/ with the corresponding interface
in outreach-frontend/types/, and every route in app/api/ with the paths
outreach-frontend/lib/api/ actually calls. Pure stdlib: it parses both trees, so
it runs without installing anything and without a database.
uv run python scripts/check_contract.py (or: python3 scripts/check_contract.py)
Exit code is the number of mismatches.
"""
from __future__ import annotations
import ast
import re
import sys
from pathlib import Path
BACKEND = Path(__file__).resolve().parent.parent
FRONTEND = BACKEND.parent / "outreach-frontend"
PAIRS = [
("types/lead.ts", "Lead", "schemas/lead.py", "LeadOut"),
("types/problem.ts", "Problem", "schemas/problem.py", "ProblemOut"),
("types/problem.ts", "ProblemComment", "schemas/problem.py", "ProblemCommentOut"),
("types/metrics.ts", "CrmMetrics", "schemas/metrics.py", "CrmMetrics"),
("types/metrics.ts", "WeekStats", "schemas/metrics.py", "WeekStats"),
("types/metrics.ts", "TeamMetrics", "schemas/metrics.py", "TeamMetrics"),
("types/metrics.ts", "SystemStatus", "schemas/metrics.py", "SystemStatus"),
("types/scan.ts", "ScanResult", "schemas/scan.py", "ScanResult"),
("types/scan.ts", "DecisionMaker", "schemas/scan.py", "DecisionMaker"),
("types/scan.ts", "LinkedInResearch", "schemas/scan.py", "LinkedInResearch"),
("types/api.ts", "Job", "schemas/job.py", "JobOut"),
]
def ts_interface_fields(rel: str, name: str) -> list[str]:
src = (FRONTEND / rel).read_text(encoding="utf-8")
m = re.search(rf"export interface {name}(?:<[^>]*>)?\s*\{{(.*?)\n\}}", src, re.S)
if not m:
raise SystemExit(f"interface {name} not found in {rel}")
body = re.sub(r"//.*", "", re.sub(r"/\*.*?\*/", "", m.group(1), flags=re.S))
return re.findall(r"^\s*([a-z_][A-Za-z0-9_]*)\??\s*:", body, re.M)
def py_model_fields(rel: str, cls: str) -> list[str]:
tree = ast.parse((BACKEND / "app" / rel).read_text(encoding="utf-8"))
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef) and node.name == cls:
return [s.target.id for s in node.body if isinstance(s, ast.AnnAssign)]
raise SystemExit(f"class {cls} not found in {rel}")
def backend_routes() -> set[tuple[str, str]]:
routes: set[tuple[str, str]] = set()
for path in (BACKEND / "app" / "api").glob("*.py"):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for dec in node.decorator_list:
if (
isinstance(dec, ast.Call)
and isinstance(dec.func, ast.Attribute)
and dec.func.attr in {"get", "post", "patch", "delete"}
and dec.args
and isinstance(dec.args[0], ast.Constant)
):
routes.add((dec.func.attr.upper(), _normalise(dec.args[0].value)))
return routes
def frontend_routes() -> set[tuple[str, str]]:
"""Pulls api.get/post/patch/delete("...") calls out of lib/api/*.ts."""
routes: set[tuple[str, str]] = set()
for path in (FRONTEND / "lib" / "api").glob("*.ts"):
src = path.read_text(encoding="utf-8")
for verb, raw in re.findall(
r"api\.(get|post|patch|delete)<[^>]*>\(\s*[`\"']([^`\"']+)", src
):
routes.add((verb.upper(), _normalise(raw)))
return routes
def _normalise(p: str) -> str:
"""/leads/${id} and /leads/{lead_id} both become /leads/:id."""
p = re.sub(r"\$\{[^}]+\}", ":id", p)
p = re.sub(r"\{[^}]+\}", ":id", p)
return "/" + p.strip("/")
def main() -> int:
failures = 0
print("── response shapes ──")
for ts_rel, iface, py_rel, cls in PAIRS:
want, got = ts_interface_fields(ts_rel, iface), py_model_fields(py_rel, cls)
missing = [f for f in want if f not in got]
extra = [f for f in got if f not in want]
if missing or extra:
failures += 1
print(f"DIFF {iface:18s} missing={missing} extra={extra}")
else:
print(f"OK {iface:18s} {len(want)} fields")
print("\n── routes ──")
be, fe = backend_routes(), frontend_routes()
# /jobs/{discover|scan|import} is one client call with a variable kind slug.
fe.discard(("POST", "/jobs/:id"))
for kind in ("discover", "scan", "import"):
fe.add(("POST", f"/jobs/{kind}"))
for verb, path in sorted(fe):
if (verb, path) in be:
print(f"OK {verb:6s} {path}")
else:
failures += 1
print(f"MISS {verb:6s} {path} β€” frontend calls it, backend has no route")
unused = sorted(be - fe)
if unused:
print("\nbackend routes the frontend does not call (fine, just noting):")
for verb, path in unused:
print(f" {verb:6s} {path}")
print(f"\nmismatches: {failures}")
return failures
if __name__ == "__main__":
sys.exit(main())