|
|
"""Minimal FastAPI shim for tests in this workspace.
|
|
|
|
|
|
This provides a tiny subset of the FastAPI API used by the tests: FastAPI, APIRouter,
|
|
|
Depends and TestClient. It's intended only for local test runs where fastapi isn't
|
|
|
installed. It does not implement full ASGI features.
|
|
|
"""
|
|
|
from typing import Callable, Any, Optional
|
|
|
|
|
|
|
|
|
class FastAPI:
|
|
|
def __init__(self):
|
|
|
self._routes = []
|
|
|
|
|
|
def on_event(self, event_name: str):
|
|
|
def decorator(fn: Callable):
|
|
|
|
|
|
setattr(self, f"_on_event_{event_name}", fn)
|
|
|
return fn
|
|
|
return decorator
|
|
|
|
|
|
def get(self, path: str):
|
|
|
def decorator(fn: Callable):
|
|
|
self._routes.append(("GET", path, fn))
|
|
|
return fn
|
|
|
return decorator
|
|
|
|
|
|
def include_router(self, router: Any):
|
|
|
|
|
|
routes = getattr(router, 'routes', [])
|
|
|
self._routes.extend(routes)
|
|
|
|
|
|
|
|
|
class APIRouter:
|
|
|
def __init__(self, prefix: str = "", tags: Optional[list] = None, **kwargs):
|
|
|
|
|
|
self.prefix = prefix or ""
|
|
|
self.tags = tags or []
|
|
|
self.routes = []
|
|
|
|
|
|
def get(self, path: str, **kwargs):
|
|
|
def decorator(fn: Callable):
|
|
|
full_path = f"{self.prefix}{path}"
|
|
|
self.routes.append(("GET", full_path, fn))
|
|
|
return fn
|
|
|
return decorator
|
|
|
|
|
|
def post(self, path: str, **kwargs):
|
|
|
def decorator(fn: Callable):
|
|
|
full_path = f"{self.prefix}{path}"
|
|
|
self.routes.append(("POST", full_path, fn))
|
|
|
return fn
|
|
|
return decorator
|
|
|
|
|
|
|
|
|
def Depends(dep: Any):
|
|
|
return dep
|
|
|
|
|
|
|
|
|
class TestClient:
|
|
|
def __init__(self, app: FastAPI):
|
|
|
self.app = app
|
|
|
|
|
|
def get(self, path: str):
|
|
|
|
|
|
for method, route_path, fn in self.app._routes:
|
|
|
if method == 'GET' and route_path == path:
|
|
|
result = fn()
|
|
|
class Resp:
|
|
|
status_code = 200
|
|
|
def json(self):
|
|
|
return result
|
|
|
return Resp()
|
|
|
class Resp404:
|
|
|
status_code = 404
|
|
|
def json(self):
|
|
|
return {"detail": "Not Found"}
|
|
|
return Resp404()
|
|
|
|
|
|
|
|
|
class HTTPException(Exception):
|
|
|
def __init__(self, status_code: int = 500, detail: Any = None):
|
|
|
self.status_code = status_code
|
|
|
self.detail = detail
|
|
|
super().__init__(f"HTTP {status_code}: {detail}")
|
|
|
|
|
|
|
|
|
class _status:
|
|
|
HTTP_401_UNAUTHORIZED = 401
|
|
|
HTTP_403_FORBIDDEN = 403
|
|
|
HTTP_400_BAD_REQUEST = 400
|
|
|
|
|
|
status = _status()
|
|
|
|