AFI conformance test suite

Substrate-level gate: same @client fixture registered in both backends
emits equivalent schemas, therefore the codegen produces equivalent
TypeScript regardless of which backend the frontend is generated against.
Catches adapter symmetry problems (Pydantic→OpenAPI converter divergence,
metadata leakage, ordering non-determinism) without docker, browser, or
Playwright.

What ships:

backends/mizan-fastapi/src/mizan_fastapi/schema.py — build_schema():
- Builds OpenAPI 3.0 from registered Mizan functions, mirroring the
  shape mizan-django's export emits.
- Drives FastAPI's native OpenAPI generation by registering a stub POST
  endpoint per function with its Input/Output Pydantic models, then
  appends x-mizan-functions and x-mizan-contexts extensions.
- Param-elevation logic mirrors mizan-django/src/mizan/export/__init__.py
  exactly (sharedBy tracking, required iff every function in context has
  the param).
- snake_to_camel and metadata field shapes match Django for byte-equality
  on the AFI surface.

tests/afi/ — the conformance harness:
- fixture.py: 5 @client functions covering the protocol axes (plain,
  context, mutation+affects). No channels/forms — those aren't AFI-common.
- django_app/: minimal Django project (settings, urls, AppConfig.ready
  registers the fixture). manage.py adds tests/afi/ to sys.path so both
  backends import the same fixture module.
- fastapi_app.py: thin make_app() that registers fixture and mounts router.
- schema_normalizer.py: drops backend-specific framing — Ninja-vs-FastAPI
  envelope differences (info/servers/tags), Django-only function fields
  (form metadata), x-mizan-channels. Plus afi_subset() and
  function_io_schemas() helpers for narrower comparisons.
- test_codegen_parity.py: three gates
  1. x-mizan-functions match across backends
  2. x-mizan-contexts match across backends
  3. Per-function Input/Output OpenAPI schemas match (what codegen feeds
     to openapi-typescript for type generation)

The full normalized OpenAPI envelopes do diverge — FastAPI adds
HTTPValidationError, the two converters wrap things slightly differently
in non-AFI-essential ways. That's not in the test scope. The codegen
only consumes x-mizan-functions, x-mizan-contexts, and the per-function
type schemas; those are what the test gates.

Makefile: test-afi target added; rolls into the test aggregate.

Verified: 3/3 conformance tests pass. Other surfaces unaffected —
mizan-core 15/15, mizan-django 348 pass, mizan-fastapi 11/11.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-06 18:59:01 -04:00
parent aaaf80cdbf
commit 0a95f3c860
16 changed files with 584 additions and 2 deletions

View File

@@ -35,6 +35,7 @@ from .executor import (
execute_function,
)
from .router import router, mizan_exception_handler, mizan_validation_handler
from .schema import build_schema
__all__ = [
"router",
@@ -42,6 +43,7 @@ __all__ = [
"mizan_validation_handler",
"execute_function",
"compute_invalidation",
"build_schema",
"ErrorCode",
"MizanError",
"NotFound",

View File

@@ -0,0 +1,209 @@
"""
Mizan schema export for FastAPI backends.
Builds an OpenAPI 3.0 document from the registered Mizan functions, mirroring
the shape mizan-django emits via Django Ninja so the codegen consumes either
backend identically.
Usage:
from mizan_fastapi.schema import build_schema
schema = build_schema() # uses globally registered functions
"""
from __future__ import annotations
import re
from typing import Any
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from pydantic import BaseModel, create_model
from mizan_core.registry import get_all_functions, get_context_groups, get_function
__all__ = ["build_schema", "snake_to_camel"]
# Common user identity param names — mirrors mizan-django's _USER_SCOPED_PARAMS
_USER_SCOPED_PARAMS = {"user_id", "user", "owner_id", "account_id"}
def snake_to_camel(name: str) -> str:
"""Convert snake_case or dotted.name to camelCase. Mirrors mizan-django."""
components = re.split(r"[._]", name)
return components[0] + "".join(c.title() for c in components[1:])
def _has_input(input_cls: Any) -> bool:
return (
input_cls is not None
and input_cls is not BaseModel
and hasattr(input_cls, "model_fields")
and bool(input_cls.model_fields)
)
def _annotation_to_jsonschema_type(annotation: Any) -> str:
if annotation is int:
return "integer"
if annotation is float:
return "number"
if annotation is bool:
return "boolean"
return "string"
def _function_metadata(name: str, fn_class: Any) -> dict[str, Any]:
"""Build one entry of x-mizan-functions. Mirrors Django's shape exactly."""
camel = snake_to_camel(name)
meta = getattr(fn_class, "_meta", {})
input_cls = getattr(fn_class, "Input", None)
has_input = _has_input(input_cls)
entry: dict[str, Any] = {
"name": name,
"camelName": camel,
"hasInput": has_input,
"inputType": f"{camel}Input" if has_input else None,
"outputType": f"{camel}Output",
"transport": "websocket" if meta.get("websocket") else "http",
"isContext": meta.get("context", False),
# Form metadata — always emitted so the schema shape matches Django's,
# even for FastAPI projects that don't use forms (these stay False/None).
"isForm": meta.get("form", False),
"formName": meta.get("form_name"),
"formRole": meta.get("form_role"),
}
if meta.get("affects"):
entry["affects"] = meta["affects"]
return entry
def _context_metadata(context_groups: dict[str, list[str]]) -> dict[str, Any]:
"""Build x-mizan-contexts. Mirrors Django's param-elevation logic."""
out: dict[str, Any] = {}
for ctx_name, fn_names in context_groups.items():
param_info: dict[str, dict[str, Any]] = {}
for fn_name in fn_names:
fn_cls = get_function(fn_name)
if fn_cls is None:
continue
input_cls = getattr(fn_cls, "Input", None)
if not _has_input(input_cls):
continue
for field_name, field_info in input_cls.model_fields.items():
if field_name not in param_info:
param_info[field_name] = {
"type": _annotation_to_jsonschema_type(field_info.annotation),
"sharedBy": [],
}
param_info[field_name]["sharedBy"].append(fn_name)
# A param is required iff every function in the context declares it.
for p_meta in param_info.values():
p_meta["required"] = len(p_meta["sharedBy"]) == len(fn_names)
out[ctx_name] = {
"functions": list(fn_names),
"params": param_info,
}
return out
def build_schema() -> dict[str, Any]:
"""
Build an OpenAPI 3.0 schema for all registered Mizan functions.
Drives FastAPI's native OpenAPI generation by registering a stub endpoint
per function with the function's Input/Output Pydantic models, then
appends the protocol's `x-mizan-functions` and `x-mizan-contexts`
extensions.
Returns a dict in the same shape mizan-django's schema export emits, so
the same codegen pipeline consumes either.
"""
functions = get_all_functions()
context_groups = get_context_groups()
schema_app = FastAPI(
title="mizan Server Functions",
version="1.0.0",
description="Auto-generated schema for mizan server functions",
)
# Per-function endpoints + renamed Pydantic models so component names are
# camelCase + "Input"/"Output" rather than the user's original class names.
schema_classes: dict[str, type[BaseModel]] = {}
function_metadata: list[dict[str, Any]] = []
for name, fn_class in functions.items():
camel = snake_to_camel(name)
input_cls = getattr(fn_class, "Input", None)
output_cls = getattr(fn_class, "Output", None) or BaseModel
has_input = _has_input(input_cls)
input_type_name = f"{camel}Input" if has_input else None
output_type_name = f"{camel}Output"
if has_input:
schema_classes[input_type_name] = create_model(
input_type_name, __base__=input_cls,
)
schema_classes[output_type_name] = create_model(
output_type_name, __base__=output_cls,
)
# Stub endpoint — only exists so FastAPI walks Pydantic types into
# components.schemas. Never invoked. Annotations are set explicitly
# rather than via closures so forward-ref resolution doesn't trip on
# locally-bound type names.
if has_input:
async def stub(payload):
return None
stub.__annotations__ = {"payload": schema_classes[input_type_name]}
else:
async def stub():
return None
schema_app.post(
f"/mizan/{name}",
response_model=schema_classes[output_type_name],
operation_id=camel,
summary=fn_class.__doc__ or f"Call {name}",
)(stub)
function_metadata.append(_function_metadata(name, fn_class))
schema = get_openapi(
title=schema_app.title,
version=schema_app.version,
description=schema_app.description,
routes=schema_app.routes,
)
schema["x-mizan-functions"] = function_metadata
if context_groups:
schema["x-mizan-contexts"] = _context_metadata(context_groups)
# Attach x-mizan operation metadata, mirroring Django.
paths = schema.get("paths", {})
for fn_meta in function_metadata:
op = paths.get(f"/mizan/{fn_meta['name']}", {}).get("post")
if op is not None:
op["x-mizan"] = {
"transport": fn_meta["transport"],
"isContext": fn_meta["isContext"],
}
return schema