"""Резюм exhaustive-обхода Авито не пробивает уже пройденную территорию (#3315).
Прогон 5718 (01.09.2026) 21 минуту жёг бан-бюджет обеих живых нод внутри ЗАЧТЁННОЙ
территории: `room_studii:4000000:4999999` лежит в done-леджере, а лог показывает
probe поддиапазонов `[4062500, 4124999]`, `[4093750, 4124999]`. Причина — skip
проверялся в ЛИСТЕ, после probe: дерево деления пере-пробивалось живыми запросами
каждый прогон. Вторая грань: ключи чекпоинта — границы ДИНАМИЧЕСКОЙ бисекции, при
сдвиге рынка новые листья ключом не равны старым даже внутри покрытого диапазона,
поэтому сравнение ключей строкой на резюме бесполезно — нужны интервалы.
Сеть не нужна: `_fetch_serp_html` подменяется счётчиком. Проверка ПО ЗНАЧЕНИЮ —
сколько раз обход сходил в сеть; на origin/main первые три теста красные (обход
делает N > 0 запросов по территории, которая уже в чекпоинте).
"""
from __future__ import annotations
import asyncio
import os
from types import SimpleNamespace
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from scraper_kit.providers.avito.serp import ROOM_SLUGS, AvitoScraper
_STUDII = ROOM_SLUGS[0] # ("студии", "studii-…")
_LABEL = "room_studii"
# Плотная выдача: total=2000 > cap=1400 → непокрытый диапазон обязан делиться дальше.
# Карточка обязательна: total>0 при 0 распознанных карточек = DOM-drift guard (#audit-scrapers).
_CARD = (
''
)
_DENSE_SERP = f'2000{_CARD}'
def _walk(lo: int, hi: int | None, skip_buckets: set[str] | None) -> int:
"""Прогнать бисекцию полосы [lo, hi] и вернуть ЧИСЛО сетевых запросов."""
s = AvitoScraper(SimpleNamespace(avito_serp_ekb_only=True, scraper_fetch_mode="http")) # type: ignore[arg-type]
calls = 0
async def fake_fetch(url: str, page: int) -> str:
nonlocal calls
calls += 1
return _DENSE_SERP
s._fetch_serp_html = fake_fetch # type: ignore[method-assign]
asyncio.run(
s._walk_price_range(
room_slug=_STUDII[1],
room_label=_LABEL,
lo=lo,
hi=hi,
seen={},
price_cap_per_bucket=1400,
max_pages_per_bucket=1,
skip_buckets=skip_buckets,
)
)
return calls
def test_subrange_of_done_bucket_costs_zero_requests() -> None:
"""Головной: полоса ВНУТРИ done-корзины не делает ни одного запроса.
Ровно случай из лога 5718: ключ `room_studii:4000000:4999999` зачтён, дерево
спустилось в `[4062500, 4124999]`.
"""
calls = _walk(4_062_500, 4_124_999, {f"{_LABEL}:4000000:4999999"})
assert calls == 0, (
f"полоса [4062500, 4124999] целиком внутри зачтённой [4000000, 4999999], "
f"а обход сходил в сеть {calls} раз(а) — бан-бюджет горит на готовой территории"
)
def test_partially_covered_range_still_probes() -> None:
"""Контроль честности skip'а: непокрытый остаток обязан пробиваться.
`[3500000, 4200000]` покрыта done-корзиной только сверху — нижние 500k никто
не собирал, пропустить их = молча потерять инвентарь.
"""
calls = _walk(3_500_000, 4_200_000, {f"{_LABEL}:4000000:4999999"})
assert calls > 0, "частично покрытая полоса пропущена целиком — потеря инвентаря"
def test_adjacent_done_keys_merge_across_the_seam() -> None:
"""Смежные корзины [4М,5М) + [5М,6М) покрывают полосу через их стык.
Бисекция делит по `mid`, и половины ложатся встык (hi=mid, lo=mid+1): без
слияния смежных отрезков любая полоса, пересекающая шов, пробивалась бы заново.
"""
ledger = {f"{_LABEL}:4000000:4999999", f"{_LABEL}:5000000:5999999"}
calls = _walk(4_500_000, 5_500_000, ledger)
assert calls == 0, (
f"полоса [4.5М, 5.5М] лежит в объединении двух смежных done-корзин, "
f"а обход сходил в сеть {calls} раз(а) — стык не сшит"
)
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, {"room_1_komn:0:open"}) == baseline
def test_fully_done_room_resumes_with_zero_requests() -> None:
"""Приёмка issue: резюм по полностью готовой комнатности = НОЛЬ запросов.
Открытый верхний брекет в леджере записан как `label:lo:open` — это [lo, ∞).
"""
s = AvitoScraper(SimpleNamespace(avito_serp_ekb_only=True, scraper_fetch_mode="http")) # type: ignore[arg-type]
calls = 0
async def fake_fetch(url: str, page: int) -> str:
nonlocal calls
calls += 1
return _DENSE_SERP
s._fetch_serp_html = fake_fetch # type: ignore[method-assign]
asyncio.run(
s.fetch_all_secondary(
rooms_buckets=[_STUDII],
max_pages_per_bucket=1,
skip_buckets={f"{_LABEL}:0:open"},
)
)
assert calls == 0, f"резюм готовой комнатности сделал {calls} запрос(ов) вместо нуля"
def test_junk_key_in_ledger_neither_crashes_nor_covers() -> None:
"""Битый ключ рядом с валидным: обход не падает, покрытие даёт только валидный."""
ledger = {f"{_LABEL}:мусор:5", f"{_LABEL}:1000000:2000000"}
assert _walk(1_200_000, 1_300_000, ledger) == 0, "валидный ключ перестал покрывать"
assert _walk(999_999, 1_300_000, ledger) > 0, "битый ключ дорисовал несобранное покрытие"