Add route manifest contract validation

This commit is contained in:
jwinterm
2026-07-06 04:07:10 -04:00
parent c1d8337365
commit 407d4d106a
13 changed files with 619 additions and 42 deletions
+1 -1
View File
@@ -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):
+134
View File
@@ -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()