102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
"""Make the repository root importable for the pipeline test suites, and give every
|
|
suite the application session the API requires since US07-02.
|
|
|
|
The suites drive the API the way the browser does — module-level ``httpx`` calls and
|
|
``TestClient`` — so instead of threading a cookie through several hundred call sites,
|
|
both clients bootstrap the session themselves exactly like ``frontend/js/api.js``:
|
|
fetch ``/api/v1/session`` once, then send the cookie plus the CSRF header, and
|
|
re-bootstrap once on 401 (a restarted server issues a new session).
|
|
|
|
Security tests deliberately bypass this by constructing their own ``httpx.Client``;
|
|
only the module-level helpers are wrapped.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
if str(REPO) not in sys.path:
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
import httpx
|
|
import httpx._api # httpx.get/post resolve `request` in this module
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from photo_pipeline.api.security import CSRF_HEADER, SESSION_COOKIE
|
|
|
|
_SESSIONS: dict[str, tuple[str, str]] = {} # base url -> (session id, csrf token)
|
|
|
|
|
|
def _base(url) -> str:
|
|
parts = httpx.URL(str(url))
|
|
return f"{parts.scheme}://{parts.netloc.decode()}"
|
|
|
|
|
|
def _bootstrap(base: str) -> tuple[str, str]:
|
|
if base not in _SESSIONS:
|
|
response = httpx.Client(timeout=10).get(f"{base}/api/v1/session")
|
|
_SESSIONS[base] = (
|
|
response.cookies.get(SESSION_COOKIE, ""),
|
|
response.json().get("csrf_token", ""),
|
|
)
|
|
return _SESSIONS[base]
|
|
|
|
|
|
def session_client(base_url: str, **kwargs) -> httpx.Client:
|
|
"""An ``httpx.Client`` that has bootstrapped its own application session.
|
|
|
|
For suites that keep one client for a whole journey; it holds its own cookie, so
|
|
a client made after a server restart picks up the new session by construction.
|
|
"""
|
|
client = httpx.Client(base_url=base_url, **kwargs)
|
|
client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
|
|
return client
|
|
|
|
|
|
def _authorized(url, headers, cookies):
|
|
session_id, token = _bootstrap(_base(url))
|
|
headers = dict(headers or {})
|
|
headers.setdefault(CSRF_HEADER, token)
|
|
cookies = dict(cookies or {})
|
|
cookies.setdefault(SESSION_COOKIE, session_id)
|
|
return headers, cookies
|
|
|
|
|
|
@pytest.fixture(autouse=True, scope="session")
|
|
def _api_session():
|
|
real_request, real_stream = httpx._api.request, httpx._api.stream
|
|
real_client_request, real_client_init = TestClient.request, TestClient.__init__
|
|
|
|
def request(method, url, *, headers=None, cookies=None, **kwargs):
|
|
sent, jar = _authorized(url, headers, cookies)
|
|
response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
|
|
if response.status_code == 401:
|
|
_SESSIONS.pop(_base(url), None)
|
|
sent, jar = _authorized(url, headers, cookies)
|
|
response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
|
|
return response
|
|
|
|
def stream(method, url, *, headers=None, cookies=None, **kwargs):
|
|
sent, jar = _authorized(url, headers, cookies)
|
|
return real_stream(method, url, headers=sent, cookies=jar, **kwargs)
|
|
|
|
def client_init(self, app, *args, base_url="http://127.0.0.1", **kwargs):
|
|
# The default "http://testserver" is not a local host, which is exactly what
|
|
# the Host check refuses; in-process tests are still same-origin callers.
|
|
real_client_init(self, app, *args, base_url=base_url, **kwargs)
|
|
|
|
def client_request(self, method, url, *, headers=None, **kwargs):
|
|
if CSRF_HEADER not in self.headers:
|
|
response = real_client_request(self, "GET", "/api/v1/session")
|
|
self.headers[CSRF_HEADER] = response.json()["csrf_token"]
|
|
return real_client_request(self, method, url, headers=headers, **kwargs)
|
|
|
|
httpx._api.request, httpx.request = request, request
|
|
httpx._api.stream, httpx.stream = stream, stream
|
|
TestClient.request, TestClient.__init__ = client_request, client_init
|
|
yield
|
|
httpx._api.request, httpx.request = real_request, real_request
|
|
httpx._api.stream, httpx.stream = real_stream, real_stream
|
|
TestClient.request, TestClient.__init__ = real_client_request, real_client_init
|