tests/test_event_loop_blocking.py

"""Il route /api/detect non deve bloccare l'event loop asincrono.

Regressione: /api/detect era ``async def`` ma chiamava ``analyze_text`` (lavoro
CPU/GPU pesante) direttamente nella coroutine, bloccando l'intero event loop per
tutta la durata dell'analisi. Conseguenza lato UI: durante la PRIMA analisi (con
i modelli ancora da caricare, quindi lunga) il monitor di sistema restava vuoto
perché /api/system/stats non poteva essere servito finché /api/detect non
finiva. Gli altri endpoint pesanti offloadano già su threadpool con
run_in_executor; questo test verifica che anche /api/detect lo faccia.
"""

from __future__ import annotations

import asyncio
import time

import httpx

from config.runtime.settings import get_settings
from web.app import create_app
from web.services.analysis import AnalysisService
from web.services.jobs import JobManager
from web.services.registry import ServiceRegistry


def test_detect_does_not_block_event_loop(monkeypatch):
    app = create_app()
    # Replica lo stato del lifespan SENZA il thread di warmup (non serve qui:
    # analyze_text è patchato per simulare un'analisi lenta).
    settings = get_settings()
    app.state.registry = ServiceRegistry(settings)
    app.state.jobs = JobManager()

    # Simula la prima analisi lenta (caricamento modelli) come lavoro SINCRONO
    # bloccante, poi delega alla vera analyze_text per una risposta valida.
    original = AnalysisService.analyze_text

    def slow_analyze(self, *args, **kwargs):
        time.sleep(0.6)
        return original(self, *args, **kwargs)

    monkeypatch.setattr(AnalysisService, "analyze_text", slow_analyze)

    order: list[str] = []

    async def scenario():
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
            async def do_detect():
                r = await ac.post(
                    "/api/detect",
                    json={"text": "CIG 7654321000", "min_confidence": 0.7, "layers": ["rules"]},
                )
                assert r.status_code == 200
                order.append("detect")

            async def do_stats():
                # Lascia partire /api/detect, poi interroga il monitor mentre
                # l'analisi è in corso.
                await asyncio.sleep(0.1)
                r = await ac.get("/api/system/stats")
                assert r.status_code == 200
                order.append("stats")

            await asyncio.gather(do_detect(), do_stats())

    asyncio.run(scenario())

    # Se l'event loop NON è bloccato, /api/system/stats risponde durante
    # l'analisi e completa PRIMA di /api/detect.
    assert order[0] == "stats", (
        f"event loop bloccato durante /api/detect (ordine={order}): il monitor "
        "non riceve gli stats finché l'analisi non termina"
    )