127 lines
4.4 KiB
Python
Executable File
127 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Fail when checked-in API, event, MCP or n8n contracts drift from the code."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from app.main import app
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def fail(message: str, failures: list[str]) -> None:
|
|
failures.append(message)
|
|
|
|
|
|
def check_openapi(failures: list[str]) -> None:
|
|
committed = yaml.safe_load((ROOT / "contracts/openapi.yaml").read_text(encoding="utf-8"))
|
|
generated = app.openapi()
|
|
if committed != generated:
|
|
fail(
|
|
"contracts/openapi.yaml differs from app.openapi(); regenerate it with "
|
|
"scripts/generate-openapi.py",
|
|
failures,
|
|
)
|
|
|
|
|
|
def check_mcp(failures: list[str]) -> None:
|
|
contract = json.loads((ROOT / "contracts/mcp-tools.json").read_text(encoding="utf-8"))
|
|
contracted = {
|
|
(tool["endpoint"]["method"].upper(), tool["endpoint"]["path"])
|
|
for tool in contract["tools"]
|
|
if tool.get("read_only") is True and isinstance(tool.get("endpoint"), dict)
|
|
}
|
|
generated = app.openapi()
|
|
implemented = {
|
|
(method.upper(), path)
|
|
for path, operations in generated["paths"].items()
|
|
if path.startswith("/api/v1/integrations/mcp/")
|
|
for method in operations
|
|
if method.lower() in {"get", "post", "put", "patch", "delete"}
|
|
}
|
|
if len(contracted) != len(contract["tools"]):
|
|
fail("Every MCP tool must be read-only and declare its provider endpoint", failures)
|
|
if contracted != implemented:
|
|
fail(
|
|
f"MCP endpoint drift: contract={sorted(contracted)!r}, code={sorted(implemented)!r}",
|
|
failures,
|
|
)
|
|
if any(method not in {"GET", "POST"} for method, _path in contracted):
|
|
fail("MCP contract exposes an unsupported mutation method", failures)
|
|
|
|
|
|
def check_event_schema(failures: list[str]) -> None:
|
|
schema = json.loads((ROOT / "contracts/events.schema.json").read_text(encoding="utf-8"))
|
|
expected_envelope = {
|
|
"event_id",
|
|
"event_type",
|
|
"occurred_at",
|
|
"correlation_id",
|
|
"aggregate",
|
|
"data",
|
|
}
|
|
expected_data = {
|
|
"vehicle_ref",
|
|
"inspection_ref",
|
|
"resulting_vehicle_status",
|
|
"attention_reasons",
|
|
}
|
|
if set(schema.get("required", [])) != expected_envelope:
|
|
fail("Event envelope required fields drifted", failures)
|
|
data_schema = schema.get("properties", {}).get("data", {})
|
|
if set(data_schema.get("required", [])) != expected_data:
|
|
fail("vehicle.returned.v1 data fields drifted", failures)
|
|
event_type = schema.get("properties", {}).get("event_type", {}).get("const")
|
|
if event_type != "vehicle.returned.v1":
|
|
fail("Unexpected event_type contract", failures)
|
|
|
|
|
|
def check_workflows(failures: list[str]) -> None:
|
|
workflows_dir = ROOT / "n8n/workflows"
|
|
manifest = (workflows_dir / "MANIFEST.md").read_text(encoding="utf-8")
|
|
expected = {
|
|
"fleet-ops-vehicle-return.json": "mobilityops-return-processing",
|
|
"fleet-ops-data-quality-scan.json": "mobilityops-scheduled-quality-scan",
|
|
"fleet-ops-ragcore-procedure-sync.json": "6wbkc4d1AouGpmWT",
|
|
"fleet-ops-error-handler.json": "Xppn2rAEqUuyiCJF",
|
|
"fleet-ops-alert-receiver.json": "mobilityops-alert-receiver",
|
|
}
|
|
declared_hashes = set(re.findall(r"`([0-9a-f]{64})`", manifest))
|
|
for filename, workflow_id in expected.items():
|
|
path = workflows_dir / filename
|
|
raw = path.read_bytes()
|
|
digest = hashlib.sha256(raw).hexdigest()
|
|
if digest not in declared_hashes:
|
|
fail(f"{filename} checksum is missing or stale in MANIFEST.md", failures)
|
|
definition = json.loads(raw)
|
|
if definition.get("id") != workflow_id:
|
|
fail(f"{filename} has unexpected workflow id", failures)
|
|
serialized = raw.decode("utf-8")
|
|
if "http://fleetops.itworx.tech" in serialized:
|
|
fail(f"{filename} contains a cleartext Fleet Ops callback", failures)
|
|
|
|
|
|
def main() -> int:
|
|
failures: list[str] = []
|
|
check_openapi(failures)
|
|
check_mcp(failures)
|
|
check_event_schema(failures)
|
|
check_workflows(failures)
|
|
if failures:
|
|
for failure in failures:
|
|
print(f"ERROR: {failure}", file=sys.stderr)
|
|
return 1
|
|
print("OpenAPI, event, MCP and n8n contracts are synchronized.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|