ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail when the documented Pulse API and registered HTTP routes drift apart."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MANIFEST = ROOT / "specs" / "api-routes.json"
|
|
CONTRACT = ROOT / "docs" / "architecture" / "API_CONTRACT.md"
|
|
ROUTERS = (ROOT / "cmd" / "api" / "main.go", ROOT / "internal" / "service" / "health.go")
|
|
|
|
TABLE_ROUTE = re.compile(r"^\| `(?P<method>GET|POST|PUT|PATCH|DELETE) (?P<path>/[^`]+)` \|", re.MULTILINE)
|
|
REGISTERED = re.compile(r"\.Handle(?:Func)?\(\s*\"(?P<path>/[^\"]+)\"")
|
|
|
|
|
|
def fail(messages: list[str]) -> int:
|
|
for message in messages:
|
|
print(f"API CONTRACT: FAIL: {message}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def main() -> int:
|
|
document = json.loads(MANIFEST.read_text(encoding="utf-8"))
|
|
routes = document.get("routes", [])
|
|
errors: list[str] = []
|
|
|
|
keys: list[tuple[str, str]] = []
|
|
covered_registrations: set[str] = set()
|
|
for index, route in enumerate(routes):
|
|
missing = {"method", "path", "registration", "access", "implementation"} - route.keys()
|
|
if missing:
|
|
errors.append(f"route {index} misses fields: {', '.join(sorted(missing))}")
|
|
continue
|
|
key = (route["method"], route["path"])
|
|
if key in keys:
|
|
errors.append(f"duplicate manifest route: {key[0]} {key[1]}")
|
|
keys.append(key)
|
|
covered_registrations.add(route["registration"])
|
|
implementation = ROOT / route["implementation"]
|
|
if not implementation.is_file():
|
|
errors.append(f"implementation does not exist for {key[0]} {key[1]}: {implementation.relative_to(ROOT)}")
|
|
|
|
markdown = CONTRACT.read_text(encoding="utf-8")
|
|
documented = [(match.group("method"), match.group("path")) for match in TABLE_ROUTE.finditer(markdown)]
|
|
manifest_set = set(keys)
|
|
documented_set = set(documented)
|
|
for method, path in sorted(manifest_set - documented_set):
|
|
errors.append(f"manifest route is not documented: {method} {path}")
|
|
for method, path in sorted(documented_set - manifest_set):
|
|
errors.append(f"documented route is not in manifest: {method} {path}")
|
|
if len(documented) != len(documented_set):
|
|
errors.append("the route table contains duplicate method/path rows")
|
|
|
|
registered: set[str] = set()
|
|
for source in ROUTERS:
|
|
registered.update(match.group("path") for match in REGISTERED.finditer(source.read_text(encoding="utf-8")))
|
|
for path in sorted(registered - covered_registrations):
|
|
errors.append(f"registered router path is not represented in the manifest: {path}")
|
|
for path in sorted(covered_registrations - registered):
|
|
errors.append(f"manifest registration does not exist in the router: {path}")
|
|
|
|
if errors:
|
|
return fail(errors)
|
|
print(
|
|
"API CONTRACT: PASS "
|
|
f"({len(manifest_set)} method/path contracts, {len(registered)} router registrations, no drift)"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|