gendesign/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py

"""Резюм 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 территория после отказа потеряна навсегда"
    )