From 407d4d106aa81d6cc9a5a27b4b59ce4e66169c3e Mon Sep 17 00:00:00 2001 From: jwinterm Date: Mon, 6 Jul 2026 04:07:10 -0400 Subject: [PATCH] Add route manifest contract validation --- README.md | 1 + ampg/capabilities.py | 35 +++ ampg/cli.py | 77 +++++++ ampg/config.py | 24 +- ampg/docsgen.py | 2 + ampg/route_manifest.py | 249 +++++++++++++++++++++ ampg/route_policy.py | 19 +- docs/autodocs.md | 7 +- docs/configuration.md | 7 + docs/interactive-apps.md | 7 + schemas/ampg.route-manifest.v1.schema.json | 97 ++++++++ tests/test_fixture_manifest.py | 2 +- tests/test_route_manifest.py | 134 +++++++++++ 13 files changed, 619 insertions(+), 42 deletions(-) create mode 100644 ampg/capabilities.py create mode 100644 ampg/route_manifest.py create mode 100644 schemas/ampg.route-manifest.v1.schema.json create mode 100644 tests/test_route_manifest.py diff --git a/README.md b/README.md index e78cd0e..831896a 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 -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 python3 -m ampg docs generate diff --git a/ampg/capabilities.py b/ampg/capabilities.py new file mode 100644 index 0000000..2e8ae07 --- /dev/null +++ b/ampg/capabilities.py @@ -0,0 +1,35 @@ +from __future__ import annotations + + +TIER_ORDER = { + "static": 0, + "interactive-lite": 1, + "identity": 2, + "transactional": 3, + "realtime": 4, + "internal": 5, +} +INTERACTION_TIERS = tuple(TIER_ORDER.keys()) + +IDENTITY_ADAPTERS = ( + "none", + "http-session", + "siwe", + "signed-link", +) + +PAYMENT_ADAPTERS = ( + "none", + "server-invoice", + "wallet-signature", + "static-instructions", +) + +DEFAULT_MAX_TIERS = { + "clearnet": "realtime", + "tor": "transactional", + "i2p": "transactional", + "gemini": "interactive-lite", + "ipfs": "static", + "reticulum": "interactive-lite", +} diff --git a/ampg/cli.py b/ampg/cli.py index 6565c56..4622493 100644 --- a/ampg/cli.py +++ b/ampg/cli.py @@ -10,6 +10,11 @@ from .config import load_config from .docsgen import generate_docs from .manifest import write_fixture_manifests from .plan import plan_gateway, write_plan_artifacts +from .route_manifest import ( + load_route_manifest, + route_manifest_schema_json, + validate_route_manifest, +) from .route_policy import RouteExposure, RouteIssue, route_exposures, route_issues @@ -37,6 +42,28 @@ def main(argv: list[str] | None = None) -> int: "validate", help="Fail when a public route has no compatible enabled protocol.", ) + manifest_parser = subcommands.add_parser( + "route-manifest", + help="Validate or print the app route manifest contract.", + ) + manifest_subcommands = manifest_parser.add_subparsers( + dest="route_manifest_command", + required=True, + ) + manifest_validate = manifest_subcommands.add_parser( + "validate", + help="Validate an ampg.route-manifest.v1 JSON file.", + ) + manifest_validate.add_argument("path", type=Path, help="Route manifest JSON path.") + manifest_schema = manifest_subcommands.add_parser( + "schema", + help="Print the JSON Schema for route manifests.", + ) + manifest_schema.add_argument( + "--output", + type=Path, + help="Write the JSON Schema to this path instead of stdout.", + ) audit_parser = subcommands.add_parser("audit", help="Audit source HTML quality.") audit_parser.add_argument( "--fail-on-warn", @@ -57,6 +84,8 @@ def main(argv: list[str] | None = None) -> int: try: if args.command == "docs": return _cmd_docs(args) + if args.command == "route-manifest": + return _cmd_route_manifest(args) config = load_config(args.config) if args.command == "plan": return _cmd_plan(config, write_artifacts=args.write_artifacts) @@ -158,6 +187,54 @@ def _cmd_routes(config, args) -> int: return 0 +def _cmd_route_manifest(args) -> int: + if args.route_manifest_command == "schema": + content = route_manifest_schema_json() + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(content, encoding="utf-8") + print(f"AMPG_ROUTE_MANIFEST_SCHEMA path={args.output}") + else: + print(content, end="") + return 0 + + if args.route_manifest_command == "validate": + try: + data = load_route_manifest(args.path) + except ValueError: + try: + import json + + data = json.loads(args.path.read_text(encoding="utf-8")) + issues = validate_route_manifest(data) + except Exception as exc: # noqa: BLE001 - CLI should print concise failures. + print( + "AMPG_ROUTE_MANIFEST " + f"path={args.path} status=error message=\"{exc}\"", + file=sys.stderr, + ) + return 1 + for issue in issues: + print( + "AMPG_ROUTE_MANIFEST_ISSUE " + f"path={args.path} json_path={issue.path} " + f"code={issue.code} message=\"{issue.message}\"" + ) + print( + "AMPG_ROUTE_MANIFEST " + f"path={args.path} status=fail issues={len(issues)}" + ) + return 1 + + print( + "AMPG_ROUTE_MANIFEST " + f"path={args.path} schema={data['schema']} routes={len(data['routes'])} status=ok" + ) + return 0 + + return 1 + + def _print_route_exposure(exposure: RouteExposure) -> None: print( "AMPG_ROUTE " diff --git a/ampg/config.py b/ampg/config.py index 0a69d81..9e501b2 100644 --- a/ampg/config.py +++ b/ampg/config.py @@ -1,13 +1,11 @@ from __future__ import annotations from dataclasses import dataclass, field -import json from pathlib import Path from typing import Any import tomllib - -ROUTE_MANIFEST_SCHEMA = "ampg.route-manifest.v1" +from .route_manifest import load_route_manifest DEFAULT_DAEMONS = { "clearnet": "nginx", @@ -158,7 +156,7 @@ def _parse_interactions( raise ValueError("site.interactions must be a table") route_manifest = _route_manifest_path(raw_interactions, base_dir) - manifest_data = _load_route_manifest(route_manifest, site_id) if route_manifest else {} + manifest_data = load_route_manifest(route_manifest, site_id=site_id) if route_manifest else {} default_tier = str( raw_interactions.get("default_tier", manifest_data.get("default_tier", "static")) @@ -209,24 +207,6 @@ def _route_manifest_path(raw_interactions: dict[str, Any], base_dir: Path) -> Pa return _resolve_path(base_dir, raw_path) -def _load_route_manifest(path: Path, site_id: str) -> dict[str, Any]: - try: - data = json.loads(path.read_text(encoding="utf-8")) - except FileNotFoundError as exc: - raise ValueError(f"{site_id}: route manifest not found: {path}") from exc - except json.JSONDecodeError as exc: - raise ValueError(f"{site_id}: invalid route manifest JSON: {path}: {exc}") from exc - if not isinstance(data, dict): - raise ValueError(f"{site_id}: route manifest must be a JSON object: {path}") - if data.get("schema") != ROUTE_MANIFEST_SCHEMA: - raise ValueError( - f"{site_id}: route manifest schema must be {ROUTE_MANIFEST_SCHEMA!r}: {path}" - ) - if "routes" not in data: - raise ValueError(f"{site_id}: route manifest missing required routes array: {path}") - return data - - def _route_array(raw_routes: Any, context: str) -> tuple[Any, ...]: if isinstance(raw_routes, dict): return (raw_routes,) diff --git a/ampg/docsgen.py b/ampg/docsgen.py index 12d7d7a..4c0fe36 100644 --- a/ampg/docsgen.py +++ b/ampg/docsgen.py @@ -11,6 +11,7 @@ from .metadata import ( RENDER_PROFILES, TRANSPORT_INTERACTION_CAPABILITIES, ) +from .route_manifest import ROUTE_MANIFEST_SCHEMA_PATH, route_manifest_schema_json GENERATED_DIR = Path("docs/generated") @@ -22,6 +23,7 @@ def generate_docs(root: Path, *, check: bool = False) -> list[Path]: GENERATED_DIR / "daemon-adapters.md": _daemon_adapters_doc(), GENERATED_DIR / "interaction-capabilities.md": _interaction_capabilities_doc(), GENERATED_DIR / "render-profiles.md": _render_profiles_doc(), + ROUTE_MANIFEST_SCHEMA_PATH: route_manifest_schema_json(), } changed: list[Path] = [] for rel_path, content in docs.items(): diff --git a/ampg/route_manifest.py b/ampg/route_manifest.py new file mode 100644 index 0000000..189d3f0 --- /dev/null +++ b/ampg/route_manifest.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +from pathlib import Path +from typing import Any + +from .capabilities import IDENTITY_ADAPTERS, INTERACTION_TIERS, PAYMENT_ADAPTERS + + +ROUTE_MANIFEST_SCHEMA = "ampg.route-manifest.v1" +ROUTE_MANIFEST_SCHEMA_PATH = Path("schemas/ampg.route-manifest.v1.schema.json") + +TOP_LEVEL_FIELDS = {"schema", "default_tier", "deny_routes", "routes"} +ROUTE_FIELDS = {"match", "tier", "identity", "payments", "realtime", "public_allowed"} + + +@dataclass(frozen=True) +class RouteManifestIssue: + path: str + code: str + message: str + + +def load_route_manifest(path: Path, *, site_id: str | None = None) -> dict[str, Any]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + label = f"{site_id}: " if site_id else "" + raise ValueError(f"{label}route manifest not found: {path}") from exc + except json.JSONDecodeError as exc: + label = f"{site_id}: " if site_id else "" + raise ValueError(f"{label}invalid route manifest JSON: {path}: {exc}") from exc + + issues = validate_route_manifest(data) + if issues: + label = f"{site_id}: " if site_id else "" + joined = "; ".join(f"{issue.path}: {issue.message}" for issue in issues) + raise ValueError(f"{label}invalid route manifest: {path}: {joined}") + return data + + +def validate_route_manifest(data: Any) -> list[RouteManifestIssue]: + issues: list[RouteManifestIssue] = [] + if not isinstance(data, dict): + return [ + RouteManifestIssue( + path="$", + code="type", + message="route manifest must be a JSON object", + ) + ] + + _unknown_fields("$", data, TOP_LEVEL_FIELDS, issues) + schema = data.get("schema") + if schema != ROUTE_MANIFEST_SCHEMA: + issues.append( + RouteManifestIssue( + path="$.schema", + code="schema", + message=f"must be {ROUTE_MANIFEST_SCHEMA!r}", + ) + ) + + if "default_tier" in data: + _enum_field("$.default_tier", data["default_tier"], INTERACTION_TIERS, issues) + + if "deny_routes" in data: + _string_array("$.deny_routes", data["deny_routes"], issues, require_slash=True) + + routes = data.get("routes") + if routes is None: + issues.append( + RouteManifestIssue( + path="$.routes", + code="required", + message="missing required routes array", + ) + ) + return issues + if not isinstance(routes, list): + issues.append( + RouteManifestIssue( + path="$.routes", + code="type", + message="must be an array", + ) + ) + return issues + + for index, route in enumerate(routes): + _route_entry(f"$.routes[{index}]", route, issues) + return issues + + +def route_manifest_json_schema() -> dict[str, Any]: + return { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "urn:ampg:schema:route-manifest:v1", + "title": "AMPG Route Manifest", + "description": "Route policy contract emitted by applications for AMPG.", + "type": "object", + "additionalProperties": False, + "required": ["schema", "routes"], + "properties": { + "schema": {"const": ROUTE_MANIFEST_SCHEMA}, + "default_tier": {"$ref": "#/$defs/interactionTier"}, + "deny_routes": { + "type": "array", + "items": {"$ref": "#/$defs/routePattern"}, + "default": [], + }, + "routes": { + "type": "array", + "items": {"$ref": "#/$defs/routePolicy"}, + }, + }, + "$defs": { + "interactionTier": {"enum": list(INTERACTION_TIERS)}, + "identityAdapter": {"enum": list(IDENTITY_ADAPTERS)}, + "paymentAdapter": {"enum": list(PAYMENT_ADAPTERS)}, + "routePattern": { + "type": "string", + "minLength": 1, + "pattern": "^/", + }, + "routePolicy": { + "type": "object", + "additionalProperties": False, + "required": ["match"], + "properties": { + "match": {"$ref": "#/$defs/routePattern"}, + "tier": {"$ref": "#/$defs/interactionTier"}, + "identity": {"$ref": "#/$defs/identityAdapter", "default": "none"}, + "payments": {"$ref": "#/$defs/paymentAdapter", "default": "none"}, + "realtime": {"type": "boolean", "default": False}, + "public_allowed": {"type": "boolean", "default": True}, + }, + }, + }, + } + + +def route_manifest_schema_json() -> str: + return json.dumps(route_manifest_json_schema(), indent=2, sort_keys=True) + "\n" + + +def _route_entry(path: str, value: Any, issues: list[RouteManifestIssue]) -> None: + if not isinstance(value, dict): + issues.append( + RouteManifestIssue(path=path, code="type", message="route entry must be an object") + ) + return + _unknown_fields(path, value, ROUTE_FIELDS, issues) + if "match" not in value: + issues.append( + RouteManifestIssue( + path=f"{path}.match", + code="required", + message="missing required route match", + ) + ) + else: + _route_pattern(f"{path}.match", value["match"], issues) + + if "tier" in value: + _enum_field(f"{path}.tier", value["tier"], INTERACTION_TIERS, issues) + if "identity" in value: + _enum_field(f"{path}.identity", value["identity"], IDENTITY_ADAPTERS, issues) + if "payments" in value: + _enum_field(f"{path}.payments", value["payments"], PAYMENT_ADAPTERS, issues) + if "realtime" in value: + _bool_field(f"{path}.realtime", value["realtime"], issues) + if "public_allowed" in value: + _bool_field(f"{path}.public_allowed", value["public_allowed"], issues) + + +def _unknown_fields( + path: str, + data: dict[str, Any], + allowed: set[str], + issues: list[RouteManifestIssue], +) -> None: + for field in sorted(set(data) - allowed): + issues.append( + RouteManifestIssue( + path=f"{path}.{field}", + code="unknown-field", + message="unknown field", + ) + ) + + +def _enum_field( + path: str, + value: Any, + allowed: tuple[str, ...], + issues: list[RouteManifestIssue], +) -> None: + if value not in allowed: + joined = ", ".join(allowed) + issues.append( + RouteManifestIssue( + path=path, + code="enum", + message=f"must be one of: {joined}", + ) + ) + + +def _string_array( + path: str, + value: Any, + issues: list[RouteManifestIssue], + *, + require_slash: bool = False, +) -> None: + if not isinstance(value, list): + issues.append(RouteManifestIssue(path=path, code="type", message="must be an array")) + return + for index, item in enumerate(value): + item_path = f"{path}[{index}]" + if require_slash: + _route_pattern(item_path, item, issues) + elif not isinstance(item, str) or not item: + issues.append( + RouteManifestIssue( + path=item_path, + code="type", + message="must be a non-empty string", + ) + ) + + +def _route_pattern(path: str, value: Any, issues: list[RouteManifestIssue]) -> None: + if not isinstance(value, str) or not value: + issues.append( + RouteManifestIssue(path=path, code="type", message="must be a non-empty string") + ) + return + if not value.startswith("/"): + issues.append( + RouteManifestIssue(path=path, code="pattern", message="must start with /") + ) + + +def _bool_field(path: str, value: Any, issues: list[RouteManifestIssue]) -> None: + if not isinstance(value, bool): + issues.append(RouteManifestIssue(path=path, code="type", message="must be boolean")) diff --git a/ampg/route_policy.py b/ampg/route_policy.py index 07f7f54..9ed07c1 100644 --- a/ampg/route_policy.py +++ b/ampg/route_policy.py @@ -3,27 +3,10 @@ from __future__ import annotations from dataclasses import dataclass from fnmatch import fnmatchcase +from .capabilities import DEFAULT_MAX_TIERS, TIER_ORDER from .config import GatewayConfig, ProtocolConfig, RoutePolicyConfig, SiteConfig -TIER_ORDER = { - "static": 0, - "interactive-lite": 1, - "identity": 2, - "transactional": 3, - "realtime": 4, - "internal": 5, -} -DEFAULT_MAX_TIERS = { - "clearnet": "realtime", - "tor": "transactional", - "i2p": "transactional", - "gemini": "interactive-lite", - "ipfs": "static", - "reticulum": "interactive-lite", -} - - @dataclass(frozen=True) class RouteExposure: site_id: str diff --git a/docs/autodocs.md b/docs/autodocs.md index 160c30f..118c65c 100644 --- a/docs/autodocs.md +++ b/docs/autodocs.md @@ -19,6 +19,10 @@ code. Generated artifacts are committed, then checked in CI by regenerating and : Generated from renderer definitions. Lists stripped tags, asset limits, link modes, and defaults for each target. +`schemas/ampg.route-manifest.v1.schema.json` +: Generated from the route-manifest validator. App repos can use it to emit and validate + `ampg.route-manifest.v1` without depending on AMPG internals. + `openapi.json` : Generated if AMPG exposes a local control API for status, builds, plans, and daemon health. @@ -30,7 +34,7 @@ CI should run: ```sh python3 -m ampg docs generate python3 -m ampg docs generate --check -git diff --exit-code docs/generated openapi.json +git diff --exit-code docs/generated schemas openapi.json python3 tools/docs_check.py ``` @@ -58,5 +62,6 @@ outside the public repo. - [ ] Generate config schema docs from code. - [ ] Generate daemon adapter docs from registered adapters. - [ ] Generate renderer docs from renderer defaults. +- [ ] Generate route-manifest JSON Schema from code. - [ ] Commit generated artifacts. - [ ] Fail CI when regeneration produces a diff. diff --git a/docs/configuration.md b/docs/configuration.md index 81c5702..b189681 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -203,6 +203,13 @@ Applications can generate a JSON route manifest instead of hand-maintaining ever group in TOML. Paths are resolved relative to `gateway.toml`; inline TOML routes are appended after manifest routes. +The public schema is committed at `schemas/ampg.route-manifest.v1.schema.json`. +Applications can validate generated manifests before AMPG config parsing: + +```sh +python3 -m ampg route-manifest validate routes.json +``` + ```json { "schema": "ampg.route-manifest.v1", diff --git a/docs/interactive-apps.md b/docs/interactive-apps.md index e66b4d8..96e2584 100644 --- a/docs/interactive-apps.md +++ b/docs/interactive-apps.md @@ -55,6 +55,13 @@ Apps may provide `ampg.route-manifest.v1` JSON. AMPG imports it from `site.interactions.route_manifest`, merges its denylist with inline TOML, and appends any inline TOML route entries after the imported route entries. +The schema is generated at `schemas/ampg.route-manifest.v1.schema.json`, and app repos can +validate generated manifests with: + +```sh +python3 -m ampg route-manifest validate routes.json +``` + Required top-level fields: - `schema`: must be `ampg.route-manifest.v1`. diff --git a/schemas/ampg.route-manifest.v1.schema.json b/schemas/ampg.route-manifest.v1.schema.json new file mode 100644 index 0000000..22a880f --- /dev/null +++ b/schemas/ampg.route-manifest.v1.schema.json @@ -0,0 +1,97 @@ +{ + "$defs": { + "identityAdapter": { + "enum": [ + "none", + "http-session", + "siwe", + "signed-link" + ] + }, + "interactionTier": { + "enum": [ + "static", + "interactive-lite", + "identity", + "transactional", + "realtime", + "internal" + ] + }, + "paymentAdapter": { + "enum": [ + "none", + "server-invoice", + "wallet-signature", + "static-instructions" + ] + }, + "routePattern": { + "minLength": 1, + "pattern": "^/", + "type": "string" + }, + "routePolicy": { + "additionalProperties": false, + "properties": { + "identity": { + "$ref": "#/$defs/identityAdapter", + "default": "none" + }, + "match": { + "$ref": "#/$defs/routePattern" + }, + "payments": { + "$ref": "#/$defs/paymentAdapter", + "default": "none" + }, + "public_allowed": { + "default": true, + "type": "boolean" + }, + "realtime": { + "default": false, + "type": "boolean" + }, + "tier": { + "$ref": "#/$defs/interactionTier" + } + }, + "required": [ + "match" + ], + "type": "object" + } + }, + "$id": "urn:ampg:schema:route-manifest:v1", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "description": "Route policy contract emitted by applications for AMPG.", + "properties": { + "default_tier": { + "$ref": "#/$defs/interactionTier" + }, + "deny_routes": { + "default": [], + "items": { + "$ref": "#/$defs/routePattern" + }, + "type": "array" + }, + "routes": { + "items": { + "$ref": "#/$defs/routePolicy" + }, + "type": "array" + }, + "schema": { + "const": "ampg.route-manifest.v1" + } + }, + "required": [ + "schema", + "routes" + ], + "title": "AMPG Route Manifest", + "type": "object" +} diff --git a/tests/test_fixture_manifest.py b/tests/test_fixture_manifest.py index dc8a8cf..e606545 100644 --- a/tests/test_fixture_manifest.py +++ b/tests/test_fixture_manifest.py @@ -300,7 +300,7 @@ renderer = "privacy-html" encoding="utf-8", ) - with self.assertRaisesRegex(ValueError, "route manifest schema"): + with self.assertRaisesRegex(ValueError, "ampg.route-manifest.v1"): load_config(config_path) def test_manifest_omits_routes_above_protocol_max_tier(self): diff --git a/tests/test_route_manifest.py b/tests/test_route_manifest.py new file mode 100644 index 0000000..96968f0 --- /dev/null +++ b/tests/test_route_manifest.py @@ -0,0 +1,134 @@ +import contextlib +import io +import json +import tempfile +import unittest +from pathlib import Path + +from ampg.capabilities import IDENTITY_ADAPTERS, INTERACTION_TIERS, PAYMENT_ADAPTERS +from ampg.cli import main +from ampg.metadata import ( + IDENTITY_ADAPTERS as METADATA_IDENTITY_ADAPTERS, + INTERACTION_TIERS as METADATA_INTERACTION_TIERS, + PAYMENT_ADAPTERS as METADATA_PAYMENT_ADAPTERS, +) +from ampg.route_manifest import ( + route_manifest_schema_json, + validate_route_manifest, +) + + +ROOT = Path(__file__).resolve().parents[1] + + +class RouteManifestTest(unittest.TestCase): + def test_example_route_manifest_is_valid(self): + data = json.loads((ROOT / "examples/route-manifest.json").read_text(encoding="utf-8")) + + self.assertEqual([], validate_route_manifest(data)) + + def test_invalid_route_manifest_reports_all_contract_issues(self): + issues = validate_route_manifest( + { + "schema": "wrong", + "default_tier": "magic", + "deny_routes": ["admin/*"], + "extra": True, + "routes": [ + { + "match": "play/*", + "tier": "paid", + "identity": "cookie-ish", + "payments": "coin", + "realtime": "yes", + "public_allowed": "no", + "extra": True, + }, + [], + ], + } + ) + + by_path = {issue.path: issue for issue in issues} + self.assertEqual("unknown-field", by_path["$.extra"].code) + self.assertEqual("schema", by_path["$.schema"].code) + self.assertEqual("enum", by_path["$.default_tier"].code) + self.assertEqual("pattern", by_path["$.deny_routes[0]"].code) + self.assertEqual("unknown-field", by_path["$.routes[0].extra"].code) + self.assertEqual("pattern", by_path["$.routes[0].match"].code) + self.assertEqual("enum", by_path["$.routes[0].tier"].code) + self.assertEqual("enum", by_path["$.routes[0].identity"].code) + self.assertEqual("enum", by_path["$.routes[0].payments"].code) + self.assertEqual("type", by_path["$.routes[0].realtime"].code) + self.assertEqual("type", by_path["$.routes[0].public_allowed"].code) + self.assertEqual("type", by_path["$.routes[1]"].code) + + def test_route_manifest_validate_cli(self): + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + status = main(["route-manifest", "validate", "examples/route-manifest.json"]) + + self.assertEqual(0, status) + self.assertIn("schema=ampg.route-manifest.v1 routes=4 status=ok", stdout.getvalue()) + + def test_route_manifest_validate_cli_reports_issues(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "routes.json" + path.write_text( + json.dumps( + { + "schema": "ampg.route-manifest.v1", + "routes": [{"match": "play/*", "tier": "paid"}], + } + ), + encoding="utf-8", + ) + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + status = main(["route-manifest", "validate", str(path)]) + + output = stdout.getvalue() + self.assertEqual(1, status) + self.assertIn("json_path=$.routes[0].match code=pattern", output) + self.assertIn("json_path=$.routes[0].tier code=enum", output) + self.assertIn("status=fail issues=2", output) + + def test_route_manifest_schema_file_matches_generated_schema(self): + schema_path = ROOT / "schemas/ampg.route-manifest.v1.schema.json" + + self.assertEqual( + route_manifest_schema_json(), + schema_path.read_text(encoding="utf-8"), + ) + + def test_route_manifest_schema_cli_can_write_schema(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "schema.json" + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + status = main(["route-manifest", "schema", "--output", str(path)]) + + self.assertEqual(0, status) + self.assertEqual(route_manifest_schema_json(), path.read_text(encoding="utf-8")) + self.assertIn(f"AMPG_ROUTE_MANIFEST_SCHEMA path={path}", stdout.getvalue()) + + def test_capability_constants_match_generated_metadata(self): + self.assertEqual( + tuple(tier.name for tier in METADATA_INTERACTION_TIERS), + INTERACTION_TIERS, + ) + self.assertEqual( + tuple(adapter.name for adapter in METADATA_IDENTITY_ADAPTERS), + IDENTITY_ADAPTERS, + ) + self.assertEqual( + tuple(adapter.name for adapter in METADATA_PAYMENT_ADAPTERS), + PAYMENT_ADAPTERS, + ) + + +if __name__ == "__main__": + unittest.main()