FastAPI and TypeScript improved

backends/mizan-fastapi/tests/test_parity.py

@@ -0,0 +1,98 @@
+"""FastAPI parity with Django: X-Mizan-Invalidate header, origin cache, token auth."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import Depends, FastAPI
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+from mizan_core.auth import AuthConfig, JWTConfig, create_access_token
+from mizan_core.cache.backend import MemoryCache
+from mizan_core.client.function import client
+from mizan_core.dispatch import CacheOrchestrator
+from mizan_core.registry import clear_registry, register
+from mizan_fastapi import (
+    MizanAuthMiddleware,
+    MizanConfig,
+    MizanError,
+    mizan_auth,
+    mizan_exception_handler,
+    router as mizan_router,
+)
+
+
+class Out(BaseModel):
+    ok: bool
+
+
+SECRET = "x" * 32
+JWT = JWTConfig(private_key=SECRET, public_key=SECRET)
+
+
+def _app(*, with_cache=False, with_auth_dep=False) -> FastAPI:
+    clear_registry()
+
+    UserCtx = "user"
+
+    @client(context=UserCtx)
+    def user_profile(request, user_id: int) -> Out:
+        return Out(ok=True)
+
+    @client(affects=UserCtx)
+    def update_profile(request, user_id: int) -> Out:
+        return Out(ok=True)
+
+    @client(auth=True)
+    def whoami(request) -> Out:
+        return Out(ok=True)
+
+    register(user_profile, "user_profile")
+    register(update_profile, "update_profile")
+    register(whoami, "whoami")
+
+    app = FastAPI()
+    cache = CacheOrchestrator(MemoryCache(), SECRET) if with_cache else CacheOrchestrator(None, None)
+    app.state.mizan_config = MizanConfig(auth=AuthConfig(jwt=JWT), cache=cache)
+    deps = [Depends(mizan_auth())] if with_auth_dep else []
+    app.include_router(mizan_router, prefix="/api/mizan", dependencies=deps)
+    app.add_exception_handler(MizanError, mizan_exception_handler)
+    return app
+
+
+def test_mutation_emits_invalidate_header():
+    c = TestClient(_app())
+    r = c.post("/api/mizan/call/", json={"fn": "update_profile", "args": {"user_id": 5}})
+    assert r.status_code == 200
+    assert r.json()["invalidate"] == [{"context": "user", "params": {"user_id": 5}}]
+    assert r.headers["X-Mizan-Invalidate"] == "user;user_id=5"
+
+
+def test_origin_cache_hit_miss():
+    c = TestClient(_app(with_cache=True))
+    r1 = c.get("/api/mizan/ctx/user/", params={"user_id": 5})
+    assert r1.status_code == 200 and r1.headers["X-Mizan-Cache"] == "MISS"
+    r2 = c.get("/api/mizan/ctx/user/", params={"user_id": 5})
+    assert r2.headers["X-Mizan-Cache"] == "HIT"
+    assert r1.content == r2.content
+
+
+def test_auth_required_rejects_anonymous():
+    c = TestClient(_app())
+    r = c.post("/api/mizan/call/", json={"fn": "whoami", "args": {}})
+    assert r.status_code == 401
+
+
+def test_auth_required_passes_with_bearer_jwt():
+    c = TestClient(_app(with_auth_dep=True))
+    tok = create_access_token("7", "sess", JWT, is_staff=True)
+    r = c.post("/api/mizan/call/", json={"fn": "whoami", "args": {}},
+              headers={"Authorization": f"Bearer {tok}"})
+    assert r.status_code == 200 and r.json()["result"] == {"ok": True}
+
+
+def test_invalid_bearer_token_rejected():
+    c = TestClient(_app())
+    r = c.post("/api/mizan/call/", json={"fn": "update_profile", "args": {"user_id": 1}},
+              headers={"Authorization": "Bearer not-a-real-token"})
+    assert r.status_code == 401