"""Tests for app.tasks.yandex_detail_backfill (the KIT task — already imports `scraper_kit.providers.yandex.detail.{YandexDetailScraper,save_detail_enrichment}`). Fetch mechanism changed: curl_cffi AsyncSession(chrome120+proxy) + YandexDetailScraper.parse (instead of old fetch_detail path). Mocks target session.get and scraper.parse. Легаси `app.services.scrapers.yandex_detail` удалён (#2277 финальный шаг scraper_kit-миграции) — `save_detail_enrichment` coverage-тесты внизу файла переведены на kit-эквивалент (тот же, что реально вызывает эта задача). """ from __future__ import annotations import json import os import re import sys from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") _wp_mock = MagicMock() sys.modules.setdefault("weasyprint", _wp_mock) import pytest # noqa: E402 from app.tasks.yandex_detail_backfill import ( # noqa: E402 CANONICAL_URL_SQL, OFFER_ID_PATTERN, OFFER_URL_PATTERN, YandexDetailBackfillResult, run_yandex_detail_backfill, ) # --------------------------------------------------------------------------- # Path constants for patching # --------------------------------------------------------------------------- _ASYNC_SESSION = "app.tasks.yandex_detail_backfill.AsyncSession" _PARSE = "app.tasks.yandex_detail_backfill.YandexDetailScraper.parse" _SAVE = "app.tasks.yandex_detail_backfill.save_detail_enrichment" _RUNS = "app.tasks.yandex_detail_backfill.runs_mod" _SLEEP = "app.tasks.yandex_detail_backfill.asyncio.sleep" # #2825: settings.scraper_proxy_url заменён на resolve_proxy_url(db, "yandex") # (пул scrape_proxies с учётом банов, fallback на settings.scraper_proxy_url внутри # app.services.proxy_egress) — эти тесты про loop/parse-логику, не про подбор прокси # (см. tests/services/test_proxy_egress.py), поэтому мокаем сам резолвер. _RESOLVE_PROXY_URL = "app.tasks.yandex_detail_backfill.resolve_proxy_url" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_snapshot(n: int) -> list[dict]: return [ {"id": i + 1, "source_url": f"https://realty.yandex.ru/offer/{i + 1}/"} for i in range(n) ] def _mock_db( snapshot: list[dict], unenrichable: int = 0, url_from_offer_id: int = 0, ) -> MagicMock: """Fake Session: execute() отдаёт снапшот через .mappings().all(), а .one() — остаток очереди вне снапшота, РАЗБИТЫЙ по причине (адрес восстановим из source_id / адресовать нечем).""" db = MagicMock() sel = MagicMock() sel.mappings.return_value.all.return_value = snapshot sel.one.return_value = SimpleNamespace( url_from_offer_id=url_from_offer_id, unenrichable_pending=unenrichable, ) db.execute.return_value = sel return db def _make_resp(status: int = 200, text: str = "ok") -> MagicMock: """Fake curl_cffi response.""" resp = MagicMock() resp.status_code = status resp.text = text return resp def _make_session_ctx(get_side_effect) -> MagicMock: """Build AsyncSession context-manager mock with session.get side_effect.""" session = AsyncMock() session.get = AsyncMock(side_effect=get_side_effect) ctx = MagicMock() ctx.__aenter__ = AsyncMock(return_value=session) ctx.__aexit__ = AsyncMock(return_value=None) session_cls = MagicMock(return_value=ctx) return session_cls, session def _mock_resolve_proxy_url(proxy: str | None = "http://proxy:3128") -> MagicMock: """Мок resolve_proxy_url(db, source) -> proxy, независимо от db/source.""" return MagicMock(return_value=proxy) # --------------------------------------------------------------------------- # Tests: run_yandex_detail_backfill # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_backfill_empty_snapshot_marks_done() -> None: """Empty snapshot -> mark_done immediately, no session.get calls.""" db = _mock_db([]) runs = MagicMock() session_cls, session = _make_session_ctx([]) with ( patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=1, params={"batch_size": 10, "budget_sec": 60} ) assert isinstance(result, YandexDetailBackfillResult) assert result.attempted == 0 assert result.enriched == 0 session.get.assert_not_called() runs.mark_done.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio async def test_backfill_processes_snapshot_to_completion() -> None: """3 listings -> all fetched (200) + parse success -> enriched=3, mark_done.""" snapshot = _make_snapshot(3) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() responses = [_make_resp(200), _make_resp(200), _make_resp(200)] session_cls, session = _make_session_ctx(responses) with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, return_value=mock_enrichment), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=2, params={"batch_size": 10, "budget_sec": 3600} ) assert result.attempted == 3 assert result.enriched == 3 assert result.failed == 0 assert session.get.call_count == 3 runs.mark_backfill_finished.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio async def test_backfill_parse_none_abort_after_max_consecutive() -> None: """5 consecutive parse→None -> abort; финал через mark_backfill_finished (#2674). Раньше — mark_done; на проде ровно этот брейкер дал 31 «успешный» прогон из 52 (attempted=5 failed=5 enriched=0). Блоков у Яндекса нет → статус 'failed'. """ snapshot = _make_snapshot(10) db = _mock_db(snapshot) runs = MagicMock() responses = [_make_resp(200)] * 10 session_cls, _session = _make_session_ctx(responses) with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, return_value=None), patch(_RUNS, runs), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}, ) assert result.attempted == 5 assert result.failed == 5 assert result.enriched == 0 runs.mark_backfill_finished.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio async def test_backfill_parse_none_resets_on_success() -> None: """1 parse→None + 1 success + 1 parse→None -> consecutive resets, no abort.""" snapshot = _make_snapshot(3) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() responses = [_make_resp(200)] * 3 session_cls, _session = _make_session_ctx(responses) parse_results = [None, mock_enrichment, None] with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, side_effect=parse_results), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=4, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}, ) assert result.attempted == 3 assert result.enriched == 1 assert result.failed == 2 runs.mark_backfill_finished.assert_called_once() @pytest.mark.asyncio async def test_backfill_non200_counts_as_fail_and_aborts() -> None: """5 consecutive HTTP 403 responses -> abort after max_consecutive_blocks.""" snapshot = _make_snapshot(10) db = _mock_db(snapshot) runs = MagicMock() responses = [_make_resp(403)] * 10 session_cls, _session = _make_session_ctx(responses) with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, return_value=MagicMock()), patch(_RUNS, runs), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=5, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}, ) assert result.failed == 5 assert result.enriched == 0 runs.mark_backfill_finished.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio async def test_backfill_budget_guard_stops_loop() -> None: """Budget expired before first listing -> session.get not called.""" snapshot = _make_snapshot(5) db = _mock_db(snapshot) runs = MagicMock() responses = [_make_resp(200)] * 5 session_cls, session = _make_session_ctx(responses) mono_values = iter([0.0, 999.0, 999.0]) with ( patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), patch("app.tasks.yandex_detail_backfill.time.monotonic", side_effect=mono_values), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): await run_yandex_detail_backfill(db, run_id=6, params={"batch_size": 5, "budget_sec": 1}) session.get.assert_not_called() runs.mark_backfill_finished.assert_called_once() @pytest.mark.asyncio async def test_backfill_top_level_exception_marks_failed() -> None: """db.execute raises -> mark_failed called, exception re-raised.""" db = MagicMock() db.execute.side_effect = RuntimeError("DB connection lost") runs = MagicMock() with ( patch(_RUNS, runs), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): with pytest.raises(RuntimeError, match="DB connection lost"): await run_yandex_detail_backfill( db, run_id=7, params={"batch_size": 5, "budget_sec": 60} ) runs.mark_failed.assert_called_once() runs.mark_backfill_finished.assert_not_called() @pytest.mark.asyncio async def test_backfill_fetch_exception_continues() -> None: """session.get raises on first listing -> failed++, loop continues for second.""" snapshot = _make_snapshot(2) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() # First call raises, second returns 200 OK responses = [RuntimeError("connection reset"), _make_resp(200)] session_cls, _session = _make_session_ctx(responses) with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, return_value=mock_enrichment), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=8, params={"batch_size": 10, "budget_sec": 3600} ) assert result.failed == 1 assert result.enriched == 1 assert result.attempted == 2 # fetch exception is caught by inner try/except (not DB) — no rollback needed db.rollback.assert_not_called() runs.mark_backfill_finished.assert_called_once() @pytest.mark.asyncio async def test_backfill_no_proxy_when_settings_none() -> None: """scraper_proxy_url=None -> AsyncSession called with proxies=None.""" snapshot = _make_snapshot(1) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() responses = [_make_resp(200)] session_cls, _session = _make_session_ctx(responses) with ( patch(_ASYNC_SESSION, session_cls), patch(_PARSE, return_value=mock_enrichment), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url(proxy=None)), ): result = await run_yandex_detail_backfill( db, run_id=9, params={"batch_size": 10, "budget_sec": 3600} ) assert result.enriched == 1 # Verify AsyncSession was constructed with proxies=None call_kwargs = session_cls.call_args[1] assert call_kwargs.get("proxies") is None # --------------------------------------------------------------------------- # Tests: save_detail_enrichment (kit — same function the task above actually calls; # легаси `app.services.scrapers.yandex_detail` удалён #2277 финальный шаг миграции) # --------------------------------------------------------------------------- def test_save_detail_enrichment_maps_fields_to_update() -> None: """save_detail_enrichment calls db.execute with UPDATE and commits.""" from scraper_kit.providers.yandex.detail import ( DetailEnrichment, MetroStation, save_detail_enrichment, ) enrichment = DetailEnrichment( offer_id="12345", source_url="https://realty.yandex.ru/offer/12345/", rooms=2, area_m2=54.5, floor=3, total_floors=9, address="Екатеринбург, ул. Ленина, 10", description="Хорошая квартира", repair_state="standard", publish_date=None, views_total=42, publish_date_relative="вчера", agency_name="Агентство «Тест»", agency_founded_year=2005, agency_objects_count=150, metro_stations=[MetroStation(name="Чкаловская", walk_min=11)], photo_urls=["https://example.com/1.jpg", "https://example.com/2.jpg"], newbuilding_url="https://realty.yandex.ru/kupit/novostrojka/test-12345/", newbuilding_id="12345", ) db = MagicMock() result_mock = MagicMock() result_mock.rowcount = 1 db.execute.return_value = result_mock saved = save_detail_enrichment(db, listing_id=99, e=enrichment) assert saved is True db.execute.assert_called_once() db.commit.assert_called_once() call_args = db.execute.call_args params = call_args[0][1] assert params["rooms"] == 2 assert params["area_m2"] == 54.5 assert params["agency_name"] == "Агентство «Тест»" metro_val = params["metro_stations"] assert metro_val is not None parsed_metro = json.loads(metro_val) assert parsed_metro[0]["name"] == "Чкаловская" assert parsed_metro[0]["walk_min"] == 11 photo_val = params["photo_urls"] assert photo_val is not None parsed_photos = json.loads(photo_val) assert len(parsed_photos) == 2 assert parsed_photos[0] == "https://example.com/1.jpg" def test_save_detail_enrichment_empty_metro_and_photos() -> None: """Empty metro_stations and photo_urls -> NULL passed for both jsonb columns.""" from scraper_kit.providers.yandex.detail import DetailEnrichment, save_detail_enrichment enrichment = DetailEnrichment( offer_id="99", source_url="https://realty.yandex.ru/offer/99/", ) db = MagicMock() result_mock = MagicMock() result_mock.rowcount = 1 db.execute.return_value = result_mock save_detail_enrichment(db, listing_id=1, e=enrichment) params = db.execute.call_args[0][1] assert params["metro_stations"] is None assert params["photo_urls"] is None def test_save_detail_enrichment_rowcount_zero_returns_false() -> None: """rowcount=0 (listing_id not found) -> returns False.""" from scraper_kit.providers.yandex.detail import DetailEnrichment, save_detail_enrichment enrichment = DetailEnrichment( offer_id="404", source_url="https://realty.yandex.ru/offer/404/", ) db = MagicMock() result_mock = MagicMock() result_mock.rowcount = 0 db.execute.return_value = result_mock saved = save_detail_enrichment(db, listing_id=404, e=enrichment) assert saved is False # --------------------------------------------------------------------------- # Очередь не должна содержать того, что парсер отвергает до сети (2026-08-06) # --------------------------------------------------------------------------- # Реальные source_url с прода (2026-08-06). Верх очереди на момент прогона 3300 # состоял ровно из таких строк: 5 попыток, 5 parse-None, abort за 23 секунды. _PROD_QUEUE_HEAD = [ ("https://macroserver.ru/id/224566/", False), ("https://prospect-federation.ru/flat/192", False), ("https://macroserver.ru/id/7223953/", False), ("https://strana.com/ekaterinburg/flat/1234", False), ("https://realty.yandex.ru/offer/7416316701146842927/", True), ("https://realty.yandex.ru/offer/7298311881327827251/", True), ] @pytest.mark.asyncio async def test_queue_gate_matches_parser_gate_and_counts_rest() -> None: """Снапшот-SELECT судит по тому же признаку, что и парсер, — сторожем, а не на слово. `YandexDetailScraper.parse` возвращает None по регулярке в URL, ещё не заглянув в HTML. Строки шире этого условия гарантированно дают parse-None и пачкой выбивают брейкер «5 подряд», обрывая ВЕСЬ прогон (32 прогона из 53 на проде). Проверяем на одних и тех же прод-URL обе стороны + что отброшенное посчитано, а не молча исчезло. """ from scraper_kit.providers.yandex.detail import YandexDetailScraper db = _mock_db([], unenrichable=3535) runs = MagicMock() session_cls, _session = _make_session_ctx([]) with ( patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=42, params={"batch_size": 10, "budget_sec": 60} ) snapshot_call = db.execute.call_args_list[0] assert "source_url ~ CAST(:offer_url_pattern AS text)" in str(snapshot_call.args[0]) pattern = snapshot_call.args[1]["offer_url_pattern"] scraper = YandexDetailScraper() for url, enrichable in _PROD_QUEUE_HEAD: assert (re.search(pattern, url) is not None) is enrichable, url if not enrichable: # HTML тут любой: отказ предрешён до его разбора. assert scraper.parse("сайт застройщика", url) is None, url assert result.unenrichable_pending == 3535 assert runs.mark_done.call_args.args[2]["unenrichable_pending"] == 3535 # --------------------------------------------------------------------------- # «Непригодно» — ярлык, а не диагноз (2026-08-12) # --------------------------------------------------------------------------- def _render_canonical_sql(offer_id: str) -> str: """Считает CANONICAL_URL_SQL как строку: '||' — конкатенация, source_id — значение.""" parts = [p.strip() for p in CANONICAL_URL_SQL.split("||")] return "".join(offer_id if p == "source_id" else p.strip("'") for p in parts) def test_recovered_url_equals_producer_canonical_form() -> None: """Адрес, вычисленный из source_id, — тот же, что пишет продюсер, и парсер его примет. Шесть прогонов подряд unenrichable_pending равнялся ровно 3535 не потому, что счётчик застыл (SELECT живой), а потому что множество замкнуто: продюсер после #2235 таких строк больше не создаёт, а выйти оттуда нельзя — source_url пишется только при вставке. Ярлык «непригодны» был неверен: у всех есть offerId, и по собранному из него URL страница парсится (прод-проба 2026-08-12, 6/6). Сторож держит ровно это: формула восстановления в SQL не должна разъехаться с `_canonical_source_url` продюсера, а результат — пройти гейт парсера. """ from scraper_kit.providers.yandex.serp import _canonical_source_url offer_id = "7416316697684470413" # реальный source_id прод-строки с macroserver.ru recovered = _render_canonical_sql(offer_id) assert re.match(OFFER_ID_PATTERN, offer_id) assert re.search(OFFER_URL_PATTERN, recovered), recovered for stored_url, is_offer_url in _PROD_QUEUE_HEAD: if is_offer_url: continue # у этих сохранённый адрес уже канонический, чинить нечего assert _canonical_source_url(stored_url, offer_id) == recovered, stored_url @pytest.mark.asyncio async def test_pending_counter_split_by_reason() -> None: """Остаток очереди делится по ПРИЧИНЕ, и восстановимое не зовётся непригодным. Одно число на две разные судьбы («адрес чиним» и «адресовать нечем») читается как «тут делать нечего» — так 3535 квартир простояли неделю вне обогащения. """ db = _mock_db([], unenrichable=0, url_from_offer_id=3535) runs = MagicMock() session_cls, _session = _make_session_ctx([]) with ( patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()), ): result = await run_yandex_detail_backfill( db, run_id=43, params={"batch_size": 10, "budget_sec": 60} ) assert result.url_from_offer_id == 3535 assert result.unenrichable_pending == 0 counters = runs.mark_done.call_args.args[2] assert counters["url_from_offer_id"] == 3535 assert counters["unenrichable_pending"] == 0 # Снапшот-SELECT берёт такие строки в работу по вычисленному адресу, а не # выбрасывает: без этой ветки они не попадут в очередь никогда. snapshot_sql = str(db.execute.call_args_list[0].args[0]) assert "OR source_id ~ CAST(:offer_id_pattern AS text)" in snapshot_sql assert CANONICAL_URL_SQL in snapshot_sql