tests/test_slow_work_runs_in_a_thread.py
Rama main · réplica de lectura del repositorio original.
"""Slow work in an `async` route runs in a thread, so the app goes on answering.
An async route runs on the event loop, the one thread every other request
needs. OCR there, at up to a minute a page, held them all, /healthz with them,
so the liveness probe could restart the pod mid-import. The setup wizard's
database test was the same failure (test_setup_wizard.py).
"""
from __future__ import annotations
import asyncio
import gc
import time
import httpx
from app import ocr, pagemap, statements
from test_routes import make_login, session_csrf
HTML = {"accept": "text/html"}
SLOW = 1.0
def health_during(app, send, *, cookies=None) -> tuple[float, float]:
"""Seconds until `send` finished, and until /healthz, asked just after
`send` began, answered."""
async def both():
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver",
cookies=cookies) as http:
began = time.monotonic()
async def slow():
await send(http)
return time.monotonic() - began
async def health():
await asyncio.sleep(0.1)
await http.get("/healthz")
return time.monotonic() - began
return tuple(await asyncio.gather(slow(), health()))
# A full garbage collection, late in the suite, stopped the loop for 0.7 s
# on a CI runner. That is not the slow work holding it, so none runs here.
enabled = gc.isenabled()
gc.disable()
try:
return asyncio.run(both())
finally:
if enabled:
gc.enable()
def _slowly(answer):
def slow(*_args, **_kwargs):
time.sleep(SLOW)
if isinstance(answer, Exception):
raise answer
return answer
return slow
def _upload(path: str, token: str):
async def send(http):
await http.post(path, headers=HTML, data={"_csrf": token},
files={"file": ("scan.pdf", b"%PDF-1.4 fake", "application/pdf")})
return send
def test_a_scanned_statement_is_read_in_a_thread(client, session_factory, app_module,
monkeypatch):
monkeypatch.setattr(statements, "_pdf_text", lambda data: "")
monkeypatch.setattr(ocr, "available", lambda: True)
monkeypatch.setattr(ocr, "read_pdf", _slowly("ALPHA paid 104.70 on 15 July 2026"))
make_login(client, session_factory)
took, health_took = health_during(
app_module.app, _upload("/imports-exports/statement", session_csrf(session_factory)),
cookies=client.cookies)
assert took >= SLOW
assert health_took < 0.6, "the health check waited for the OCR"
def test_a_statement_is_drawn_for_the_visual_mapper_in_a_thread(client, session_factory,
app_module, monkeypatch):
monkeypatch.setattr(pagemap, "_render", _slowly(ValueError("not a PDF")))
make_login(client, session_factory)
took, health_took = health_during(
app_module.app,
_upload("/imports-exports/statement/visual", session_csrf(session_factory)),
cookies=client.cookies)
assert took >= SLOW
assert health_took < 0.6, "the health check waited for the pages to be drawn"
def test_a_chart_preview_is_worked_out_in_a_thread(client, session_factory, app_module,
monkeypatch):
"""The preview runs the whole chart engine, as often as the builder is
dragged; the charts page, which runs the same engine, never held the loop."""
monkeypatch.setattr(app_module.charts_build, "run", _slowly({"series": []}))
make_login(client, session_factory)
token = session_csrf(session_factory)
async def preview(http):
await http.post("/charts/preview", json={"kind": "line"},
headers={"x-csrf-token": token})
took, health_took = health_during(app_module.app, preview, cookies=client.cookies)
assert took >= SLOW
assert health_took < 0.6, "the health check waited for the chart"
def test_a_back_channel_logout_is_verified_in_a_thread(client, app_module, monkeypatch):
"""Its verification fetched the provider's configuration and keys, ten
seconds' timeout each, on the event loop."""
conf = app_module.settings.auth.oidc
for name, value in (("enabled", True), ("issuer", "https://idp.example.test"),
("client_id", "stocktake"), ("client_secret", "a-secret"),
("redirect_uri", "https://app.example.test/login/oidc/callback")):
monkeypatch.setattr(conf, name, value)
monkeypatch.setattr(app_module.oidc, "verify_logout_token",
_slowly(app_module.oidc.OidcError("not verified")))
async def logout(http):
await http.post("/oidc/backchannel-logout", data={"logout_token": "x.y.z"})
took, health_took = health_during(app_module.app, logout)
assert took >= SLOW
assert health_took < 0.6, "the health check waited for the verification"
def test_a_password_is_checked_in_a_thread(client, session_factory, app_module, monkeypatch):
"""Argon2 takes tens of milliseconds by design, for unknown addresses too,
and every sign-in spent them on the one thread every other request
needs."""
from test_routes import pre_auth_csrf
make_login(client, session_factory)
client.post("/logout", data={"_csrf": session_csrf(session_factory)}, follow_redirects=False)
token = pre_auth_csrf(client)
monkeypatch.setattr(app_module.auth, "verify_password", _slowly(False))
async def sign_in(http):
await http.post("/login", headers=HTML, data={
"email": "user@example.test", "password": "wrong", "_csrf": token})
took, health_took = health_during(app_module.app, sign_in, cookies=client.cookies)
assert took >= SLOW
assert health_took < 0.6, "the health check waited for the password"
def test_no_async_route_runs_argon2_on_the_event_loop(app_module):
"""Every handler that checks or makes a password hash, read off its
syntax tree: an `async def` that calls one directly holds the loop for as
long as argon2 takes. `asyncio.to_thread(auth.verify_password, …)` names
it without calling it, and passes."""
import ast
import inspect
import textwrap
argon2 = {"verify_password", "hash_password"}
found = []
for route in app_module.app.routes:
handler = getattr(route, "endpoint", None)
if handler is None or not inspect.iscoroutinefunction(handler):
continue
tree = ast.parse(textwrap.dedent(inspect.getsource(handler)))
calls = {node.func.attr if isinstance(node.func, ast.Attribute)
else getattr(node.func, "id", "")
for node in ast.walk(tree) if isinstance(node, ast.Call)}
found += [route.path for name in calls & argon2]
assert found == []