"""Тесты exhaustive-загрузки Avito: fetch_all_secondary + _walk_price_range.
Зеркало tests/scrapers/test_cian_exhaustive.py. Мокаем сетевые вызовы
(_fetch_rooms_page_html, _extract_total_count, _parse_html) для детерминированного
контроля бисекции без реального HTTP.
Дополнительно: _build_rooms_url с pmin/pmax (backward-compat) и _extract_total_count
на реальном span-фрагменте `data-marker="page-title/count"`.
"""
from __future__ import annotations
import os
from unittest.mock import MagicMock, patch
# Settings requires DATABASE_URL at import time — set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import pytest
from app.services.scrapers.avito import (
_AVITO_MAX_PRICE,
_AVITO_MIN_BRACKET,
_AVITO_OFFERS_PER_PAGE,
AvitoScraper,
)
from app.services.scrapers.base import ScrapedLot
def _make_lot(source_id: str, price: int = 3_000_000, segment: str = "vtorichka") -> ScrapedLot:
"""Минимальный ScrapedLot для теста."""
return ScrapedLot(
source="avito",
source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/{source_id}",
source_id=source_id,
address="Екатеринбург",
price_rub=price,
listing_segment=segment,
)
@pytest.fixture
def scraper() -> AvitoScraper:
"""AvitoScraper без реального __aenter__ (без HTTP/IP-rotation)."""
s = AvitoScraper.__new__(AvitoScraper)
s.name = "avito"
s.base_url = "https://www.avito.ru"
s.request_delay_sec = 0.0
s._cffi = MagicMock() # не None
s._browser = None
s.parse_failures = 0
return s
# ── _build_rooms_url с pmin/pmax ──────────────────────────────────────────────
def test_build_rooms_url_both_prices(scraper: AvitoScraper) -> None:
"""min_price + max_price → pmin/pmax оба в query."""
url = scraper._build_rooms_url(
"1-komnatnye-X", page=2, min_price=3_000_000, max_price=4_000_000
)
assert "pmin=3000000" in url
assert "pmax=4000000" in url
assert "p=2" in url
assert "s=104" in url
assert "/kvartiry/prodam/1-komnatnye-X?" in url
def test_build_rooms_url_min_only(scraper: AvitoScraper) -> None:
"""Только min_price → pmin в query, pmax отсутствует."""
url = scraper._build_rooms_url("1-komnatnye-X", min_price=3_000_000)
assert "pmin=3000000" in url
assert "pmax" not in url
def test_build_rooms_url_no_prices_backward_compat(scraper: AvitoScraper) -> None:
"""Без цен URL не меняется (backward-compat для fetch_by_rooms)."""
url = scraper._build_rooms_url("1-komnatnye-X", page=1)
assert "pmin" not in url
assert "pmax" not in url
assert url.endswith("/kvartiry/prodam/1-komnatnye-X?s=104&p=1")
# ── _extract_total_count ──────────────────────────────────────────────────────
# Реальный span-фрагмент из живого Avito SERP (NBSP-разделитель разрядов).
_REAL_COUNT_SPAN = '1\xa0178'
def test_extract_total_count_real_span(scraper: AvitoScraper) -> None:
"""Реальный span с NBSP `1\\xa0178` → 1178."""
html = f"Квартиры
{_REAL_COUNT_SPAN}"
assert scraper._extract_total_count(html) == 1178
def test_extract_total_count_nbsp_entity(scraper: AvitoScraper) -> None:
"""Вариант с ` `-entity → корректно очищается."""
html = '12 345'
assert scraper._extract_total_count(html) == 12345
def test_extract_total_count_zero(scraper: AvitoScraper) -> None:
"""0 результатов → 0 (не None)."""
html = '0'
assert scraper._extract_total_count(html) == 0
def test_extract_total_count_missing_returns_none(scraper: AvitoScraper) -> None:
"""Captcha/firewall (нет span) → None."""
html = "Доступ ограничен"
assert scraper._extract_total_count(html) is None
# ── Бисекция: split / paginate / dedup ────────────────────────────────────────
@pytest.mark.asyncio
async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None:
"""total > cap → бакет делится; оба суб-бакета пагинируются."""
mid = _AVITO_MAX_PRICE // 2
call_log: list[tuple] = []
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
call_log.append((page, min_price, max_price))
return f"page={page} min={min_price} max={max_price}"
def fake_total(html: str) -> int | None:
import re
min_m = re.search(r"min=(\w+)", html)
max_m = re.search(r"max=(\w+)", html)
min_p = int(min_m.group(1)) if min_m and min_m.group(1) != "None" else 0
max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else _AVITO_MAX_PRICE
if max_p == mid:
return 800 # левая половина — пагинировать
if min_p == mid + 1:
return 700 # правая половина — пагинировать
return 3000 # весь диапазон — split
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
min_m = re.search(r"min=(\w+)", html)
max_m = re.search(r"max=(\w+)", html)
min_p = min_m.group(1) if min_m else "0"
max_p = max_m.group(1) if max_m else "MAX"
if page == 1:
return [
_make_lot(f"lot_{min_p}_{max_p}_1"),
_make_lot(f"lot_{min_p}_{max_p}_2"),
]
return []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X")],
price_cap_per_bucket=1400,
)
assert len(lots) > 0
# Должны быть пробы суб-бакетов (после split весь диапазон не пагинируется)
probed_max = [mx for (p, _mn, mx) in call_log if p == 1]
assert any(mx == mid for mx in probed_max), f"Ожидался probe левой половины, got {probed_max}"
@pytest.mark.asyncio
async def test_fetch_all_secondary_paginates_when_under_cap(scraper: AvitoScraper) -> None:
"""total <= cap → пагинирует бакет полностью."""
pages_fetched: list[int] = []
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
pages_fetched.append(page)
return f"page={page}"
def fake_total(html: str) -> int | None:
return 100 # ceil(100/50) = 2 страницы
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
if page <= 2:
return [_make_lot(f"lot_p{page}_{i}") for i in range(_AVITO_OFFERS_PER_PAGE)]
return []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("2-комн.", "2-komnatnye-X")],
price_cap_per_bucket=1400,
)
assert len(lots) == 100, f"Ожидалось 100 уникальных лотов, получено {len(lots)}"
assert 2 in pages_fetched, "Должна быть запрошена страница 2"
@pytest.mark.asyncio
async def test_fetch_all_secondary_min_bracket_guard(scraper: AvitoScraper) -> None:
"""hi - lo < MIN_BRACKET → пагинируем как есть (нет бесконечной рекурсии)."""
call_count = 0
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
nonlocal call_count
call_count += 1
if call_count > 200:
raise RuntimeError("Бесконечная рекурсия — MIN_BRACKET guard не сработал")
return f"page={page}"
def fake_total(html: str) -> int | None:
return 5000 # всегда > cap → без guard делилось бы бесконечно
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
return []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
await scraper._walk_price_range(
room_slug="1-komnatnye-X",
room_label="room_1_komn",
lo=1_000_000,
hi=1_000_000 + _AVITO_MIN_BRACKET - 1,
seen={},
price_cap_per_bucket=1400,
max_pages_per_bucket=100,
)
# Guard сработал: рекурсии нет, только один проход пагинации (probe + до cap страниц).
assert call_count <= 101, f"Слишком много вызовов ({call_count}) — guard не сработал"
@pytest.mark.asyncio
async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: AvitoScraper) -> None:
"""Параллельная пагинация не ломает дедуп по source_id."""
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
return f"page={page}"
def fake_total(html: str) -> int | None:
return 150 # ceil(150/50) = 3 страницы
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
if page == 3:
return [_make_lot(f"lot_p1_{i}") for i in range(10)] # дубликаты стр 1
return [_make_lot(f"lot_p{page}_{i}") for i in range(10)]
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X")],
price_cap_per_bucket=1400,
concurrency=5,
)
# Страницы 1+2 = 20 уникальных, страница 3 дублирует → 20
assert len(lots) == 20, f"Ожидалось 20 уникальных лотов (дедуп), получено {len(lots)}"
@pytest.mark.asyncio
async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None:
"""on_bucket получает (bucket_key, lots) — bucket_key = room_label:lo:hi."""
received: list[tuple[str, int]] = []
def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
received.append((bucket_key, len(lots)))
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
return f"page={page}"
def fake_total(html: str) -> int | None:
return 5 # 1 страница
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
return [_make_lot(f"lot_p{page}_{i}") for i in range(5)]
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
await scraper.fetch_all_secondary(
rooms_buckets=[("2-комн.", "2-komnatnye-X")],
price_cap_per_bucket=1400,
on_bucket=fake_on_bucket,
)
assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}"
key, count = received[0]
assert key == f"room_2_komn:0:{_AVITO_MAX_PRICE}", f"Неверный bucket_key: {key!r}"
assert count == 5
@pytest.mark.asyncio
async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None:
"""skip_buckets: probe выполняется, но пагинация и on_bucket пропускаются."""
fetch_calls: list[int] = []
on_bucket_calls: list[str] = []
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
fetch_calls.append(page)
return f"page={page}"
def fake_total(html: str) -> int | None:
return 5
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
return [_make_lot("lot_1")]
def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
on_bucket_calls.append(bucket_key)
skip_set = {f"room_1_komn:0:{_AVITO_MAX_PRICE}"}
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X")],
price_cap_per_bucket=1400,
on_bucket=fake_on_bucket,
skip_buckets=skip_set,
)
assert fetch_calls == [1], f"Ожидался ровно 1 вызов (probe), got {fetch_calls}"
assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакета"
assert lots == []
@pytest.mark.asyncio
async def test_secondary_only_drops_novostroyki(scraper: AvitoScraper) -> None:
"""secondary_only=True: новостройки (listing_segment=='novostroyki') отброшены."""
mixed = [
_make_lot("nb_1", segment="novostroyki"),
_make_lot("vt_1", segment="vtorichka"),
_make_lot("vt_2", segment="vtorichka"),
]
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
return f"page={page}"
def fake_total(html: str) -> int | None:
return 3
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
return mixed if page == 1 else []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X")],
price_cap_per_bucket=1400,
secondary_only=True,
)
ids = {lot.source_id for lot in lots}
assert "nb_1" not in ids
assert ids == {"vt_1", "vt_2"}
@pytest.mark.asyncio
async def test_on_bucket_cancel_stops_run(scraper: AvitoScraper) -> None:
"""on_bucket кидает RuntimeError('cancelled') → прогон прерывается."""
def cancel_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
raise RuntimeError("cancelled")
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
return f"page={page}"
def fake_total(html: str) -> int | None:
return 5
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
return [_make_lot("lot_1")]
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
with pytest.raises(RuntimeError, match="cancelled"):
await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X"), ("2-комн.", "2-komnatnye-X")],
price_cap_per_bucket=1400,
on_bucket=cancel_on_bucket,
)