99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""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
|