feat: register contract-driven API routes

README.md

@@ -3,9 +3,9 @@
 Stateful asynchronous Proxmox VE API simulator for testing API clients and
 infrastructure tooling without a real hypervisor cluster.
 
-The project is in its foundation stage. Only liveness and PostgreSQL-backed
+The runnable foundation and authoritative contract toolchain are implemented.
-readiness endpoints exist. No Proxmox endpoint is claimed as compatible yet; the
+Imported methods can be registered dynamically, but no stateful Proxmox method
-official contract importer and stateful vertical slice are tracked in
+is claimed as compatible yet; the vertical slice is tracked in
 [the implementation plan](docs/implementation-plan.md).
 
 ## Development
@@ -49,6 +49,12 @@ for CI policy checks:
   --format markdown
 ```
 
+Set `CONTRACT_SNAPSHOT` to a normalized snapshot file to register its methods
+under both `/api2/json` and `/api2/extjs`. Routes without a semantic handler
+return an explicit 501 by default. `CONTRACT_FALLBACK=schema-default` enables
+schema-only exploration; `fixture` serves only values explicitly embedded in a
+method contract.
+
 See [the architecture](docs/architecture.md) for component boundaries and
 durability decisions. Commands for not-yet-implemented milestones intentionally
 return a non-zero status instead of pretending to succeed.

app/api/registry.py

@@ -0,0 +1,115 @@
+"""Contract-driven dynamic route and semantic handler registry."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from app.contracts.model import Method, Schema, Snapshot
+
+Handler = Callable[[Request, dict[str, Any]], Awaitable[Any]]
+FallbackMode = Literal["error", "schema-default", "fixture"]
+
+
+class RouteCollisionError(ValueError):
+    pass
+
+
+@dataclass(slots=True)
+class HandlerRegistry:
+    _handlers: dict[tuple[str, str], Handler] = field(default_factory=dict)
+
+    def register(self, path: str, verb: str, handler: Handler) -> None:
+        key = (path, verb.upper())
+        if key in self._handlers:
+            raise RouteCollisionError(f"duplicate semantic handler: {verb} {path}")
+        self._handlers[key] = handler
+
+    def get(self, path: str, verb: str) -> Handler | None:
+        return self._handlers.get((path, verb.upper()))
+
+
+def register_contract_routes(
+    app: FastAPI,
+    snapshot: Snapshot,
+    handlers: HandlerRegistry,
+    fallback: FallbackMode = "error",
+) -> None:
+    seen: set[tuple[str, str, str]] = set()
+    for contract_path in snapshot.paths:
+        for contract_method in contract_path.methods:
+            for renderer in ("json", "extjs"):
+                route = f"/api2/{renderer}{contract_path.path}"
+                key = (route, contract_method.verb, renderer)
+                if key in seen:
+                    raise RouteCollisionError(
+                        f"duplicate contract route: {contract_method.verb} {route}"
+                    )
+                seen.add(key)
+                endpoint = _endpoint(
+                    contract_path.path,
+                    contract_method,
+                    renderer,
+                    handlers,
+                    fallback,
+                )
+                app.add_api_route(
+                    route,
+                    endpoint,
+                    methods=[contract_method.verb],
+                    name=f"contract:{renderer}:{contract_method.verb}:{contract_path.path}",
+                    openapi_extra={"x-proxmox-method-checksum": contract_method.checksum},
+                )
+
+
+def _endpoint(
+    semantic_path: str,
+    method: Method,
+    renderer: str,
+    handlers: HandlerRegistry,
+    fallback: FallbackMode,
+) -> Callable[[Request], Awaitable[JSONResponse]]:
+    async def dispatch(request: Request) -> JSONResponse:
+        handler = handlers.get(semantic_path, method.verb)
+        inputs = {
+            "path": dict(request.path_params),
+            "query": dict(request.query_params),
+        }
+        if handler is not None:
+            data = await handler(request, inputs)
+        elif fallback == "schema-default":
+            data = _schema_default(method.returns)
+        elif fallback == "fixture" and "fixture" in method.extra:
+            data = method.extra["fixture"]
+        else:
+            return JSONResponse(
+                status_code=501,
+                content={"data": None, "errors": "method semantics are not implemented"},
+            )
+        if renderer == "extjs":
+            return JSONResponse({"data": data, "success": True})
+        return JSONResponse({"data": data})
+
+    return dispatch
+
+
+def _schema_default(schema: Schema) -> Any:
+    if schema.default is not None:
+        return schema.default
+    if schema.type == "array":
+        return []
+    if schema.type == "object":
+        return {
+            name: _schema_default(definition)
+            for name, definition in schema.properties.items()
+            if not definition.optional
+        }
+    if schema.type == "boolean":
+        return False
+    if schema.type in {"integer", "number"}:
+        return 0
+    return None

app/config.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from functools import lru_cache
+from pathlib import Path
+from typing import Literal
 
 from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -30,6 +32,8 @@ class Settings(BaseSettings):
     db_command_timeout_seconds: float = Field(default=30.0, gt=0, le=300)
     log_level: str = "INFO"
     request_id_header: str = "X-Request-ID"
+    contract_snapshot: Path | None = None
+    contract_fallback: Literal["error", "schema-default", "fixture"] = "error"
 
 
 @lru_cache(maxsize=1)

app/main.py

@@ -6,7 +6,9 @@ from fastapi import FastAPI
 
 from app.api.errors import unhandled_exception_handler
 from app.api.middleware import RequestContextMiddleware
+from app.api.registry import HandlerRegistry, register_contract_routes
 from app.config import Settings, get_settings
+from app.contracts.model import Snapshot
 from app.lifespan import DatabaseFactory, create_lifespan, default_database_factory
 from app.logging import configure_logging
 from app.observability.health import router as health_router
@@ -15,6 +17,7 @@ from app.observability.health import router as health_router
 def create_app(
     settings: Settings | None = None,
     database_factory: DatabaseFactory = default_database_factory,
+    handlers: HandlerRegistry | None = None,
 ) -> FastAPI:
     """Create an isolated application instance with explicit resource factories."""
 
@@ -28,6 +31,14 @@ def create_app(
     app.add_middleware(RequestContextMiddleware, header_name=resolved.request_id_header)
     app.add_exception_handler(Exception, unhandled_exception_handler)
     app.include_router(health_router)
+    if resolved.contract_snapshot is not None:
+        snapshot = Snapshot.model_validate_json(resolved.contract_snapshot.read_bytes())
+        register_contract_routes(
+            app,
+            snapshot,
+            handlers or HandlerRegistry(),
+            resolved.contract_fallback,
+        )
     return app
 
 

tests/unit/test_dynamic_routes.py

@@ -0,0 +1,90 @@
+"""Contract-driven route registry tests."""
+
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi import Request
+from httpx import ASGITransport, AsyncClient
+from pydantic import ValidationError
+
+from app.api.registry import HandlerRegistry, RouteCollisionError
+from app.config import Settings
+from app.contracts.model import Method, PathContract, Schema, Snapshot
+from app.main import create_app
+from tests.unit.test_health import FakeDatabase
+
+
+def contract_snapshot(*methods: Method) -> Snapshot:
+    paths = (PathContract(path="/version", methods=methods),)
+    return Snapshot(
+        source_version="test",
+        retrieved_at=datetime(2026, 1, 1, tzinfo=UTC),
+        raw_sha256="0" * 64,
+        paths=paths,
+        path_count=1,
+        method_count=len(methods),
+    )
+
+
+def get_method() -> Method:
+    return Method(
+        verb="GET",
+        name="version",
+        returns=Schema(type="object", properties={"version": Schema(type="string")}),
+        checksum="1" * 64,
+    )
+
+
+async def request_app(
+    tmp_path: Path, fallback: str, handlers: HandlerRegistry | None = None
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    snapshot_path = tmp_path / "snapshot.json"
+    snapshot_path.write_bytes(contract_snapshot(get_method()).canonical_bytes())
+    settings = Settings(contract_snapshot=snapshot_path, contract_fallback=fallback)
+    database = FakeDatabase(True)
+    app = create_app(settings, lambda _settings: database, handlers)
+    async with app.router.lifespan_context(app):
+        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+            json_response = await client.get("/api2/json/version")
+            extjs_response = await client.get("/api2/extjs/version")
+    return json_response.json(), extjs_response.json()
+
+
+async def test_registered_handler_serves_both_renderers(tmp_path: Path) -> None:
+    handlers = HandlerRegistry()
+
+    async def version(_request: Request, _inputs: dict[str, Any]) -> dict[str, str]:
+        return {"version": "9.2.3"}
+
+    handlers.register("/version", "GET", version)
+
+    json_body, extjs_body = await request_app(tmp_path, "error", handlers)
+
+    assert json_body == {"data": {"version": "9.2.3"}}
+    assert extjs_body == {"data": {"version": "9.2.3"}, "success": True}
+
+
+async def test_explicit_fallback_modes(tmp_path: Path) -> None:
+    error_body, _ = await request_app(tmp_path, "error")
+    default_body, _ = await request_app(tmp_path, "schema-default")
+
+    assert error_body["errors"] == "method semantics are not implemented"
+    assert default_body == {"data": {"version": None}}
+
+
+def test_duplicate_snapshot_routes_are_rejected() -> None:
+    with pytest.raises(ValidationError, match="duplicate"):
+        contract_snapshot(get_method(), get_method())
+
+
+def test_duplicate_semantic_handlers_are_rejected() -> None:
+    handlers = HandlerRegistry()
+
+    async def handler(_request: Request, _inputs: dict[str, Any]) -> None:
+        return None
+
+    handlers.register("/version", "GET", handler)
+    with pytest.raises(RouteCollisionError, match="duplicate"):
+        handlers.register("/version", "GET", handler)