diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 9adbaded..0a2da10a 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -906,8 +906,18 @@ def _defer_external_refresh(label: str, work: Callable[[Session], Awaitable[obje db = SessionLocal() try: await work(db) - except Exception: - logger.exception("deferred %s: догрузка не удалась (кэш не прогрет)", label) + except Exception as exc: + if caused_by_no_proxy(exc): + # #3398: пустой пул — НАША инфраструктура, HTTP-запрос не уходил вовсе + # (NoProxyAvailableError поднимается в curl_proxy_url ДО запроса). WARNING, + # не ERROR: GlitchTip слушает event_level=ERROR, а это штатная деградация + # прогрева, не сбой площадки — событие тут было бы шумом. + # На проде ESTIMATE_EXTERNAL_SOURCES_BACKGROUND=true, поэтому настоящий + # фетч уходит именно сюда: synchronous-ветка с fetch_on_miss=False отдаёт + # None ДО прокси-слоя и до своего WARNING в estimate_quality не доходит. + logger.warning("deferred %s: пул прокси пуст — кэш не прогрет: %s", label, exc) + else: + logger.exception("deferred %s: догрузка не удалась (кэш не прогрет)", label) finally: db.close() @@ -1006,14 +1016,27 @@ async def _get_or_fetch_yandex_valuation_cached( # Fresh fetch try: async with YandexValuationScraper( - RealScraperConfig(), delay_provider=get_scraper_delay + RealScraperConfig(), + delay_provider=get_scraper_delay, + # #3398: без provider'а `curl_proxy_url` считает use_pool=False (флаг AND + # provider is not None) — пул выключен по построению, curl-сессия уходит на + # env-прокси SCRAPER_PROXY_URL (мёртвый узел #2613 → 407 на каждой пробе). + # Lease один на сессию: acquire в __aenter__, release в __aexit__ — в т.ч. + # на исключении и на отмене по таймауту _with_budget. + proxy_provider=RealProxyProvider(), ) as scraper: result = await scraper.fetch_house_history( address=address, offer_category=offer_category, offer_type=offer_type, ) - except Exception: + except Exception as e: + if caused_by_no_proxy(e): + # #3398: «пул прокси пуст» — НАША инфраструктура, HTTP-запрос не уходил вовсе. + # Отдельный текст (и WARNING, не ERROR): это не сбой Яндекса, GlitchTip-событие + # здесь было бы шумом. Возврат None — тот же graceful путь, /estimate не 5xx. + logger.warning("yandex_valuation: пул прокси пуст — продолжаем без Yandex: %s", e) + return None # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING, # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip @@ -1023,7 +1046,15 @@ async def _get_or_fetch_yandex_valuation_cached( return None if result is None: - logger.info("yandex_valuation: empty result for address=%s", address[:60]) + # #3398: None здесь НЕразличимо «дом не найден» и «фетч не дошёл» — kit ловит + # ProxyError/HTTP-код внутри fetch_house_history и логирует своей строкой + # («yandex valuation fetch failed» / «returned NNN»). Прежний текст «empty result» + # читался как «у Яндекса нет данных», хотя при мёртвом прокси это был 407. + logger.info( + "yandex_valuation: пусто для address=%s — данных нет ЛИБО фетч не дошёл " + "(причина строкой выше от scraper_kit: прокси/HTTP)", + address[:60], + ) return None # Save to cache (UPSERT on (source, cache_key)) @@ -4624,6 +4655,21 @@ async def estimate_quality( "deal_type": "sale", "use_cache": True, "house_id": target_house_id, + # #3398: без provider'а curl_proxy_url считает use_pool=False (флаг AND + # provider is not None) → env-прокси CIAN_PROXY_URL/SCRAPER_PROXY_URL, мёртвый + # узел (#2613) → «Cian valuation fetch failed: … 407». Lease живёт внутри + # curl_proxy_url: acquire до запроса, release в finally. + # + # Весь этот dict переиспользуется отложенной фоновой догрузкой ниже + # (`_defer_external_refresh` захватывает `_c_kwargs` замыканием), то есть + # `config` и `proxy_provider` — ОДИН инстанс на два вызова, которые могут идти + # одновременно (фон стартует после ответа, но живёт своей задачей). Корректно + # это ровно пока оба stateless: `RealScraperConfig` — read-only снимок настроек, + # `RealProxyProvider` не хранит полей вообще и открывает короткую сессию БД на + # каждую операцию (acquire/release/mark_health), поэтому lease'ы двух вызовов не + # пересекаются. Появится у любого из них per-вызов состояние (кэш lease'а, + # счётчик, открытая сессия) — фоновой задаче нужен СВОЙ инстанс, а не общий. + "proxy_provider": RealProxyProvider(), } try: cian_val = await _with_budget( @@ -4643,13 +4689,19 @@ async def estimate_quality( cian_val.sale_accuracy, cian_val.external_house_id, ) - except Exception: - # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration - # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING, - # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip - # (только breadcrumb) — config-wiring регрессия здесь была бы не видна - # мониторингу. .exception() логирует на ERROR + traceback (#2337). - logger.exception("cian_valuation: lookup failed (graceful)") + except Exception as e: + if caused_by_no_proxy(e): + # #3398: пустой пул поднимает NoProxyAvailableError ДО HTTP (curl_proxy_url), + # то есть это не сбой Циана. WARNING, не ERROR: в GlitchTip такое событие — + # шум. Оценка отдаётся без cian-источника, /estimate не 5xx. + logger.warning("cian_valuation: пул прокси пуст — продолжаем без Cian: %s", e) + else: + # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration + # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING, + # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip + # (только breadcrumb) — config-wiring регрессия здесь была бы не видна + # мониторингу. .exception() логирует на ERROR + traceback (#2337). + logger.exception("cian_valuation: lookup failed (graceful)") # ── Pre-fetch: same-building anchor comps ───────────────────────────────── # Guard mirrors the original in-block guard exactly; when false → ([], None). diff --git a/tradein-mvp/backend/tests/test_3398_estimator_valuations_proxy_pool.py b/tradein-mvp/backend/tests/test_3398_estimator_valuations_proxy_pool.py new file mode 100644 index 00000000..6c019c49 --- /dev/null +++ b/tradein-mvp/backend/tests/test_3398_estimator_valuations_proxy_pool.py @@ -0,0 +1,390 @@ +"""Yandex/Cian valuation в /estimate идут через пул прокси, пустой пул не роняет ответ (#3398). + +Корень тот же, что у IMV (#3386/PR #3397): `estimator` звал kit БЕЗ `proxy_provider` — +`YandexValuationScraper(RealScraperConfig(), delay_provider=...)` и +`estimate_via_cian_valuation(**_c_kwargs)`. `providers/_proxy.py::curl_proxy_url` считает +`use_pool = флаг AND proxy_provider is not None`, поэтому пул был выключен ПО ПОСТРОЕНИЮ, а +curl уходил на env-прокси `SCRAPER_PROXY_URL` (выключенный узел, #2613) → на проде каждая +проба `/estimate` давала `407 CONNECT tunnel failed` и «yandex_valuation: empty result». + +Проверки по значению: + (а) оба call site передают provider'а (на main здесь None); + (б) пул пуст + environment=production → оценка отдаётся БЕЗ этих источников, без + исключения, HTTP не уходит, а причина в логе честная — «пул прокси пуст»; + (б2) прод-режим `ESTIMATE_EXTERNAL_SOURCES_BACKGROUND=true`: синхронный вызов уходит + с `fetch_on_miss=False` и до прокси-слоя не доходит вовсе, настоящий фетч делает + `_defer_external_refresh` — там пустой пул тоже WARNING, а не ERROR с трейсбеком + (иначе GlitchTip получает событие на каждый /estimate по новому адресу); + (в) lease освобождён ровно один раз (успех и ошибка фетча) — через настоящий + `curl_proxy_url`, а не через мок провайдера. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from contextlib import ExitStack +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import anyio +import pytest +from scraper_kit.contracts import ProxyLease + +from app.core.config import settings +from app.schemas.trade_in import TradeInEstimateInput +from app.services import estimator +from app.services.estimator import _get_or_fetch_yandex_valuation_cached, estimate_quality +from app.services.geocoder import GeocodeResult + +_ADDRESS = "Первоуральск, ул. Ленина, 5" + + +# ── провайдеры-дублёры ────────────────────────────────────────────────────── + + +class _EmptyPoolProvider: + """acquire → None (все узлы забанены/заняты).""" + + def acquire(self, provider: str) -> ProxyLease | None: + return None + + def release(self, lease: ProxyLease) -> None: # pragma: no cover — не должно звучать + raise AssertionError("release без lease") + + def mark_health(self, lease: ProxyLease, ok: bool, **kw: Any) -> None: # pragma: no cover + raise AssertionError("mark_health без lease") + + def mark_banned(self, lease: ProxyLease, *, source: str) -> None: # pragma: no cover + raise AssertionError("mark_banned без lease") + + +class _CountingProvider: + def __init__(self) -> None: + self.acquired: list[str] = [] + self.released: list[int] = [] + self.health: list[bool] = [] + + def acquire(self, provider: str) -> ProxyLease | None: + self.acquired.append(provider) + return ProxyLease(id=7, url="http://pool-node:3128", kind="datacenter", rotate_url=None) + + def release(self, lease: ProxyLease) -> None: + self.released.append(lease.id) + + def mark_health(self, lease: ProxyLease, ok: bool, **kw: Any) -> None: + self.health.append(ok) + + def mark_banned(self, lease: ProxyLease, *, source: str) -> None: + pass + + +def _no_http(*a: Any, **kw: Any) -> Any: # pragma: no cover — не должно вызваться + raise AssertionError("HTTP-запрос при пустом пуле — прокси-гейт не сработал") + + +class _FakeResponse: + def __init__(self, status_code: int, text: str) -> None: + self.status_code = status_code + self.text = text + + +def _fake_curl_session(*, get: Any) -> Any: + """Заглушка curl_cffi.AsyncSession: и как async-CM (cian), и как объект (yandex).""" + session = MagicMock() + session.get = get + session.close = AsyncMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=False) + return lambda *a, **kw: session + + +def _use_pool_in_production(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(settings, "use_proxy_pool_curl", True) + monkeypatch.setattr(settings, "environment", "production") + # fetch_on_miss=False (фоновый режим) вернул бы None ДО прокси-слоя — тест + # проверял бы тогда пустоту кэша, а не проводку пула. + monkeypatch.setattr(settings, "estimate_external_sources_background", False) + + +# ── (а) yandex: провайдер доезжает до скрейпера ────────────────────────────── + + +def _db_cache_miss() -> MagicMock: + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = None + return db + + +def _call_yandex(db: Any) -> Any: + async def _run() -> Any: + return await _get_or_fetch_yandex_valuation_cached(db, address=_ADDRESS) + + return anyio.run(_run) + + +def test_yandex_valuation_call_passes_proxy_provider() -> None: + """YandexValuationScraper получает proxy_provider (на main здесь None → пул off).""" + captured: dict[str, Any] = {} + + def _fake_scraper_cls(*args: Any, **kwargs: Any) -> Any: + captured.update(kwargs) + scraper = MagicMock() + scraper.fetch_house_history = AsyncMock(return_value=None) + scraper.__aenter__ = AsyncMock(return_value=scraper) + scraper.__aexit__ = AsyncMock(return_value=False) + return scraper + + with patch.object(estimator, "YandexValuationScraper", _fake_scraper_cls): + assert _call_yandex(_db_cache_miss()) is None + + assert captured, "скрейпер не создавался — тест ничего не проверил" + assert captured.get("proxy_provider") is not None + + +# ── (а) cian: провайдер доезжает до kit-функции ────────────────────────────── + + +def _geo() -> GeocodeResult: + return GeocodeResult(lat=56.909, lon=59.960, full_address=_ADDRESS, provider="nominatim") + + +def _listing(i: int) -> dict[str, Any]: + return { + "source": "avito", + "source_url": f"https://avito.ru/offer/{i}", + "address": f"д. {i}", + "lat": 56.909, + "lon": 59.960, + "rooms": 2, + "area_m2": 45.0, + "floor": 5, + "total_floors": 9, + "price_rub": (200_000.0 + i * 1_000) * 45.0, + "price_per_m2": 200_000.0 + i * 1_000, + "listing_date": datetime(2026, 5, 1, tzinfo=UTC), + "days_on_market": 10, + "photo_urls": [], + "scraped_at": datetime(2026, 5, 20, tzinfo=UTC), + "distance_m": 150.0, + "relevance_score": 0.1, + } + + +def _run_estimate(*, extra_patches: list[Any], drain_deferred: bool = False) -> Any: + """estimate_quality() со всеми внешними источниками кроме Cian заглушенными. + + Cian намеренно НЕ патчится списком по умолчанию: тесты (б)/(в) гоняют настоящую + kit-функцию, чтобы прокси-слой (`curl_proxy_url`) реально отработал. + + drain_deferred: дождаться задач фоновой догрузки (`_DEFERRED_REFRESH_TASKS`) ВНУТРИ + того же loop'а и не снимая патчей. Без ожидания `anyio.run` закрывает loop сразу + после ответа, задача умирает не начавшись («Task was destroyed but it is pending») + и тест по фоновому пути был бы зелёным по построению. + """ + payload = TradeInEstimateInput(address=_ADDRESS, area_m2=45.0, rooms=2, floor=5, total_floors=9) + + patches = [ + patch("app.services.estimator.geocode", new=AsyncMock(return_value=_geo())), + patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)), + patch("app.services.estimator.match_house_readonly", return_value=None), + patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)), + patch( + "app.services.estimator._fetch_analogs", + return_value=([_listing(i) for i in range(8)], False, "W"), + ), + patch("app.services.estimator._fetch_anchor_comps", return_value=([], None)), + patch("app.services.estimator._fetch_deals", return_value=[]), + patch("app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)), + patch( + "app.services.estimator._get_or_fetch_yandex_valuation_cached", + new=AsyncMock(return_value=None), + ), + patch("app.services.estimator._fetch_dkp_corridor", return_value=None), + patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)), + *extra_patches, + ] + + async def _run() -> Any: + with ExitStack() as stack: # список патчей переменной длины — не `with (...)` + for p in patches: + stack.enter_context(p) + est = await estimate_quality(payload, MagicMock()) + if drain_deferred: + # gather с return_exceptions=False: задача сама гасит свои ошибки, а если + # перестанет — тест обязан покраснеть, а не проглотить. + await asyncio.gather(*list(estimator._DEFERRED_REFRESH_TASKS)) + return est + + return anyio.run(_run) + + +def test_cian_valuation_call_passes_proxy_provider(monkeypatch: pytest.MonkeyPatch) -> None: + """estimate_via_cian_valuation получает proxy_provider (на main здесь None).""" + _use_pool_in_production(monkeypatch) + captured: dict[str, Any] = {} + + async def _fake_cian(_db: Any, **kwargs: Any) -> None: + captured.update(kwargs) + return None + + est = _run_estimate( + extra_patches=[patch("app.services.estimator.estimate_via_cian_valuation", new=_fake_cian)] + ) + assert est is not None + assert captured, "cian-вызов не состоялся — тест ничего не проверил" + assert captured.get("proxy_provider") is not None + + +# ── (б) пустой пул в проде: деградация, а не 5xx ───────────────────────────── + + +def test_yandex_empty_pool_in_production_degrades( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Пул пуст → None (ответ без Yandex), HTTP не уходит, причина в логе честная.""" + _use_pool_in_production(monkeypatch) + result: Any = "unset" + + with ( + # Без create=True: RealProxyProvider импортирован в estimator — патч обязан + # бить по РЕАЛЬНОМУ атрибуту, иначе переименование импорта оставит тест зелёным. + patch.object(estimator, "RealProxyProvider", _EmptyPoolProvider), + patch("scraper_kit.providers.yandex.valuation._CurlCffiSession", _no_http), + caplog.at_level(logging.WARNING, logger="app.services.estimator"), + ): + result = _call_yandex(_db_cache_miss()) + + assert result is None + assert "пул прокси пуст" in caplog.text, caplog.text + assert "fetch failed" not in caplog.text, "причина подменена на неспецифичную" + + +def test_cian_empty_pool_in_production_degrades( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Тот же контракт для Cian: оценка возвращается, исключение не всплывает наружу.""" + _use_pool_in_production(monkeypatch) + + with ( + patch.object(estimator, "RealProxyProvider", _EmptyPoolProvider), + patch("scraper_kit.providers.cian.valuation._load_from_cache", return_value=None), + patch( + "scraper_kit.providers.cian.valuation.load_session", + return_value={"cookie": "x"}, + ), + patch("scraper_kit.providers.cian.valuation.AsyncSession", _no_http), + caplog.at_level(logging.WARNING, logger="app.services.estimator"), + ): + est = _run_estimate(extra_patches=[]) + + assert est is not None # /estimate отвечает, просто без cian-источника + assert "пул прокси пуст" in caplog.text, caplog.text + assert "lookup failed" not in caplog.text, "причина подменена на неспецифичную" + + +# ── (б2) фоновый режим (прод-конфиг): пустой пул — WARNING, не ERROR ───────── + + +def test_cian_deferred_refresh_empty_pool_warns_without_traceback( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Прод-путь: ESTIMATE_EXTERNAL_SOURCES_BACKGROUND=true → фетч уходит в фон. + + На проде (`docker-compose.prod.yml`) флаг включён, поэтому синхронный вызов идёт с + `fetch_on_miss=False` и возвращает None ДО прокси-слоя (`providers/cian/valuation.py` + «cache MISS — fetch отложен») — ветка WARNING в `estimate_quality` там не звучит. + Настоящий фетч делает `_defer_external_refresh`, где `NoProxyAvailableError` попадал в + общий `except Exception: logger.exception(...)` → ERROR + traceback → событие в + GlitchTip: ровно тот шум, который #3398 и убирает. + """ + _use_pool_in_production(monkeypatch) + monkeypatch.setattr(settings, "estimate_external_sources_background", True) + estimator._DEFERRED_REFRESH_TASKS.clear() # чужие мёртвые задачи из прошлых loop'ов + + with ( + patch.object(estimator, "RealProxyProvider", _EmptyPoolProvider), + # Фоновая задача открывает СВОЮ сессию (сессия запроса закрыта вместе с ответом) — + # в тесте она не должна ходить в реальную БД. + patch.object(estimator, "SessionLocal", MagicMock(return_value=MagicMock())), + patch("scraper_kit.providers.cian.valuation._load_from_cache", return_value=None), + patch("scraper_kit.providers.cian.valuation.load_session", return_value={"cookie": "x"}), + patch("scraper_kit.providers.cian.valuation.AsyncSession", _no_http), + caplog.at_level(logging.INFO, logger="app.services.estimator"), + ): + est = _run_estimate(extra_patches=[], drain_deferred=True) + + assert est is not None + ours = [r for r in caplog.records if r.name == "app.services.estimator"] + deferred = [r for r in ours if "deferred cian_valuation" in r.getMessage()] + assert deferred, [r.getMessage() for r in ours] # фоновая задача не отработала + assert all("пул прокси пуст" in r.getMessage() for r in deferred), [ + r.getMessage() for r in deferred + ] + errors = [r for r in ours if r.levelno >= logging.ERROR] + assert not errors, [(r.levelname, r.getMessage()) for r in errors] + assert not any(r.exc_info for r in ours), "traceback приложен — GlitchTip получит событие" + assert "Traceback" not in caplog.text, caplog.text + + +# ── (в) lease освобождён ровно один раз, через настоящий curl_proxy_url ────── + + +def _yandex_with_pool(monkeypatch: pytest.MonkeyPatch, *, get: Any) -> _CountingProvider: + _use_pool_in_production(monkeypatch) + provider = _CountingProvider() + with ( + patch.object(estimator, "RealProxyProvider", lambda: provider), + patch( + "scraper_kit.providers.yandex.valuation._CurlCffiSession", _fake_curl_session(get=get) + ), + ): + _call_yandex(_db_cache_miss()) + return provider + + +def test_yandex_lease_released_once_on_success(monkeypatch: pytest.MonkeyPatch) -> None: + provider = _yandex_with_pool( + monkeypatch, get=AsyncMock(return_value=_FakeResponse(200, "")) + ) + assert provider.acquired == ["yandex"] + assert provider.released == [7] + assert provider.health == [True] + + +def test_yandex_lease_released_once_on_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Ошибка транспорта (на проде — 407 от мёртвого env-прокси) не течёт lease'ом.""" + provider = _yandex_with_pool(monkeypatch, get=AsyncMock(side_effect=OSError("proxy 407"))) + assert provider.acquired == ["yandex"] + assert provider.released == [7] + + +def test_cian_lease_released_once_on_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None: + _use_pool_in_production(monkeypatch) + provider = _CountingProvider() + + with ( + patch.object(estimator, "RealProxyProvider", lambda: provider), + patch("scraper_kit.providers.cian.valuation._load_from_cache", return_value=None), + patch( + "scraper_kit.providers.cian.valuation.load_session", + return_value={"cookie": "x"}, + ), + patch( + "scraper_kit.providers.cian.valuation.AsyncSession", + _fake_curl_session(get=AsyncMock(side_effect=OSError("proxy 407"))), + ), + ): + est = _run_estimate(extra_patches=[]) + + assert est is not None + assert provider.acquired == ["cian"] + assert provider.released == [7] + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-q"]))