"""Внешние источники оценки не ждутся в запросе, а догружаются в фоне. Замер на проде 2026-08-22: расчёт по новому адресу 8-12 с, из них ~6 с ждёт Yandex и ~1.5 с Cian; собственный расчёт укладывается в секунду. По уже виденному адресу (кэш 24 ч) — 0.4-0.8 с. Ключевое свойство, которое делает переключение безопасным: деградация источника в None — НЕ новое состояние ответа. Ровно так же ведёт себя таймаут `estimate_*_valuation_timeout_s`, и этот путь работает в проде сегодня. Тесты проверяют три вещи, каждая из которых при поломке молча вернула бы секунды ожидания в пользовательский путь: 1. `fetch_on_miss=False` действительно НЕ ходит в сеть при промахе кэша; 2. попадание в кэш работает одинаково в обоих режимах; 3. отложенная задача берёт СВОЮ сессию и гасит свои ошибки. """ from __future__ import annotations import asyncio from typing import Any from unittest.mock import MagicMock import pytest class _NoRowsDB: """Сессия, у которой кэш всегда пуст.""" def __init__(self) -> None: self.commits = 0 self.closed = False def execute(self, *_a: Any, **_kw: Any) -> Any: result = MagicMock() result.mappings.return_value.first.return_value = None return result def commit(self) -> None: self.commits += 1 def rollback(self) -> None: return None def close(self) -> None: self.closed = True # ── Yandex ─────────────────────────────────────────────────────────────────── def test_yandex_cache_only_does_not_touch_network(monkeypatch: pytest.MonkeyPatch) -> None: """fetch_on_miss=False → None и НИ ОДНОГО обращения к скраперу.""" from app.services import estimator def _boom(*_a: Any, **_kw: Any) -> Any: raise AssertionError("сеть не должна дёргаться в режиме «только кэш»") monkeypatch.setattr(estimator, "YandexValuationScraper", _boom) out = asyncio.run( estimator._get_or_fetch_yandex_valuation_cached( _NoRowsDB(), # type: ignore[arg-type] address="Екатеринбург, улица Сурикова, 4", fetch_on_miss=False, ) ) assert out is None def test_yandex_default_still_fetches(monkeypatch: pytest.MonkeyPatch) -> None: """Дефолт не изменился: без флага источник по-прежнему грузится в запросе.""" from app.services import estimator called: dict[str, bool] = {} class _Scraper: def __init__(self, *_a: Any, **_kw: Any) -> None: called["constructed"] = True async def __aenter__(self) -> _Scraper: return self async def __aexit__(self, *_a: Any) -> None: return None async def fetch_house_history(self, **_kw: Any) -> None: return None monkeypatch.setattr(estimator, "YandexValuationScraper", _Scraper) asyncio.run( estimator._get_or_fetch_yandex_valuation_cached( _NoRowsDB(), # type: ignore[arg-type] address="Екатеринбург, улица Сурикова, 4", ) ) assert called.get("constructed") is True # ── отложенная догрузка ────────────────────────────────────────────────────── def test_deferred_refresh_uses_its_own_session(monkeypatch: pytest.MonkeyPatch) -> None: """Сессия запроса закрывается вместе с ответом — фон обязан открыть свою. Переиспользование чужой сессии зафиксировало бы её незавершённую работу: обе функции источников делают внутри себя `db.commit()`. """ from app.services import estimator own = _NoRowsDB() request_db = _NoRowsDB() monkeypatch.setattr(estimator, "SessionLocal", lambda: own) seen: dict[str, Any] = {} async def _work(db: Any) -> None: seen["db"] = db async def _drive() -> None: estimator._defer_external_refresh("test", _work) await asyncio.sleep(0) await asyncio.sleep(0) asyncio.run(_drive()) assert seen["db"] is own, "фоновая задача взяла не свою сессию" assert seen["db"] is not request_db assert own.closed is True, "фоновая сессия осталась незакрытой" def test_deferred_refresh_swallows_errors(monkeypatch: pytest.MonkeyPatch) -> None: """Провал прогрева кэша не должен всплывать: это не часть ответа.""" from app.services import estimator own = _NoRowsDB() monkeypatch.setattr(estimator, "SessionLocal", lambda: own) async def _work(_db: Any) -> None: raise RuntimeError("внешний сервис лёг") async def _drive() -> None: estimator._defer_external_refresh("test", _work) await asyncio.sleep(0) await asyncio.sleep(0) asyncio.run(_drive()) # не должно поднять assert own.closed is True def test_deferred_refresh_without_event_loop_is_noop() -> None: """Синхронный контекст (скрипт/тест) — прогрев необязателен, не падаем.""" from app.services import estimator async def _work(_db: Any) -> None: raise AssertionError("не должно вызваться") estimator._defer_external_refresh("test", _work) # без запущенного loop def test_task_reference_is_retained(monkeypatch: pytest.MonkeyPatch) -> None: """Ссылка на задачу удерживается: иначе GC может убить её на полпути.""" from app.services import estimator monkeypatch.setattr(estimator, "SessionLocal", _NoRowsDB) started = asyncio.Event() release = asyncio.Event() async def _work(_db: Any) -> None: started.set() await release.wait() async def _drive() -> None: estimator._defer_external_refresh("test", _work) await started.wait() assert len(estimator._DEFERRED_REFRESH_TASKS) == 1 release.set() await asyncio.sleep(0) await asyncio.sleep(0) assert len(estimator._DEFERRED_REFRESH_TASKS) == 0, "ссылка не убрана после завершения" asyncio.run(_drive()) def test_deferred_queue_is_bounded(monkeypatch: pytest.MonkeyPatch) -> None: """Всплеск по новым адресам не должен породить сотни задач. Ровно тот случай, ради которого режим и сделан: у backend один воркер, mem_limit 768m, у Postgres max_connections 100. Неограниченный create_task превратил бы ускорение в отказ. """ from app.services import estimator monkeypatch.setattr(estimator, "SessionLocal", _NoRowsDB) monkeypatch.setattr(estimator, "_MAX_DEFERRED_REFRESH_TASKS", 3) release = asyncio.Event() async def _work(_db: Any) -> None: await release.wait() async def _drive() -> None: for _ in range(50): estimator._defer_external_refresh("test", _work) await asyncio.sleep(0) assert len(estimator._DEFERRED_REFRESH_TASKS) <= 3, ( f"очередь догрузки не ограничена: {len(estimator._DEFERRED_REFRESH_TASKS)} задач" ) release.set() for _ in range(5): await asyncio.sleep(0) asyncio.run(_drive())