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

"""Avito IMV time-budget — slow/hanging IMV must degrade gracefully, not block
/estimate (#b2c-antiabuse-2).

Prod risk: `_get_or_fetch_imv_cached` was the ONLY external call in the estimate
pipeline WITHOUT a `_with_budget` guard. `evaluate_via_imv` chains up to 3
sequential HTTP requests (warm-up + geocode + evaluate), each with its own 25s
timeout (`_HTTP_TIMEOUT_SEC` in scraper_kit.providers.avito.imv), plus a possible
ONE internal retry with a "cleaned" address on `IMVAddressNotFoundError` — the
unbounded worst-case reached ~150s. Every other slow enrichment (geocode/
house_metadata/yandex_valuation/cian_valuation) was already wrapped in
`_with_budget`; this mirrors that guard for IMV.

Contracts locked here:
  1. `estimate_avito_imv_timeout_s` exists with the documented default (20.0s).
  2. An IMV timeout (TimeoutError surfaced by the _with_budget wrapper) degrades
     to an AggregatedEstimate WITHOUT 'avito_imv' in sources_used and no 5xx —
     the same graceful None path as a network error.
  3. `_with_budget` actually enforces the timeout (elapsed time bound), proving
     the guard is live and not merely present in code.

Style mirrors test_estimator_cian_budget.py.
"""

import os

# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")

import asyncio
import time
from unittest.mock import AsyncMock, MagicMock, patch

import anyio


def _make_fake_geo():
    from app.services.geocoder import GeocodeResult

    return GeocodeResult(
        lat=56.838,
        lon=60.595,
        full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
        provider="nominatim",
    )


def _make_payload_full():
    """house_type/repair_state set so the IMV-gated guard is actually satisfied
    (imv_house_type/imv_renovation both non-None — see _IMV_HOUSE_TYPE_MAP /
    _IMV_REPAIR_MAP in estimator.py)."""
    from app.schemas.trade_in import TradeInEstimateInput

    return TradeInEstimateInput(
        address="ЕКБ, ул. Учителей, 18",
        area_m2=38.8,
        rooms=1,
        floor=4,
        total_floors=16,
        house_type="panel",
        repair_state="standard",
    )


def test_avito_imv_timeout_setting_default() -> None:
    """The new budget setting exists and defaults to 20.0s."""
    from app.core.config import settings

    assert hasattr(settings, "estimate_avito_imv_timeout_s")
    assert settings.estimate_avito_imv_timeout_s == 20.0


def test_estimate_avito_imv_timeout_degrades_no_5xx() -> None:
    """IMV raising TimeoutError → estimate without it, no exception.

    The _with_budget() guard wraps the IMV call; a TimeoutError must map to the
    same graceful None path as a network error. Estimator returns an
    AggregatedEstimate; 'avito_imv' absent from sources_used.
    """
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload_full()

    imv_mock = AsyncMock(side_effect=TimeoutError("imv slow"))

    async def _run() -> None:
        with (
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
            patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
            patch("app.services.estimator._fetch_deals", return_value=[]),
            patch("app.services.estimator._get_or_fetch_imv_cached", new=imv_mock),
            patch(
                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
                new=AsyncMock(return_value=None),
            ),
            patch(
                "app.services.estimator.estimate_via_cian_valuation",
                new=AsyncMock(return_value=None),
            ),
            patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
        ):
            result = await estimate_quality(payload, db)

        assert result.estimate_id is not None  # no 5xx — degraded gracefully
        assert "avito_imv" not in result.sources_used

    anyio.run(_run)


def test_estimate_avito_imv_timeout_within_budget() -> None:
    """`_with_budget` actually enforces the configured timeout — a hanging IMV
    coroutine is cancelled at the budget, not left to run the unbounded
    (~150s worst-case) evaluate_via_imv chain. Uses a short synthetic budget
    (not the real 20.0s default) so this test stays fast."""
    from app.services.estimator import _with_budget

    short_budget_s = 0.2

    async def _hangs_forever() -> None:
        await asyncio.sleep(999)

    async def _run() -> None:
        start = time.monotonic()
        result = await _with_budget(_hangs_forever(), short_budget_s, label="avito_imv")
        elapsed = time.monotonic() - start
        assert result is None
        # Generous slack for CI scheduling jitter — proves the guard actually
        # cancels near the budget, not merely that it exists in code.
        assert elapsed < short_budget_s + 2.0

    anyio.run(_run)