gendesign/tradein-mvp/backend/tests/test_3082_estimate_concurrency_cap.py

"""#3082: потолок одновременных оценок на POST /estimate.

Рейт-лимит меряет частоту (300/60с вправе стартовать в одну секунду), квота —
счётная и помесячная: параллелизм не ограничивал никто, и пила одновременных
оценок (0.8–2.4с каждая, соединение общего пула + внешние тиры) выедала пул и
тормозила весь /api/v1/*. Фикс — asyncio.Semaphore по образцу public/mera.py.

Красный на origin/main по ЗНАЧЕНИЮ: без семафора все 5 конкурентных запросов
проходят (0×429), с ним пятый получает быстрый 429 «занят». Никаких обращений
к новым именам модуля напрямую — monkeypatch констант с raising=False, чтобы
на main тест падал ассертом о поведении, а не AttributeError (см.
red-must-mean-wrong-value).
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import asyncio
from datetime import UTC, datetime, timedelta
from unittest.mock import patch
from uuid import uuid4

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

from app.api.v1 import trade_in as trade_in_module
from app.core.db import get_db
from app.core.ratelimit import SlidingWindowLimiter
from app.schemas.trade_in import AggregatedEstimate

_CONCURRENCY = 4  # прод-значение _ESTIMATE_CONCURRENCY; здесь литералом (см. док-стринг)


def _canned_estimate() -> AggregatedEstimate:
    return AggregatedEstimate(
        estimate_id=uuid4(),
        median_price_rub=5_000_000,
        range_low_rub=4_500_000,
        range_high_rub=5_500_000,
        median_price_per_m2=100_000,
        confidence="medium",
        n_analogs=8,
        period_months=24,
        analogs=[],
        actual_deals=[],
        expires_at=datetime.now(tz=UTC) + timedelta(hours=24),
    )


@pytest.fixture()
def app() -> FastAPI:
    application = FastAPI()
    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")

    def _override_db():
        yield None

    application.dependency_overrides[get_db] = _override_db
    return application


@pytest.fixture(autouse=True)
def _wide_rate_limiter(monkeypatch: pytest.MonkeyPatch) -> None:
    """Рейт-лимит не должен мешать тесту параллелизма: 100/60с."""
    monkeypatch.setattr(
        trade_in_module, "_estimate_limiter", SlidingWindowLimiter(limit=100, window_s=60.0)
    )


async def test_fifth_concurrent_estimate_gets_fast_429(
    app: FastAPI, monkeypatch: pytest.MonkeyPatch
) -> None:
    """4 оценки висят в работе → 5-я не ждёт в очереди, а быстро получает 429
    с текстом про занятость и Retry-After; после освобождения слотов те же 4
    завершаются 200 и следующий запрос снова проходит (release в finally)."""
    # raising=False: на main этих имён нет — тест должен упасть ассертом ниже,
    # а не AttributeError здесь.
    monkeypatch.setattr(trade_in_module, "_ESTIMATE_SLOT_WAIT_S", 0.1, raising=False)
    monkeypatch.setattr(
        trade_in_module, "_estimate_slots", asyncio.Semaphore(_CONCURRENCY), raising=False
    )

    gate = asyncio.Event()
    # Счётчик входов в оценку: слот к этому моменту уже захвачен эндпоинтом.
    entered = asyncio.Semaphore(0)

    async def _slow_estimate(*args, **kwargs) -> AggregatedEstimate:
        entered.release()
        await gate.wait()
        return _canned_estimate()

    payload = {"address": "г. Екатеринбург, ул. Малышева, 1", "area_m2": 50.0, "rooms": 2}
    with (
        patch("app.services.account_quota.check_and_raise"),
        patch("app.services.account_quota.increment", return_value=True),
        patch("app.services.estimator.estimate_quality", new=_slow_estimate),
    ):
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            holders = [
                asyncio.create_task(client.post("/api/v1/trade-in/estimate", json=payload))
                for _ in range(_CONCURRENCY)
            ]
            # Дождаться, что все держатели ВОШЛИ в оценку, то есть слоты заняты.
            # Сон вместо этого — гонка (#3270): на нагруженном раннере пятый
            # запрос успевал захватить свободный слот и навсегда вис на
            # gate.wait(), потому что gate.set() стоит ниже по этому же корутину.
            for i in range(_CONCURRENCY):
                try:
                    await asyncio.wait_for(entered.acquire(), timeout=10)
                except TimeoutError:  # pragma: no cover — диагностика
                    gate.set()
                    pytest.fail(f"до оценки дошли {i} держателей из {_CONCURRENCY}")

            try:
                fifth = await asyncio.wait_for(
                    client.post("/api/v1/trade-in/estimate", json=payload), timeout=10
                )
            except TimeoutError:  # pragma: no cover — диагностика вместо 120с виса
                gate.set()
                pytest.fail("5-й запрос повис вместо быстрого 429 — слот оказался свободен")
            assert fifth.status_code == 429, (
                f"5-й конкурентный запрос прошёл ({fifth.status_code}) — "
                "потолка одновременности нет"
            )
            assert "занят" in fifth.json()["detail"]
            assert "Retry-After" in fifth.headers

            gate.set()
            done = await asyncio.gather(*holders)
            assert [r.status_code for r in done] == [200] * _CONCURRENCY

            # Слоты вернулись (release в finally) — новый запрос проходит.
            gate.set()
            again = await client.post("/api/v1/trade-in/estimate", json=payload)
            assert again.status_code == 200


async def test_within_limit_behaviour_unchanged(
    app: FastAPI, monkeypatch: pytest.MonkeyPatch
) -> None:
    """В пределах лимита семафор прозрачен: одиночный запрос — 200, как раньше."""
    monkeypatch.setattr(trade_in_module, "_ESTIMATE_SLOT_WAIT_S", 0.1, raising=False)

    async def _fast_estimate(*args, **kwargs) -> AggregatedEstimate:
        return _canned_estimate()

    payload = {"address": "г. Екатеринбург, ул. Малышева, 1", "area_m2": 50.0, "rooms": 2}
    with (
        patch("app.services.account_quota.check_and_raise"),
        patch("app.services.account_quota.increment", return_value=True),
        patch("app.services.estimator.estimate_quality", new=_fast_estimate),
    ):
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            resp = await client.post("/api/v1/trade-in/estimate", json=payload)
            assert resp.status_code == 200