"""Тесты 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 httpx import pytest from app.services.scrapers.avito import ( _AVITO_MAX_PRICE, _AVITO_MIN_BRACKET, _AVITO_OFFERS_PER_PAGE, AvitoScraper, ) from app.services.scrapers.avito_exceptions import AvitoBlockedError 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, ) # ── Probe-устойчивость к browser-ошибкам (HTTP 500 / timeout) ───────────────── def _http_500_error() -> httpx.HTTPStatusError: """Сборка httpx.HTTPStatusError(500) как из browser_fetcher.raise_for_status().""" request = httpx.Request("POST", "http://tradein-browser:3000/fetch") response = httpx.Response(500, request=request) return httpx.HTTPStatusError("Server error '500'", request=request, response=response) @pytest.mark.asyncio async def test_probe_browser_500_on_wide_bucket_splits_not_crash( scraper: AvitoScraper, ) -> None: """Probe бросает HTTP 500 на ШИРОКОМ bucket → НЕ крах, происходит split. Воспроизводит прод-инцидент run 199: студии pmax=200M таймаутят на goto → browser HTTP 500. Раньше это роняло весь full-load с 0 uniq. Теперь широкий bucket принудительно делится пополам; под-бакеты пагинируются нормально. """ probe_calls: list[tuple[int | None, int]] = [] async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE: probe_calls.append((min_price, max_price)) raise _http_500_error() return f"page={page} min={min_price} max={max_price}" def fake_total(html: str) -> int | None: return 50 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 max_m = re.search(r"max=(\w+)", html) max_p = max_m.group(1) if max_m else "MAX" if page == 1: return [_make_lot(f"lot_{max_p}_{i}") for i in range(3)] 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=[("студии", "studii-X")], price_cap_per_bucket=1400, ) assert probe_calls, "Probe полного диапазона должен был вызваться и упасть" assert len(lots) > 0, f"Ожидались лоты из под-бакетов после split, got {len(lots)}" @pytest.mark.asyncio async def test_probe_timeout_on_wide_bucket_splits_not_crash( scraper: AvitoScraper, ) -> None: """httpx.TimeoutException на широком bucket обрабатывается как HTTP 500 → split.""" probe_failed = False async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: nonlocal probe_failed if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE: probe_failed = True raise httpx.TimeoutException("Page.goto timeout") return f"page={page} max={max_price}" def fake_total(html: str) -> int | None: return 40 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 max_m = re.search(r"max=(\w+)", html) max_p = max_m.group(1) if max_m else "MAX" return [_make_lot(f"lot_{max_p}_{i}") for i in range(2)] 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=[("студии", "studii-X")], price_cap_per_bucket=1400, ) assert probe_failed, "Probe должен был таймаутнуть" assert len(lots) > 0, "После split под-бакеты должны собрать лоты" @pytest.mark.asyncio async def test_probe_browser_500_on_narrow_bucket_skips_not_crash( scraper: AvitoScraper, ) -> None: """Probe бросает HTTP 500 на УЗКОМ bucket (< MIN_BRACKET) → skip+return, без краха. Узкий bucket делить бесполезно — хвост теряется осознанно. Прогон не падает, остальные комнатности продолжаются. """ second_room_paginated = False async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: nonlocal second_room_paginated if room_slug == "narrow-X": raise _http_500_error() if room_slug == "ok-X" and page == 1: second_room_paginated = True 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("ok_lot")] if "page=1" in html else [] narrow_lo = 1_000_000 narrow_hi = narrow_lo + _AVITO_MIN_BRACKET - 1 seen: dict[str, ScrapedLot] = {} 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="narrow-X", room_label="room_narrow", lo=narrow_lo, hi=narrow_hi, seen=seen, price_cap_per_bucket=1400, max_pages_per_bucket=100, ) assert seen == {}, "Узкий bucket с упавшим probe не должен дать лотов" await scraper._walk_price_range( room_slug="ok-X", room_label="room_ok", lo=0, hi=_AVITO_MAX_PRICE, seen=seen, price_cap_per_bucket=1400, max_pages_per_bucket=100, ) assert second_room_paginated, "Вторая комнатность должна пагинироваться после skip" assert "ok_lot" in seen @pytest.mark.asyncio async def test_probe_avito_blocked_propagates_not_swallowed( scraper: AvitoScraper, ) -> None: """AvitoBlockedError из probe ВСПЛЫВАЕТ (hard-block), а не глушится как total=None.""" async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: raise AvitoBlockedError("Avito SERP firewall (browser-mode) — IP banned") def fake_total(html: str) -> int | None: return 100 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=lambda h, source_url_base: []), patch.object(scraper, "_rotate_ip", return_value=False), ): with pytest.raises(AvitoBlockedError): await scraper.fetch_all_secondary( rooms_buckets=[("1-комн.", "1-komnatnye-X")], price_cap_per_bucket=1400, )