"""Тесты exhaustive-загрузки Cian: fetch_all_secondary + _walk_price_range. Мокаем сетевые вызовы (_fetch_page_html, _extract_total_offers, sleep_between_requests) для детерминированного контроля поведения без реального HTTP. """ from __future__ import annotations from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.services.scrapers.base import ScrapedLot from app.services.scrapers.cian import _MAX_PRICE, _MIN_BRACKET, CianScraper def _make_lot(source_id: str, price: int = 3_000_000) -> ScrapedLot: """Минимальный ScrapedLot для теста.""" return ScrapedLot( source="cian", source_url=f"https://ekb.cian.ru/sale/flat/{source_id}/", source_id=source_id, address="Екатеринбург", price_rub=price, ) def _make_html_stub(total_offers: int, lot_ids: list[str]) -> str: """Заглушка HTML — не парсится напрямую; используется через мокнутые методы.""" return f"total={total_offers} ids={lot_ids}" @pytest.fixture def scraper() -> CianScraper: """CianScraper без реального __aenter__ (без HTTP warm-up).""" s = CianScraper.__new__(CianScraper) s.name = "cian" s.base_url = "https://ekb.cian.ru" s.request_delay_sec = 0.0 s._cffi = MagicMock() # не None → assert self._cffi is not None пройдёт return s @pytest.mark.asyncio async def test_fetch_all_secondary_splits_on_cap(scraper: CianScraper) -> None: """totalOffers > cap → бакет разбивается; в итоге оба суб-бакета пагинируются.""" # Настройка: rooms=(1,), один бакет [0, None] # Probe [0, None]: totalOffers=3000 > cap=1400 → split на [0, MAX_PRICE] # Probe [0, MAX_PRICE]: totalOffers=3000 > 1400 → split на [0, mid] и [mid+1, MAX_PRICE] # Probe [0, mid] (mid = MAX_PRICE//2 = 100_000_000): totalOffers=800 <= cap → пагинировать # Probe [mid+1, MAX_PRICE]: totalOffers=700 <= cap → пагинировать mid = _MAX_PRICE // 2 call_log: list[tuple] = [] async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: call_log.append((rooms, page, min_price, max_price)) return f"page={page} min={min_price} max={max_price}" def fake_extract_total_offers(html: str) -> int | None: # Парсим min/max из fake html 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 _MAX_PRICE # Имитируем разделение: весь диапазон > 1400, левая и правая половины <= 1400 if max_p is None or max_p >= _MAX_PRICE: if min_p == 0: return 3000 # весь диапазон — нужно split if max_p == mid: return 800 # левая половина — пагинировать if min_p == mid + 1: return 700 # правая половина — пагинировать return 3000 # всё остальное — split def fake_parse_serp_html(html: str) -> list[ScrapedLot]: import re page_m = re.search(r"page=(\d+)", html) min_m = re.search(r"min=(\w+)", html) max_m = re.search(r"max=(\w+)", html) page = int(page_m.group(1)) if page_m else 1 min_p = min_m.group(1) if min_m else "0" max_p = max_m.group(1) if max_m else "MAX" # Возвращаем по 2 лота на первой странице каждого пагинируемого бакета if page == 1: return [ _make_lot(f"lot_{min_p}_{max_p}_p{page}_1"), _make_lot(f"lot_{min_p}_{max_p}_p{page}_2"), ] return [] # страница 2+ — пусто → early stop async def fake_sleep(*_: object, **__: object) -> None: pass with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): lots = await scraper.fetch_all_secondary( rooms_buckets=[(1,)], price_cap_per_bucket=1400, ) # Должны получить лоты из ОБОИХ суб-бакетов после split assert len(lots) > 0, "Должны получить лоты после разделения бакета" # Проверяем что был probe с большим диапазоном И с sub-бакетами probed_ranges = [(min_p, max_p) for (_, page, min_p, max_p) in call_log if page == 1] # Должен быть probe [0, None] или [0, MAX] — начальный probe всего диапазона initial_probes = [r for r in probed_ranges if r[0] is None or r[0] == 0] assert ( len(initial_probes) >= 1 ), f"Ожидался probe всего диапазона, probed_ranges={probed_ranges}" @pytest.mark.asyncio async def test_fetch_all_secondary_paginates_when_under_cap(scraper: CianScraper) -> None: """totalOffers <= cap → пагинирует бакет полностью (страница за страницей).""" pages_fetched: list[int] = [] async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: pages_fetched.append(page) return f"page={page}" def fake_extract_total_offers(html: str) -> int | None: return 56 # 56 офферов → ceil(56/28) = 2 страницы def fake_parse_serp_html(html: 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(28)] return [] with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): lots = await scraper.fetch_all_secondary( rooms_buckets=[(2,)], price_cap_per_bucket=1400, ) # 28 лотов на стр 1 + 28 на стр 2 = 56, дедуп по source_id assert len(lots) == 56, f"Ожидалось 56 уникальных лотов, получено {len(lots)}" # Должно быть 2 страницы пагинации (page=1 probe + page=2) assert 2 in pages_fetched, "Должна быть запрошена страница 2" @pytest.mark.asyncio async def test_fetch_all_secondary_min_bracket_guard(scraper: CianScraper) -> None: """Если hi - lo < MIN_BRACKET → пагинируем как есть (не делим бесконечно).""" call_count = 0 async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: nonlocal call_count call_count += 1 if call_count > 100: raise RuntimeError("Бесконечная рекурсия — MIN_BRACKET guard не сработал") return f"page={page}" def fake_extract_total_offers(html: str) -> int | None: # Всегда > cap → без guard делилось бы бесконечно return 5000 def fake_parse_serp_html(html: str) -> list[ScrapedLot]: return [] # пустой → early stop в пагинации with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): # Запускаем с очень узким диапазоном — guard должен остановить деление await scraper._walk_price_range( rooms=(1,), lo=1_000_000, hi=1_000_000 + _MIN_BRACKET - 1, # bracket < MIN_BRACKET seen={}, price_cap_per_bucket=1400, max_pages_per_bucket=54, ) # Guard сработал: нет RuntimeError (бесконечная рекурсия не возникла). # Параллельная пагинация запускает до max_pages_per_bucket=54 страниц единовременно # (все страницы бакета параллельны), поэтому call_count может быть до 54. # Важно что нет рекурсии (bracket < MIN_BRACKET → один проход пагинации, не деление). assert call_count <= 55, f"Слишком много вызовов ({call_count}) — guard не сработал" @pytest.mark.asyncio async def test_fetch_all_secondary_on_bucket_called_per_leaf(scraper: CianScraper) -> None: """on_bucket вызывается после каждого leaf-бакета с лотами бакета.""" # totalOffers=56 ≤ cap → leaf-бакет, пагинируется параллельно 2 страницы bucket_calls: list[list[ScrapedLot]] = [] def fake_on_bucket(lots: list[ScrapedLot]) -> None: bucket_calls.append(list(lots)) async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: return f"page={page}" def fake_extract_total_offers(html: str) -> int | None: return 56 # ceil(56/28) = 2 страницы def fake_parse_serp_html(html: 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(10)] with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): await scraper.fetch_all_secondary( rooms_buckets=[(1,)], price_cap_per_bucket=1400, concurrency=3, on_bucket=fake_on_bucket, ) # on_bucket вызван ровно 1 раз (один leaf-бакет для одной комнатности) assert len(bucket_calls) == 1, f"Ожидался 1 вызов on_bucket, получено {len(bucket_calls)}" # Лоты из обеих страниц переданы в on_bucket assert ( len(bucket_calls[0]) == 20 ), f"Ожидалось 20 лотов в on_bucket (2 стр × 10), получено {len(bucket_calls[0])}" @pytest.mark.asyncio async def test_fetch_all_secondary_on_bucket_cancel_stops_run(scraper: CianScraper) -> None: """Если on_bucket кидает RuntimeError('cancelled') — прогон прерывается.""" def cancel_on_bucket(lots: list[ScrapedLot]) -> None: raise RuntimeError("cancelled") async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: return f"page={page}" def fake_extract_total_offers(html: str) -> int | None: return 28 # один лист → 1 страница def fake_parse_serp_html(html: str) -> list[ScrapedLot]: return [_make_lot("lot_1")] with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): with pytest.raises(RuntimeError, match="cancelled"): await scraper.fetch_all_secondary( rooms_buckets=[(1,), (2,)], price_cap_per_bucket=1400, on_bucket=cancel_on_bucket, ) @pytest.mark.asyncio async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: CianScraper) -> None: """Параллельная пагинация не ломает дедупликацию по source_id.""" fetch_pages: list[int] = [] async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None ) -> str: fetch_pages.append(page) return f"page={page}" def fake_extract_total_offers(html: str) -> int | None: return 84 # ceil(84/28) = 3 страницы def fake_parse_serp_html(html: str) -> list[ScrapedLot]: import re page_m = re.search(r"page=(\d+)", html) page = int(page_m.group(1)) if page_m else 1 # Страница 1 и 2 дают уникальные лоты; страница 3 дублирует страницу 1 if page == 3: return [_make_lot(f"lot_p1_{i}") for i in range(10)] # дубликаты return [_make_lot(f"lot_p{page}_{i}") for i in range(10)] with ( patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), patch.object(scraper, "_rotate_ip", return_value=False), ): lots = await scraper.fetch_all_secondary( rooms_buckets=[(1,)], price_cap_per_bucket=1400, concurrency=5, ) # Страницы 1+2 = 20 уникальных, страница 3 дублирует → итого 20 assert len(lots) == 20, f"Ожидалось 20 уникальных лотов (дедуп), получено {len(lots)}" # Все 3 страницы запрошены (параллельно или нет — нам важен факт) assert 3 in fetch_pages or 3 in [ p for p in fetch_pages ], f"Страница 3 не была запрошена, fetch_pages={fetch_pages}"