"""Резюм exhaustive-обхода Яндекса не пробивает уже пройденную территорию (#3359). То же, что #3315 у avito: гейт `should_skip` есть в общем движке, но предикат из done-леджера строил только avito — у yandex дерево бисекции на резюме спускалось внутрь зачтённых полос живыми probe-запросами к gate-API. Формат ключей У YANDEX ДРУГОЙ: `_combo_label` даёт «rooms:lo-hi» (разделитель границ «-», а не «:») и «rooms:lo-None» для открытого верхнего брекета («None» вместо «open»). Ключи НЕ подгоняются под avito (это протухило бы живые чекпоинты) — параметризован парсер (`range_sep`/`open_token`). Сеть не нужна: горлышко фетча `_fetch_page_json` подменяется счётчиком; probe читает totalItems из подсунутого gate-payload'а. Проверка ПО ЗНАЧЕНИЮ — сколько раз обход сходил в сеть. """ from __future__ import annotations import asyncio import os from typing import Any os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") from scraper_kit.base import ScrapedLot from scraper_kit.providers.yandex.serp import YandexRealtyScraper from app.services.scraper_adapters import RealScraperConfig _ROOMS = "2" # Плотная выдача: totalItems=2000 > cap=500 → непокрытый диапазон обязан делиться. _DENSE_PAGE: dict[str, Any] = { "response": { "search": { "offers": { "entities": [], "pager": {"totalItems": 2000, "totalPages": 100, "page": 0}, } } } } def _scraper() -> tuple[YandexRealtyScraper, list[int]]: """Скрапер с заглушенной сетью; в списке — счётчик реальных фетчей.""" s = YandexRealtyScraper(RealScraperConfig()) calls = [0] async def fake_fetch( rooms: str | None, page: int, price_min: int | None, price_max: int | None, new_flat: str = "NO", ) -> dict[str, Any]: calls[0] += 1 return _DENSE_PAGE s._fetch_page_json = fake_fetch # type: ignore[method-assign] s.request_delay_sec = 0.0 return s, calls def _walk(lo: int, hi: int | None, skip_buckets: set[str] | None) -> int: """Прогнать бисекцию полосы [lo, hi] и вернуть ЧИСЛО сетевых запросов.""" s, calls = _scraper() seen: dict[str, ScrapedLot] = {} asyncio.run( s._walk_price_range( rooms=_ROOMS, lo=lo, hi=hi, seen=seen, price_cap_per_bucket=500, max_pages_per_bucket=1, skip_buckets=skip_buckets, ) ) return calls[0] def test_subrange_of_done_bucket_costs_zero_requests() -> None: """Головной: полоса ВНУТРИ done-корзины не делает ни одного запроса.""" calls = _walk(4_062_500, 4_124_999, {f"{_ROOMS}:4000000-4999999"}) assert calls == 0, ( f"полоса [4062500, 4124999] целиком внутри зачтённой [4000000, 4999999], " f"а обход сходил в сеть {calls} раз(а) — бан-бюджет горит на готовой территории" ) def test_partially_covered_range_still_probes() -> None: """Контроль честности skip'а: непокрытый остаток обязан пробиваться.""" calls = _walk(3_500_000, 4_200_000, {f"{_ROOMS}:4000000-4999999"}) assert calls > 0, "частично покрытая полоса пропущена целиком — потеря инвентаря" def test_empty_ledger_keeps_previous_behaviour() -> None: """Регресс-контроль: без леджера обход прежний — пробивает и делит.""" baseline = _walk(4_000_000, 4_999_999, None) assert baseline > 1, f"обход без леджера деградировал: {baseline} запрос(ов)" assert _walk(4_000_000, 4_999_999, set()) == baseline # Чужая комнатность в леджере не покрывает нашу. assert _walk(4_000_000, 4_999_999, {"3:0-None"}) == baseline # Ключ incremental-режима (gate-combo с префиксом сегмента) покрытия не даёт: # он не проходит отсев по префиксу label, а смешение режимов дополнительно # блокирует _pick_resume (params IS NOT DISTINCT FROM). assert _walk(4_000_000, 4_999_999, {f"secondary/{_ROOMS}:0-None"}) == baseline def test_fully_done_room_resumes_with_zero_requests() -> None: """Приёмка issue: резюм по полностью готовой комнатности = НОЛЬ запросов. Открытый верхний брекет в леджере yandex'а записан как `rooms:lo-None`. """ s, calls = _scraper() asyncio.run( s.fetch_all_secondary( rooms_buckets=[_ROOMS], max_pages_per_bucket=1, skip_buckets={f"{_ROOMS}:0-None"}, ) ) assert calls[0] == 0, f"резюм готовой комнатности сделал {calls[0]} запрос(ов) вместо нуля" def _walk_degraded(skip_buckets: set[str] | None) -> tuple[list[tuple[str, bool]], int]: """Прогон с мёртвой сетью (probe провалился → degraded-ветка). Возвращает (что бакет-колбэк отметил, число сетевых запросов). Колбэк — той же формы, что pipeline._on_bucket: третий позиционный аргумент = признак полноты. """ s, calls = _scraper() async def dead_fetch(*_a: Any, **_k: Any) -> None: calls[0] += 1 return None async def no_rotate() -> bool: return False s._fetch_page_json = dead_fetch # type: ignore[method-assign] s._rotate_ip = no_rotate # type: ignore[method-assign] marked: list[tuple[str, bool]] = [] def on_bucket(key: str, _lots: list[Any], complete: bool = True) -> None: marked.append((key, complete)) asyncio.run( s._walk_price_range( rooms=_ROOMS, lo=4_000_000, hi=4_999_999, seen={}, price_cap_per_bucket=500, max_pages_per_bucket=1, on_bucket=on_bucket, skip_buckets=skip_buckets, ) ) return marked, calls[0] def test_degraded_bucket_stays_out_of_done_ledger() -> None: """Бакет с провалившимся probe НЕ зачитывается как пройденный.""" marked, _ = _walk_degraded(None) assert marked, "degraded-ветка не вызвала on_bucket — тест ничего не проверяет" ledger = {key for key, complete in marked if complete} assert not ledger, ( f"бакеты {sorted(ledger)} собраны degraded-пагинацией (полнота неизвестна), " "но помечены complete — в леджере их интервал сольётся с соседними и резюм " "не переобойдёт недобор" ) def test_degraded_band_is_rewalked_on_resume() -> None: """По значению: леджер после degraded-прогона не гасит эту полосу на резюме.""" marked, _ = _walk_degraded(None) ledger = {key for key, complete in marked if complete} _, calls = _walk_degraded(ledger or None) assert calls > 0, ( "резюм не сделал ни одного запроса по полосе, собранной лишь частично — " "best-effort территория после отказа потеряна навсегда" )