Spaces:
Sleeping
Sleeping
File size: 2,668 Bytes
315a6b9 6b429be 315a6b9 6b429be 315a6b9 6b429be 315a6b9 de9192c 315a6b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Fixtures locales aux tests de l'interface web FastAPI.
Le serveur Picarones expose dans :mod:`picarones.interfaces.web.state` trois
états globaux partagés entre routes :
- ``JOBS_SEMAPHORE`` — sémaphore borné pour les benchmarks concurrents,
- ``RATE_LIMITER`` — rate limiter par IP,
- ``picarones.interfaces.web.routers.corpus._BROWSE_ROOTS`` — répertoires
autorisés à la navigation, calculés au chargement.
Ces états peuvent polluer des tests indépendants. Cette fixture
réinitialise les trois entre chaque test web — purge le rate
limiter, ré-injecte un sémaphore frais, recalcule les browse roots
selon l'environnement courant.
Discipline : la fixture est ``autouse=True`` mais n'est définie
que dans ``tests/web/conftest.py`` — les tests des cercles 1 et 2
(``tests/core/``, ``tests/measurements/``, etc.) ne paient pas le
coût de l'import de ``picarones.interfaces.web.*`` à chaque test.
"""
from __future__ import annotations
import threading
import pytest
@pytest.fixture(autouse=True)
def _isolate_web_app_state():
"""Réinitialise sémaphore, rate limiter et browse roots entre tests.
Sans cette fixture :
- une suite séquentielle de 32+ tests qui font ``POST /api/benchmark/*``
peut épuiser le sémaphore (les threads daemon ne libèrent qu'à la
fin du benchmark Tesseract, qui prend une fraction de seconde mais
plusieurs tests consécutifs se bousculent) ;
- un test qui mute ``_BROWSE_ROOTS`` localement laisse une liste
pointant vers un ``tmp_path`` purgé ;
- les tests qui appellent le rate limiter directement laissent des
timestamps dans son bucket.
"""
from picarones.interfaces.web import app as web_app
from picarones.interfaces.web import security as web_sec
from picarones.interfaces.web import state as web_state
from picarones.interfaces.web.routers import corpus as web_corpus_router
# Sauvegarde
original_browse_roots = list(web_corpus_router._BROWSE_ROOTS)
# Sémaphore frais à chaque test (capacité large, voir les env vars
# par défaut dans ``tests/conftest.py``). Le module ``state``
# détient le singleton ; ``app`` en a une référence importée comme
# ``_JOBS_SEMAPHORE`` — on synchronise les deux.
new_sem = threading.Semaphore(web_sec.get_max_concurrent_jobs())
web_state.JOBS_SEMAPHORE = new_sem
web_app._JOBS_SEMAPHORE = new_sem
web_state.RATE_LIMITER.reset()
web_state.RATE_LIMITER.max_per_hour = web_sec.get_rate_limit_per_hour()
yield
# Restauration
web_corpus_router._BROWSE_ROOTS[:] = original_browse_roots
web_state.RATE_LIMITER.reset()
|