Expand fixture manifests with route policies

ampg/config.py

@@ -48,6 +48,23 @@ class ProtocolConfig:
     options: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(frozen=True)
+class RoutePolicyConfig:
+    match: str
+    tier: str
+    identity: str = "none"
+    payments: str = "none"
+    realtime: bool = False
+    public_allowed: bool = True
+
+
+@dataclass(frozen=True)
+class InteractionConfig:
+    default_tier: str = "static"
+    deny_routes: tuple[str, ...] = ()
+    routes: tuple[RoutePolicyConfig, ...] = ()
+
+
 @dataclass(frozen=True)
 class SiteConfig:
     id: str
@@ -55,6 +72,7 @@ class SiteConfig:
     source: SourceConfig
     outputs: OutputConfig
     protocols: dict[str, ProtocolConfig]
+    interactions: InteractionConfig
 
 
 @dataclass(frozen=True)
@@ -76,6 +94,7 @@ def _parse_site(raw_site: dict[str, Any], base_dir: Path) -> SiteConfig:
     source = raw_site.get("source") or {}
     outputs = raw_site.get("outputs") or {}
     protocols = raw_site.get("protocols") or {}
+    interactions = raw_site.get("interactions") or {}
 
     site_id = _required_str(raw_site, "id")
     domain = _required_str(raw_site, "domain")
@@ -103,6 +122,7 @@ def _parse_site(raw_site: dict[str, Any], base_dir: Path) -> SiteConfig:
         ),
         outputs=OutputConfig(root=output_root, plan_root=plan_root),
         protocols=parsed_protocols,
+        interactions=_parse_interactions(interactions),
     )
 
 
@@ -124,6 +144,35 @@ def _parse_protocol(name: str, raw_protocol: dict[str, Any]) -> ProtocolConfig:
     )
 
 
+def _parse_interactions(raw_interactions: dict[str, Any]) -> InteractionConfig:
+    raw_routes = raw_interactions.get("route", ())
+    if isinstance(raw_routes, dict):
+        raw_routes = (raw_routes,)
+    if not isinstance(raw_routes, (list, tuple)):
+        raise ValueError("site.interactions.route must be a table array")
+    deny_routes = raw_interactions.get("deny_routes", ())
+    if not isinstance(deny_routes, (list, tuple)):
+        raise ValueError("site.interactions.deny_routes must be an array")
+    return InteractionConfig(
+        default_tier=str(raw_interactions.get("default_tier", "static")),
+        deny_routes=tuple(str(route) for route in deny_routes),
+        routes=tuple(_parse_route_policy(raw_route) for raw_route in raw_routes),
+    )
+
+
+def _parse_route_policy(raw_route: dict[str, Any]) -> RoutePolicyConfig:
+    if not isinstance(raw_route, dict):
+        raise ValueError("site.interactions.route entries must be tables")
+    return RoutePolicyConfig(
+        match=_required_str(raw_route, "match"),
+        tier=str(raw_route.get("tier", "static")),
+        identity=str(raw_route.get("identity", "none")),
+        payments=str(raw_route.get("payments", "none")),
+        realtime=bool(raw_route.get("realtime", False)),
+        public_allowed=bool(raw_route.get("public_allowed", True)),
+    )
+
+
 def _resolve_path(base_dir: Path, raw_path: str) -> Path:
     path = Path(raw_path).expanduser()
     if not path.is_absolute():

ampg/manifest.py

@@ -5,7 +5,7 @@ import json
 from pathlib import Path
 from typing import Any
 
-from .config import GatewayConfig, ProtocolConfig, SiteConfig
+from .config import GatewayConfig, ProtocolConfig, RoutePolicyConfig, SiteConfig
 
 
 SCHEMA = "ampg.fixture-manifest.v1"
@@ -19,11 +19,16 @@ class ManifestWriteResult:
 
 
 def fixture_manifest(site: SiteConfig) -> dict[str, Any]:
-    fixtures = [
-        _fixture_entry(site, protocol)
-        for protocol in site.protocols.values()
-        if protocol.enabled
-    ]
+    fixtures = []
+    for protocol in site.protocols.values():
+        if not protocol.enabled:
+            continue
+        fixtures.append(_fixture_entry(site, protocol, None))
+        fixtures.extend(
+            _fixture_entry(site, protocol, route_policy)
+            for route_policy in site.interactions.routes
+            if _route_is_public(route_policy)
+        )
     return {
         "schema": SCHEMA,
         "site": {
@@ -59,21 +64,31 @@ def manifest_path(site: SiteConfig) -> Path:
     return site.outputs.root / "ampg-fixture-manifest.json"
 
 
-def _fixture_entry(site: SiteConfig, protocol: ProtocolConfig) -> dict[str, Any]:
+def _fixture_entry(
+    site: SiteConfig,
+    protocol: ProtocolConfig,
+    route_policy: RoutePolicyConfig | None,
+) -> dict[str, Any]:
     url, address_status = _fixture_url(site, protocol)
     transport = _transport_for(protocol.name)
-    return {
+    entry = {
         "protocol": protocol.name,
         "renderer": protocol.renderer,
-        "url": url,
+        "url": _url_for_route(url, route_policy),
         "address_status": address_status,
-        "output_path": protocol.name,
+        "output_path": _output_path_for_route(protocol.name, route_policy),
         "checks": {
             "transport": transport,
             "profile": transport,
         },
-        "interaction": _interaction_policy(protocol),
+        "interaction": _interaction_policy(protocol, route_policy),
     }
+    if route_policy:
+        entry["route"] = {
+            "match": route_policy.match,
+            "fixture_path": _fixture_path(route_policy.match),
+        }
+    return entry
 
 
 def _transport_for(protocol_name: str) -> str:
@@ -82,7 +97,22 @@ def _transport_for(protocol_name: str) -> str:
     return protocol_name
 
 
-def _interaction_policy(protocol: ProtocolConfig) -> dict[str, Any]:
+def _route_is_public(route_policy: RoutePolicyConfig) -> bool:
+    return route_policy.public_allowed and route_policy.tier != "internal"
+
+
+def _interaction_policy(
+    protocol: ProtocolConfig,
+    route_policy: RoutePolicyConfig | None,
+) -> dict[str, Any]:
+    if route_policy:
+        return {
+            "tier": route_policy.tier,
+            "identity": route_policy.identity,
+            "payments": route_policy.payments,
+            "realtime": route_policy.realtime,
+            "public_allowed": route_policy.public_allowed,
+        }
     return {
         "tier": str(protocol.options.get("tier", protocol.options.get("max_tier", "static"))),
         "identity": str(protocol.options.get("identity", "none")),
@@ -92,6 +122,31 @@ def _interaction_policy(protocol: ProtocolConfig) -> dict[str, Any]:
     }
 
 
+def _url_for_route(base_url: str, route_policy: RoutePolicyConfig | None) -> str:
+    if not route_policy:
+        return base_url
+    fixture_path = _fixture_path(route_policy.match)
+    if base_url.startswith(("rns://", "lxmf://", "nomad://", "ipfs://", "ipns://")):
+        return base_url
+    return base_url.rstrip("/") + fixture_path
+
+
+def _output_path_for_route(protocol_name: str, route_policy: RoutePolicyConfig | None) -> str:
+    if not route_policy:
+        return protocol_name
+    return protocol_name + _fixture_path(route_policy.match)
+
+
+def _fixture_path(match: str) -> str:
+    value = match.strip()
+    if not value.startswith("/"):
+        value = "/" + value
+    value = value.replace("*", "")
+    if not value or value == "/":
+        return "/"
+    return value if value.endswith("/") else value + "/"
+
+
 def _fixture_url(site: SiteConfig, protocol: ProtocolConfig) -> tuple[str, str]:
     configured_url = protocol.options.get("fixture_url") or protocol.options.get("browser_url")
     if configured_url: