Aleksandr Meshchriakov
8ed3e1175c
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 5m5s
CI/CD Pipeline / Run Tests (push) Failing after 5m5s
CI/CD Pipeline / Build Docker Images (push) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (push) Has been skipped
CI/CD Pipeline / Deploy to Server (push) Has been skipped
135 lines
3.9 KiB
Python
135 lines
3.9 KiB
Python
"""Lightweight in-memory HTTP router for integration tests (no sockets)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from types import SimpleNamespace
|
|
from typing import Callable
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
from requests.adapters import BaseAdapter
|
|
from requests.models import Response as RequestsResponse
|
|
|
|
|
|
@dataclass
|
|
class Response:
|
|
status: int = 200
|
|
body: bytes = b""
|
|
headers: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
RouteHandler = Callable[[SimpleNamespace, bytes], Response]
|
|
|
|
|
|
def _json_response(data: object, status: int = 200) -> Response:
|
|
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
|
return Response(
|
|
status=status,
|
|
body=body,
|
|
headers={"Content-Type": "application/json; charset=utf-8"},
|
|
)
|
|
|
|
|
|
class _InMemoryAdapter(BaseAdapter):
|
|
def __init__(self, routes: dict[tuple[str, str], Response | RouteHandler]) -> None:
|
|
super().__init__()
|
|
self._routes = routes
|
|
|
|
def send(self, request, **_kwargs): # noqa: D401
|
|
parsed = urlparse(request.url)
|
|
key = (request.method.upper(), parsed.path)
|
|
route = self._routes.get(key)
|
|
|
|
if route is None:
|
|
response = Response(status=404, body=b"", headers={})
|
|
else:
|
|
body = request.body or b""
|
|
if isinstance(body, str):
|
|
body = body.encode("utf-8")
|
|
if callable(route):
|
|
response = route(
|
|
SimpleNamespace(
|
|
path=parsed.path,
|
|
query=parsed.query,
|
|
method=request.method.upper(),
|
|
),
|
|
body,
|
|
)
|
|
else:
|
|
response = route
|
|
|
|
return self._build_response(request, response)
|
|
|
|
def _build_response(self, request, response: Response) -> RequestsResponse:
|
|
resp = RequestsResponse()
|
|
resp.status_code = response.status
|
|
resp._content = response.body
|
|
resp.headers.update(response.headers)
|
|
resp.url = request.url
|
|
resp.request = request
|
|
return resp
|
|
|
|
def close(self) -> None: # noqa: D401
|
|
return
|
|
|
|
|
|
class TestHTTPServer:
|
|
"""Context-managed in-memory HTTP router with requests adapter."""
|
|
|
|
def __init__(self) -> None:
|
|
self._routes: dict[tuple[str, str], Response | RouteHandler] = {}
|
|
self._adapter = _InMemoryAdapter(self._routes)
|
|
self._base_url = "http://testserver"
|
|
self._started = False
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return self._base_url
|
|
|
|
@property
|
|
def adapter(self) -> BaseAdapter:
|
|
return self._adapter
|
|
|
|
def mount(self, client_or_session) -> None:
|
|
session = getattr(client_or_session, "session", client_or_session)
|
|
session.mount(self._base_url, self._adapter)
|
|
session.mount(self._base_url.replace("http://", "https://", 1), self._adapter)
|
|
|
|
def add_json(self, path: str, data: object, *, status: int = 200) -> None:
|
|
self._routes[("GET", path)] = _json_response(data, status=status)
|
|
|
|
def add_bytes(
|
|
self,
|
|
path: str,
|
|
data: bytes,
|
|
*,
|
|
content_type: str = "application/octet-stream",
|
|
status: int = 200,
|
|
) -> None:
|
|
self._routes[("GET", path)] = Response(
|
|
status=status,
|
|
body=data,
|
|
headers={"Content-Type": content_type},
|
|
)
|
|
|
|
def add_route(self, method: str, path: str, handler: RouteHandler) -> None:
|
|
self._routes[(method.upper(), path)] = handler
|
|
|
|
def start(self) -> None:
|
|
self._started = True
|
|
|
|
def stop(self) -> None:
|
|
self._started = False
|
|
|
|
def __enter__(self) -> "TestHTTPServer":
|
|
self.start()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
self.stop()
|
|
|
|
|
|
__all__ = ["TestHTTPServer", "Response"]
|