| | """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()
|
| |
|