"""Провайдер-уровень: детект-правило «капча vs честная пустая выдача» (#2625). Issue #2625: капча Циана и пустые выдачи Яндекса засчитывались как успешный прогон (status='done', errors_count=0, lots_fetched=0) — не было сигнала для run-level банана (см. `scraper_kit.orchestration.pipeline.run_cian_city_sweep` / `run_yandex_city_sweep` / `run_cian_full_load` / `run_yandex_full_load`, которые читают эти счётчики). Это НЕ тесты оркестрации (те — test_scraper_kit_pipeline_parity2.py:: test_*_extraction_failed_marks_banned и соседи) — здесь проверяется именно низкоуровневое правило на самом scraper-инстансе: Cian: extract_state() вернул None → state_extraction_failures++ (капча/смена вёрстки). Валидный state с offers=[] → НЕ failure (честная пустая выдача). Yandex: gate-API payload/pager так и не извлеклись после retries (tarpit / JSON-ошибка / gate-error-payload / отсутствие response.search.offers) → gate_fetch_failures++. Валидный payload с entities=[] → НЕ failure. Без сети, без БД, без browser_fetcher — прямые вызовы sync/async-методов scraper'ов. """ from __future__ import annotations import types from unittest.mock import AsyncMock import pytest def _cian_config() -> types.SimpleNamespace: return types.SimpleNamespace(glitchtip_dsn=None) # ── Cian: Redux-state extraction (_extract_state_tracked) ───────────────────── def test_cian_parse_serp_html_captcha_marks_failure() -> None: """extract_state() → None (капча/смена вёрстки) → attempts=1, failures=1.""" from scraper_kit.providers.cian.serp import CianScraper scraper = CianScraper(_cian_config()) lots = scraper._parse_serp_html("captcha page, no window._cianConfig") assert lots == [] assert scraper.state_extraction_attempts == 1 assert scraper.state_extraction_failures == 1 def test_cian_parse_serp_html_honest_empty_not_a_failure(monkeypatch: pytest.MonkeyPatch) -> None: """Валидный state с offers=[] (честная пустая выдача) → attempts=1, failures=0.""" from scraper_kit.providers.cian import serp as cian_serp monkeypatch.setattr( cian_serp, "extract_state", lambda html, mfe, key: {"results": {"offers": [], "totalOffers": 0}}, ) scraper = cian_serp.CianScraper(_cian_config()) lots = scraper._parse_serp_html("valid empty SERP") assert lots == [] assert scraper.state_extraction_attempts == 1 assert scraper.state_extraction_failures == 0 def test_cian_mixed_captcha_then_honest_empty_only_first_counts_as_failure( monkeypatch: pytest.MonkeyPatch, ) -> None: """Партиция «не все запросы прогона провалились» на уровне одного scraper'а: один captcha-fail + один честный 0 → attempts=2, failures=1 (не 2).""" from scraper_kit.providers.cian import serp as cian_serp scraper = cian_serp.CianScraper(_cian_config()) monkeypatch.setattr(cian_serp, "extract_state", lambda html, mfe, key: None) scraper._parse_serp_html("captcha") monkeypatch.setattr( cian_serp, "extract_state", lambda html, mfe, key: {"results": {"offers": [], "totalOffers": 0}}, ) scraper._parse_serp_html("honest empty") assert scraper.state_extraction_attempts == 2 assert scraper.state_extraction_failures == 1 def test_cian_extract_total_offers_shares_same_tracked_counters( monkeypatch: pytest.MonkeyPatch, ) -> None: """_extract_total_offers (probe path, full_load bisection) — тот же trackер, что _parse_serp_html (SERP-парсинг): оба должны учитываться в run-level детекте.""" from scraper_kit.providers.cian import serp as cian_serp monkeypatch.setattr(cian_serp, "extract_state", lambda html, mfe, key: None) scraper = cian_serp.CianScraper(_cian_config()) total = scraper._extract_total_offers("captcha on probe") assert total is None assert scraper.state_extraction_attempts == 1 assert scraper.state_extraction_failures == 1 # ── Yandex: gate-API structure extraction (_track_gate_result) ──────────────── class _FakeGateBrowser: """BrowserFetcher-заглушка: fetch() возвращает заранее заданные тела ответа (по одному на вызов, FIFO) — camoufox-обёртку

{...}

не эмулируем, _http_get достаёт JSON через первый `{` fallback (см. _extract_json_from_content).""" def __init__(self, bodies: list[str]) -> None: self._bodies = list(bodies) self.calls = 0 async def fetch(self, url: str) -> str: self.calls += 1 return self._bodies.pop(0) def _yandex_scraper_with_bodies(bodies: list[str]) -> object: from scraper_kit.providers.yandex.serp import YandexRealtyScraper scraper = YandexRealtyScraper(types.SimpleNamespace()) scraper._browser = _FakeGateBrowser(bodies) # type: ignore[assignment] return scraper class _RaisingBrowser: """BrowserFetcher-заглушка: fetch() ВСЕГДА поднимает исключение (transport failure — сеть/browser сбой, НЕ content-ответ) — deep-review fix 1 (#2600).""" async def fetch(self, url: str) -> str: raise RuntimeError("connection reset by peer") def _yandex_scraper_with_raising_browser() -> object: from scraper_kit.providers.yandex.serp import YandexRealtyScraper scraper = YandexRealtyScraper(types.SimpleNamespace()) scraper._browser = _RaisingBrowser() # type: ignore[assignment] return scraper @pytest.mark.asyncio async def test_yandex_fetch_page_json_gate_error_marks_failure() -> None: """gate-error payload (нет response.search.offers) → attempts=1, failures=1.""" scraper = _yandex_scraper_with_bodies(['{"error": "captcha"}']) payload = await scraper._fetch_page_json(None, 1, None, None) assert payload is None assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 1 @pytest.mark.asyncio async def test_yandex_fetch_page_json_honest_empty_not_a_failure() -> None: """Валидный payload с entities=[] (честная пустая выдача) → attempts=1, failures=0.""" body = ( '{"response": {"search": {"offers": ' '{"entities": [], "pager": {"page": 0, "totalItems": 0, "totalPages": 0}}}}}' ) scraper = _yandex_scraper_with_bodies([body]) payload = await scraper._fetch_page_json(None, 1, None, None) assert payload is not None assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 0 @pytest.mark.asyncio async def test_yandex_mixed_tarpit_then_honest_empty_only_first_counts_as_failure() -> None: """Один gate-error fail + один честный 0 → attempts=2, failures=1 (не 2).""" ok_body = ( '{"response": {"search": {"offers": ' '{"entities": [], "pager": {"page": 0, "totalItems": 0, "totalPages": 0}}}}}' ) scraper = _yandex_scraper_with_bodies(['{"error": "captcha"}', ok_body]) await scraper._fetch_page_json(None, 1, None, None) await scraper._fetch_page_json(None, 2, None, None) assert scraper.gate_fetch_attempts == 2 assert scraper.gate_fetch_failures == 1 # ── Code-review addendum (#2625): response-без-offers schema drift ──────────── # # {"response": {...}} без вложенного search.offers (schema drift / заглушка) # проходит _is_gate_error как «не ошибка» (есть "response", нет "error"), но # _extract_gate_data на нём вернёт None. Гэп был в двух местах: # 1. _fetch_page_json (probe/degraded/leaf — full_load путь) трекал ok=True # по одному лишь _is_gate_error, не проверяя реальное наличие offers. # 2. fetch_around page>=2 (city_sweep пагинация) — то же самое. # fetch_around_multi_room page=1 уже делал это правильно (эталон). @pytest.mark.asyncio async def test_yandex_fetch_page_json_schema_drift_marks_failure_not_success() -> None: """(a) response есть, но БЕЗ search.offers → _is_gate_error пропускает, _extract_gate_data проваливается → failure затрекан (не честный успех).""" body = '{"response": {"someOtherField": 1}}' scraper = _yandex_scraper_with_bodies([body]) payload = await scraper._fetch_page_json(None, 1, None, None) assert payload is not None # контракт возврата не меняется assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 1 @pytest.mark.asyncio async def test_yandex_fetch_around_page_ge2_schema_drift_marks_failure_anti_flap() -> None: """(b) page>=2 в fetch_around (пагинация внутри fetch_around_multi_room) — тот же drift-shape ПОСЛЕ успешной page=1 → failure затрекан, а не «честный конец выдачи». attempts=2, failures=1 — партиция НЕ триггерит orchestration banned-gate (тот требует failures==attempts, см. run_yandex_city_sweep).""" ok_body = ( '{"response": {"search": {"offers": ' '{"entities": [{"offerId": "1", "price": {"value": 5000000}}], ' '"pager": {"page": 0, "totalItems": 1, "totalPages": 2}}}}}' ) drift_body = '{"response": {"someOtherField": 1}}' scraper = _yandex_scraper_with_bodies([ok_body, drift_body]) scraper.request_delay_sec = 0.0 # skip real inter-request sleep in test lots_p1 = await scraper.fetch_around(56.84, 60.60, page=1) lots_p2 = await scraper.fetch_around(56.84, 60.60, page=2) assert len(lots_p1) == 1 assert lots_p2 == [] assert scraper.gate_fetch_attempts == 2 assert scraper.gate_fetch_failures == 1 # только page2, не весь прогон @pytest.mark.asyncio async def test_yandex_fetch_page_json_one_request_one_attempt_no_double_count() -> None: """(c) Инвариант «1 запрос = 1 attempt»: caller (как _probe) сам вызывает _extract_gate_data на уже полученном payload — это НЕ второй track-вызов, счётчик мутируется только внутри _fetch_page_json.""" from scraper_kit.providers.yandex.serp import _extract_gate_data body = '{"response": {"someOtherField": 1}}' scraper = _yandex_scraper_with_bodies([body]) payload = await scraper._fetch_page_json(None, 1, None, None) assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 1 # Caller-side re-check (то, что реально делает _probe) — не трогает счётчики. result = _extract_gate_data(payload) if payload is not None else None assert result is None assert scraper.gate_fetch_attempts == 1 # ── __aexit__: report_ban на 100% failure (#2600 п.1 + deep-review fix 1) ────── # # Здесь репортится РАНЬШЕ, чем pipeline.py's run-level banned-статус (#2625) — в # __aexit__ ДО release lease (см. CianScraper/YandexRealtyScraper __aexit__ # docstring-комментарии). Floor attempts>=_MIN_ATTEMPTS_FOR_BAN_REPORT (=3, deep- # review fix 1) — единичный admin-прогон (ровно 1 fetch_around) не должен банить # здоровый узел на 1/1=100%. class _FakeBrowserLease: """BrowserFetcher-заглушка с report_ban-recorder + async __aexit__ (для CianScraper/YandexRealtyScraper.__aexit__, который её awaits).""" def __init__(self) -> None: self.banned: list[str] = [] self.aexit_called = False def report_ban(self, reason: str) -> None: self.banned.append(reason) async def __aexit__(self, *args: object) -> None: self.aexit_called = True @pytest.mark.asyncio async def test_cian_aexit_reports_ban_on_all_attempts_failed() -> None: from scraper_kit.providers.cian.serp import CianScraper scraper = CianScraper(_cian_config()) for _ in range(3): # floor: attempts >= _MIN_ATTEMPTS_FOR_BAN_REPORT scraper._parse_serp_html("captcha page, no window._cianConfig") assert scraper.state_extraction_attempts == 3 assert scraper.state_extraction_failures == 3 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned # report_ban вызван ДО release (__aexit__ дошёл до конца) assert fake_browser.aexit_called # release всё равно случился (browser.__aexit__) @pytest.mark.asyncio async def test_cian_aexit_no_ban_report_below_attempts_floor() -> None: """deep-review fix 1: attempts=1 (единичный admin-прогон, 100% failure) НЕ репортит бан — floor attempts>=_MIN_ATTEMPTS_FOR_BAN_REPORT его не пускает.""" from scraper_kit.providers.cian.serp import CianScraper scraper = CianScraper(_cian_config()) scraper._parse_serp_html("captcha page, no window._cianConfig") assert scraper.state_extraction_attempts == 1 assert scraper.state_extraction_failures == 1 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == [] @pytest.mark.asyncio async def test_cian_aexit_no_ban_report_on_honest_empty(monkeypatch: pytest.MonkeyPatch) -> None: """Честная пустая выдача (attempts>0, failures=0) — report_ban НЕ вызывается.""" from scraper_kit.providers.cian import serp as cian_serp monkeypatch.setattr( cian_serp, "extract_state", lambda html, mfe, key: {"results": {"offers": [], "totalOffers": 0}}, ) scraper = cian_serp.CianScraper(_cian_config()) scraper._parse_serp_html("valid empty SERP") assert scraper.state_extraction_failures == 0 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == [] @pytest.mark.asyncio async def test_cian_aexit_no_ban_report_when_no_attempts_made() -> None: """attempts=0 (scraper упал до первого fetch) — guard attempts>0 не даёт ложного 100%-failure на пустой выборке из нуля попыток.""" from scraper_kit.providers.cian.serp import CianScraper scraper = CianScraper(_cian_config()) assert scraper.state_extraction_attempts == 0 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == [] @pytest.mark.asyncio async def test_yandex_aexit_reports_ban_on_all_attempts_failed() -> None: scraper = _yandex_scraper_with_bodies(['{"error": "captcha"}'] * 3) for _ in range(3): # floor: attempts >= _MIN_ATTEMPTS_FOR_BAN_REPORT await scraper._fetch_page_json(None, 1, None, None) assert scraper.gate_fetch_attempts == 3 assert scraper.gate_fetch_failures == 3 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned assert fake_browser.aexit_called @pytest.mark.asyncio async def test_yandex_aexit_no_ban_report_below_attempts_floor() -> None: """deep-review fix 1: attempts=1 (единичный admin-прогон, 100% failure) НЕ репортит бан — тот же floor, что cian.""" scraper = _yandex_scraper_with_bodies(['{"error": "captcha"}']) await scraper._fetch_page_json(None, 1, None, None) assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 1 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == [] @pytest.mark.asyncio async def test_yandex_aexit_no_ban_report_on_honest_empty() -> None: body = ( '{"response": {"search": {"offers": ' '{"entities": [], "pager": {"page": 0, "totalItems": 0, "totalPages": 0}}}}}' ) scraper = _yandex_scraper_with_bodies([body]) await scraper._fetch_page_json(None, 1, None, None) assert scraper.gate_fetch_failures == 0 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == [] # ── Yandex: transport vs content failure (deep-review fix 1, #2600 п.4) ──────── # # _http_get: fetch() raising (no response at all) -> transport_error=True. # Callers (_fetch_page_json/fetch_around/fetch_around_multi_room's page=1 probe) # must NOT feed transport_error=True into _track_gate_result — that counter feeds # report_ban (пул/бан), a transport failure is "наш прокси сдох"/сетевой сбой # (mark_health(ok=False) already covers it inside BrowserFetcher._post_fetch). @pytest.mark.asyncio async def test_http_get_transport_exception_sets_transport_error_flag() -> None: scraper = _yandex_scraper_with_raising_browser() resp = await scraper._http_get("https://realty.yandex.ru/gate/x", timeout=60) assert resp.status_code == 0 assert resp.transport_error is True @pytest.mark.asyncio async def test_http_get_no_json_found_is_not_transport_error() -> None: """Ответ пришёл (HTTP 200-эквивалент camoufox), но JSON не извлёкся — content ambiguous (тарпит-страница), НЕ transport_error.""" scraper = _yandex_scraper_with_bodies(["no json here"]) resp = await scraper._http_get("https://realty.yandex.ru/gate/x", timeout=60) assert resp.status_code == 0 assert resp.transport_error is False @pytest.mark.asyncio async def test_yandex_fetch_page_json_transport_error_does_not_track() -> None: """(a) fetch() raised (transport) — _fetch_page_json НЕ инкрементит gate_fetch_*.""" scraper = _yandex_scraper_with_raising_browser() payload = await scraper._fetch_page_json(None, 1, None, None) assert payload is None assert scraper.gate_fetch_attempts == 0 assert scraper.gate_fetch_failures == 0 @pytest.mark.asyncio async def test_yandex_fetch_around_all_transport_failures_not_tracked( monkeypatch: pytest.MonkeyPatch, ) -> None: """(a) fetch_around: ВСЕ retry-попытки — transport exception (сеть мертва целиком) -> retries-exhausted НЕ репортится как gate-failure (had_content_failure=False).""" from scraper_kit.providers.yandex import serp as yandex_serp monkeypatch.setattr(yandex_serp.asyncio, "sleep", AsyncMock()) # без реальных 2s×N scraper = _yandex_scraper_with_raising_browser() lots = await scraper.fetch_around(56.8, 60.6, page=1) assert lots == [] assert scraper.gate_fetch_attempts == 0 assert scraper.gate_fetch_failures == 0 @pytest.mark.asyncio async def test_yandex_fetch_around_mixed_transport_then_content_failure_is_tracked( monkeypatch: pytest.MonkeyPatch, ) -> None: """Микс: первая попытка — transport exception (не считается), retry рвёт JSON (content-сигнал, тарпит) -> had_content_failure=True -> retries-exhausted ВСЁ ЖЕ репортится (хотя бы одна попытка дала реальный content-сигнал).""" from scraper_kit.providers.yandex import serp as yandex_serp from scraper_kit.providers.yandex.serp import YandexRealtyScraper monkeypatch.setattr(yandex_serp.asyncio, "sleep", AsyncMock()) # без реальных 2s×N class _FlakyThenTarpitBrowser: def __init__(self) -> None: self.calls = 0 async def fetch(self, url: str) -> str: self.calls += 1 if self.calls == 1: raise RuntimeError("transient network blip") return "no json here (tarpit)" scraper = YandexRealtyScraper(types.SimpleNamespace()) scraper._browser = _FlakyThenTarpitBrowser() # type: ignore[assignment] lots = await scraper.fetch_around(56.8, 60.6, page=1) assert lots == [] assert scraper.gate_fetch_attempts == 1 assert scraper.gate_fetch_failures == 1 @pytest.mark.asyncio async def test_yandex_end_to_end_transport_errors_never_report_ban() -> None: """(a) end-to-end: 3 подряд transport-провала (сеть мертва) — gate_fetch_attempts остаётся 0, __aexit__ guard (attempts>=floor) не срабатывает -> report_ban НЕ вызывается. Отличимо от настоящего 3x content-бана (test_yandex_aexit_reports_ ban_on_all_attempts_failed выше, ГДЕ attempts=3 и banned непусто).""" scraper = _yandex_scraper_with_raising_browser() for _ in range(3): await scraper._fetch_page_json(None, 1, None, None) assert scraper.gate_fetch_attempts == 0 fake_browser = _FakeBrowserLease() scraper._browser = fake_browser # type: ignore[assignment] await scraper.__aexit__(None, None, None) assert fake_browser.banned == []