diff --git a/README.md b/README.md index 831896a..7ceb099 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/docs/configuration.md b/docs/configuration.md index b189681..9e0b80b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -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", diff --git a/docs/interactive-apps.md b/docs/interactive-apps.md index 96e2584..952f69b 100644 --- a/docs/interactive-apps.md +++ b/docs/interactive-apps.md @@ -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`. diff --git a/examples/route-catalog.json b/examples/route-catalog.json new file mode 100644 index 0000000..5319dac --- /dev/null +++ b/examples/route-catalog.json @@ -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 + } + ] +} diff --git a/examples/route-manifest.json b/examples/route-manifest.json index 9a9f752..b8510c1 100644 --- a/examples/route-manifest.json +++ b/examples/route-manifest.json @@ -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" } diff --git a/tests/test_route_manifest.py b/tests/test_route_manifest.py index 96968f0..55f44b8 100644 --- a/tests/test_route_manifest.py +++ b/tests/test_route_manifest.py @@ -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( { diff --git a/tools/generate_route_manifest.py b/tools/generate_route_manifest.py new file mode 100644 index 0000000..ce99731 --- /dev/null +++ b/tools/generate_route_manifest.py @@ -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())