125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail when a public service route is missing from docs or OpenAPI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
DOCS = Path(__file__).resolve().parents[1]
|
|
REPO = DOCS.parent
|
|
PUBLIC = DOCS / "public"
|
|
|
|
|
|
def messenger_routes() -> set[tuple[str, str]]:
|
|
source = (REPO / "micro-chat/micromsg/src/app_routes.rs").read_text(encoding="utf-8")
|
|
routes = set(re.findall(r'\("([A-Z]+)", "([^"]+)"\)\s*=>', source))
|
|
routes = {(method, path) for method, path in routes if not path.startswith("/internal/")}
|
|
if 'path.starts_with("/file/")' in source:
|
|
routes.add(("GET", "/file/{id}"))
|
|
return routes
|
|
|
|
|
|
def gateway_routes() -> set[tuple[str, str]]:
|
|
source = (REPO / "crypto-gateway/crates/gateway-api/src/main.rs").read_text(encoding="utf-8")
|
|
routes: set[tuple[str, str]] = set()
|
|
cursor = 0
|
|
while (start := source.find(".route(", cursor)) != -1:
|
|
depth = 0
|
|
end = start + len(".route(")
|
|
while end < len(source):
|
|
if source[end] == "(":
|
|
depth += 1
|
|
elif source[end] == ")":
|
|
if depth == 0:
|
|
break
|
|
depth -= 1
|
|
end += 1
|
|
block = source[start:end + 1]
|
|
match = re.search(r'\.route\(\s*"([^"]+)"\s*,(.*)\)$', block, re.S)
|
|
cursor = end + 1
|
|
if not match:
|
|
continue
|
|
path, handlers = match.groups()
|
|
if path.startswith("/v1/") or path.startswith("/admin/v1/") or path == "/health":
|
|
for method in re.findall(r'\b(get|post|delete)\s*\(', handlers):
|
|
routes.add((method.upper(), path))
|
|
web = (REPO / "crypto-gateway/crates/gateway-api/src/web.rs").read_text(encoding="utf-8")
|
|
for path in ("/pay/{token}", "/pay/{token}/status"):
|
|
if f'.route("{path}", get(' not in web:
|
|
raise AssertionError(f"hosted checkout route disappeared: {path}")
|
|
routes.add(("GET", path))
|
|
return routes
|
|
|
|
|
|
def donations_routes() -> set[tuple[str, str]]:
|
|
source = (REPO / "donations/src/main.rs").read_text(encoding="utf-8")
|
|
routes: set[tuple[str, str]] = set()
|
|
cursor = 0
|
|
while (start := source.find(".route(", cursor)) != -1:
|
|
depth = 0
|
|
end = start + len(".route(")
|
|
while end < len(source):
|
|
if source[end] == "(":
|
|
depth += 1
|
|
elif source[end] == ")":
|
|
if depth == 0:
|
|
break
|
|
depth -= 1
|
|
end += 1
|
|
block = source[start:end + 1]
|
|
match = re.search(r'\.route\(\s*"([^"]+)"\s*,(.*)\)$', block, re.S)
|
|
cursor = end + 1
|
|
if not match:
|
|
continue
|
|
path, handlers = match.groups()
|
|
if path == "/health" or path.startswith("/api/"):
|
|
for method in re.findall(r'\b(get|post|delete|put|patch)\s*\(', handlers):
|
|
routes.add((method.upper(), path))
|
|
for method in re.findall(r'\.(delete|put|patch)\s*\(', handlers):
|
|
routes.add((method.upper(), path))
|
|
return routes
|
|
|
|
|
|
def main() -> int:
|
|
manifest = json.loads((PUBLIC / "api-manifest.json").read_text(encoding="utf-8"))
|
|
documented = {
|
|
service["id"]: {(item["method"], item["path"]) for item in service["endpoints"]}
|
|
for service in manifest["services"]
|
|
}
|
|
source = {
|
|
"messenger": messenger_routes(),
|
|
"gateway": gateway_routes(),
|
|
"donations": donations_routes(),
|
|
}
|
|
failed = False
|
|
for service_id, actual in source.items():
|
|
missing = actual - documented[service_id]
|
|
stale = documented[service_id] - actual
|
|
if missing:
|
|
print(f"{service_id}: undocumented routes: {sorted(missing)}", file=sys.stderr)
|
|
failed = True
|
|
if stale:
|
|
print(f"{service_id}: stale routes: {sorted(stale)}", file=sys.stderr)
|
|
failed = True
|
|
spec = json.loads((PUBLIC / "openapi" / f"{service_id}.json").read_text(encoding="utf-8"))
|
|
in_spec = {(method.upper(), path) for path, operations in spec["paths"].items()
|
|
for method in operations if method.upper() in {"GET", "POST", "DELETE", "PUT", "PATCH"}}
|
|
if in_spec != documented[service_id]:
|
|
print(f"{service_id}: OpenAPI mismatch", file=sys.stderr)
|
|
failed = True
|
|
counted = sum(len(routes) for routes in documented.values())
|
|
if counted != manifest["total_endpoints"]:
|
|
print(f"manifest total is {manifest['total_endpoints']}, counted {counted}", file=sys.stderr)
|
|
failed = True
|
|
if failed:
|
|
return 1
|
|
print(f"verified {counted} public endpoints across {len(source)} services")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|