Add route catalog manifest generator

This commit is contained in:
jwinterm
2026-07-06 04:16:09 -04:00
parent 407d4d106a
commit 38d86adfbf
7 changed files with 200 additions and 10 deletions
+1
View File
@@ -105,6 +105,7 @@ python3 -m ampg --config examples/wownero.gateway.toml build
python3 -m ampg --config examples/wownero.gateway.toml manifest
python3 -m ampg --config examples/wownero.gateway.toml routes explain
python3 -m ampg --config examples/wownero.gateway.toml routes validate
python3 tools/generate_route_manifest.py examples/route-catalog.json examples/route-manifest.json
python3 -m ampg route-manifest validate examples/route-manifest.json
python3 -m ampg --config examples/wownero.gateway.toml audit
python3 -m ampg --config examples/i2p-only.gateway.toml build
+5
View File
@@ -207,9 +207,14 @@ The public schema is committed at `schemas/ampg.route-manifest.v1.schema.json`.
Applications can validate generated manifests before AMPG config parsing:
```sh
python3 tools/generate_route_manifest.py examples/route-catalog.json examples/route-manifest.json
python3 -m ampg route-manifest validate routes.json
```
`examples/route-catalog.json` is a neutral app-side input shape. Real framework adapters
can replace it with generated route data from the application's router, then emit the same
AMPG route-manifest contract.
```json
{
"schema": "ampg.route-manifest.v1",
+4
View File
@@ -59,9 +59,13 @@ The schema is generated at `schemas/ampg.route-manifest.v1.schema.json`, and app
validate generated manifests with:
```sh
python3 tools/generate_route_manifest.py examples/route-catalog.json examples/route-manifest.json
python3 -m ampg route-manifest validate routes.json
```
`examples/route-catalog.json` shows the adapter side of the workflow: app route metadata
goes in, the stable AMPG route manifest comes out.
Required top-level fields:
- `schema`: must be `ampg.route-manifest.v1`.
+38
View File
@@ -0,0 +1,38 @@
{
"default_tier": "interactive-lite",
"deny_routes": [
"/admin/*",
"/api/internal/*",
"/webhooks/*"
],
"routes": [
{
"name": "play",
"method": "GET",
"path": "/play/*"
},
{
"name": "checkout",
"method": "POST",
"path": "/checkout/*",
"tier": "transactional",
"identity": "http-session",
"payments": "server-invoice"
},
{
"name": "live",
"method": "GET",
"path": "/live/*",
"tier": "realtime",
"identity": "http-session",
"realtime": true
},
{
"name": "admin",
"method": "GET",
"path": "/admin/*",
"tier": "internal",
"public_allowed": false
}
]
}
+10 -10
View File
@@ -1,5 +1,4 @@
{
"schema": "ampg.route-manifest.v1",
"default_tier": "interactive-lite",
"deny_routes": [
"/admin/*",
@@ -11,21 +10,22 @@
"match": "/play/*"
},
{
"match": "/checkout/*",
"tier": "transactional",
"identity": "http-session",
"payments": "server-invoice"
"match": "/checkout/*",
"payments": "server-invoice",
"tier": "transactional"
},
{
"match": "/live/*",
"tier": "realtime",
"identity": "http-session",
"realtime": true
"match": "/live/*",
"realtime": true,
"tier": "realtime"
},
{
"match": "/admin/*",
"tier": "internal",
"public_allowed": false
"public_allowed": false,
"tier": "internal"
}
]
],
"schema": "ampg.route-manifest.v1"
}
+41
View File
@@ -1,6 +1,8 @@
import contextlib
import io
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
@@ -27,6 +29,45 @@ class RouteManifestTest(unittest.TestCase):
self.assertEqual([], validate_route_manifest(data))
def test_route_catalog_generator_matches_example_manifest(self):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "route-manifest.json"
result = subprocess.run(
[
sys.executable,
"tools/generate_route_manifest.py",
"examples/route-catalog.json",
str(output),
],
cwd=ROOT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual("", result.stderr)
self.assertEqual(0, result.returncode)
self.assertEqual(
(ROOT / "examples/route-manifest.json").read_text(encoding="utf-8"),
output.read_text(encoding="utf-8"),
)
check_result = subprocess.run(
[
sys.executable,
"tools/generate_route_manifest.py",
"examples/route-catalog.json",
str(output),
"--check",
],
cwd=ROOT,
check=False,
capture_output=True,
text=True,
)
self.assertEqual("", check_result.stderr)
self.assertEqual(0, check_result.returncode)
def test_invalid_route_manifest_reports_all_contract_issues(self):
issues = validate_route_manifest(
{
+101
View File
@@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""Generate an AMPG route manifest from a small app route catalog."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from ampg.route_manifest import ROUTE_MANIFEST_SCHEMA, validate_route_manifest # noqa: E402
PASSTHROUGH_FIELDS = ("tier", "identity", "payments", "realtime", "public_allowed")
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("catalog", type=Path, help="Input app route catalog JSON.")
parser.add_argument("output", type=Path, help="Output ampg.route-manifest.v1 JSON.")
parser.add_argument(
"--check",
action="store_true",
help="Fail if output differs instead of writing it.",
)
args = parser.parse_args(argv)
try:
catalog = _load_json(args.catalog)
manifest = route_manifest_from_catalog(catalog)
content = json.dumps(manifest, indent=2, sort_keys=True) + "\n"
if args.check:
existing = args.output.read_text(encoding="utf-8")
if existing != content:
print(
f"ROUTE_MANIFEST_GENERATE status=stale output={args.output}",
file=sys.stderr,
)
return 1
else:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(content, encoding="utf-8")
print(
"ROUTE_MANIFEST_GENERATE "
f"status=ok catalog={args.catalog} output={args.output} routes={len(manifest['routes'])}"
)
return 0
except Exception as exc: # noqa: BLE001 - CLI should print concise failures.
print(f"ROUTE_MANIFEST_GENERATE status=error message=\"{exc}\"", file=sys.stderr)
return 1
def route_manifest_from_catalog(catalog: dict[str, Any]) -> dict[str, Any]:
if not isinstance(catalog, dict):
raise ValueError("catalog must be a JSON object")
routes = catalog.get("routes")
if not isinstance(routes, list):
raise ValueError("catalog.routes must be an array")
manifest: dict[str, Any] = {
"schema": ROUTE_MANIFEST_SCHEMA,
"routes": [_route_policy(route, index) for index, route in enumerate(routes)],
}
if "default_tier" in catalog:
manifest["default_tier"] = catalog["default_tier"]
if "deny_routes" in catalog:
manifest["deny_routes"] = catalog["deny_routes"]
issues = validate_route_manifest(manifest)
if issues:
joined = "; ".join(f"{issue.path}: {issue.message}" for issue in issues)
raise ValueError(f"generated route manifest is invalid: {joined}")
return manifest
def _route_policy(route: Any, index: int) -> dict[str, Any]:
if not isinstance(route, dict):
raise ValueError(f"catalog.routes[{index}] must be an object")
path = route.get("path", route.get("match"))
if not isinstance(path, str) or not path:
raise ValueError(f"catalog.routes[{index}].path must be a non-empty string")
policy: dict[str, Any] = {"match": path}
for field in PASSTHROUGH_FIELDS:
if field in route:
policy[field] = route[field]
return policy
def _load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
if __name__ == "__main__":
sys.exit(main())