481 lines
19 KiB
Python
481 lines
19 KiB
Python
"""Golden-parity (F2): kit sweep-оркестраторы ≡ `app.services.scrape_pipeline`.
|
||
|
||
Продолжение test_scraper_kit_pipeline_parity.py (F1 покрыл avito city sweep). Здесь —
|
||
остальные 7 sweep'ов, скопированных в `scraper_kit.orchestration.pipeline` (#2135 F2):
|
||
|
||
City sweep'ы (приоритет — активны в проде):
|
||
- run_yandex_city_sweep — combos SERP + save + price-history
|
||
- run_cian_city_sweep — SERP + newbuilding_only-фильтр + save
|
||
- run_domclick_city_sweep — BFF citywide + честный статус (done / failed)
|
||
Плюс:
|
||
- run_avito_newbuilding_sweep — citywide novostroyka SERP + save
|
||
Full load'ы (smoke-parity — импорт + базовый прогон через on_bucket):
|
||
- run_avito_full_load / run_cian_full_load / run_yandex_full_load
|
||
|
||
Метод — как в F1: гоняем ОБА orchestrator'а на идентичном мокнутом I/O, сравниваем
|
||
итоговый counters.to_dict() + последовательность (method, numeric-counters) вызовов
|
||
scrape_runs. Развязка `app.*` (config/matcher/enrichment/runs/shutdown инжектируются)
|
||
не должна менять ветвление/счётчики. Без сети, без БД.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_avito_full_load as new_avito_full_load,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_avito_newbuilding_sweep as new_nb_sweep,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_cian_city_sweep as new_cian_city_sweep,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_cian_full_load as new_cian_full_load,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_domclick_city_sweep as new_domclick_city_sweep,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_yandex_city_sweep as new_yandex_city_sweep,
|
||
)
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_yandex_full_load as new_yandex_full_load,
|
||
)
|
||
|
||
from app.services.scrape_pipeline import (
|
||
run_avito_full_load as old_avito_full_load,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_avito_newbuilding_sweep as old_nb_sweep,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_cian_city_sweep as old_cian_city_sweep,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_cian_full_load as old_cian_full_load,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_domclick_city_sweep as old_domclick_city_sweep,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_yandex_city_sweep as old_yandex_city_sweep,
|
||
)
|
||
from app.services.scrape_pipeline import (
|
||
run_yandex_full_load as old_yandex_full_load,
|
||
)
|
||
|
||
OLD_PFX = "app.services.scrape_pipeline"
|
||
NEW_PFX = "scraper_kit.orchestration.pipeline"
|
||
|
||
# ── recording scrape_runs ────────────────────────────────────────────────────
|
||
|
||
_NON_NUMERIC_KEYS = {"enrichment_abort_note", "done_buckets"}
|
||
|
||
|
||
class _RunsRecorder:
|
||
"""Записывает вызовы scrape_runs-финализаторов. is_cancelled всегда False."""
|
||
|
||
def __init__(self) -> None:
|
||
self.calls: list[tuple[str, dict[str, Any]]] = []
|
||
|
||
def is_cancelled(self, db: Any, run_id: int) -> bool:
|
||
return False
|
||
|
||
def update_heartbeat(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
|
||
self.calls.append(("update_heartbeat", dict(counters)))
|
||
|
||
def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None:
|
||
self.calls.append(("mark_done", dict(counters)))
|
||
|
||
def mark_banned(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None:
|
||
self.calls.append(("mark_banned", dict(counters)))
|
||
|
||
def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None:
|
||
self.calls.append(("mark_failed", dict(counters)))
|
||
|
||
|
||
def _normalize(calls: list[tuple[str, dict[str, Any]]]) -> list[tuple[str, dict[str, int]]]:
|
||
"""Оставить имя метода + числовые counters (убрать note/done_buckets)."""
|
||
out: list[tuple[str, dict[str, int]]] = []
|
||
for method, counters in calls:
|
||
numeric = {k: v for k, v in counters.items() if k not in _NON_NUMERIC_KEYS}
|
||
out.append((method, numeric))
|
||
return out
|
||
|
||
|
||
def _config() -> SimpleNamespace:
|
||
"""Инжектируемый config (kit) + тот же объект как patched settings (old)."""
|
||
return SimpleNamespace(
|
||
scraper_fetch_mode="curl_cffi",
|
||
browser_http_endpoint="http://browser.test/fetch",
|
||
scraper_proxy_url=None,
|
||
avito_proxy_rotate_url=None,
|
||
avito_proxy_max_rotations=0,
|
||
avito_serp_ok_not_banned=True,
|
||
avito_proxy_rotate_settle_s=0.0,
|
||
proxy_rotate_attempts=1,
|
||
proxy_rotate_attempt_timeout_s=1.0,
|
||
cian_proxy_rotate_url=None,
|
||
cian_proxy_max_rotations=0,
|
||
yandex_proxy_rotate_url=None,
|
||
yandex_proxy_max_rotations=0,
|
||
scraper_skip_seen_today=False,
|
||
cian_full_load_per_fetch_timeout_s=0.0,
|
||
)
|
||
|
||
|
||
def _ctx_scraper(**attrs: Any) -> MagicMock:
|
||
"""MagicMock, пригодный и как `Cls()`, и как `async with Cls() as x`.
|
||
|
||
__aenter__ возвращает сам объект, __aexit__ — no-op. Доп. атрибуты/методы
|
||
задаются через kwargs (async-методы — AsyncMock).
|
||
"""
|
||
m = MagicMock()
|
||
m.__aenter__ = AsyncMock(return_value=m)
|
||
m.__aexit__ = AsyncMock(return_value=None)
|
||
for k, v in attrs.items():
|
||
setattr(m, k, v)
|
||
return m
|
||
|
||
|
||
def _async_session_cm() -> MagicMock:
|
||
sess = MagicMock()
|
||
sess.__aenter__ = AsyncMock(return_value=sess)
|
||
sess.__aexit__ = AsyncMock(return_value=None)
|
||
sess.close = AsyncMock()
|
||
return sess
|
||
|
||
|
||
_DriveResult = tuple[dict[str, int], list[tuple[str, dict[str, int]]]]
|
||
|
||
|
||
async def _assert_parity(old_res: _DriveResult, new_res: _DriveResult) -> dict[str, int]:
|
||
old_counters, old_calls = old_res
|
||
new_counters, new_calls = new_res
|
||
assert (
|
||
new_counters == old_counters
|
||
), f"counters diverged:\n old={old_counters}\n new={new_counters}"
|
||
assert (
|
||
new_calls == old_calls
|
||
), f"scrape_runs calls diverged:\n old={old_calls}\n new={new_calls}"
|
||
return new_counters
|
||
|
||
|
||
# ── Yandex city sweep ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def _yandex_scraper(combos: list[tuple[str, list[Any]]]) -> MagicMock:
|
||
"""Fake YandexRealtyScraper: fetch_around_multi_room вызывает on_combo по combos."""
|
||
|
||
async def _fetch(*_a: Any, on_combo: Any = None, **_k: Any) -> None:
|
||
for label, lots in combos:
|
||
on_combo(label, lots)
|
||
|
||
return _ctx_scraper(fetch_around_multi_room=_fetch)
|
||
|
||
|
||
async def _drive_yandex_city(*, new: bool) -> _DriveResult:
|
||
recorder = _RunsRecorder()
|
||
db = MagicMock()
|
||
combos = [
|
||
("2к-price0", [MagicMock(source_id="y1"), MagicMock(source_id="y2")]),
|
||
("1к-price0", [MagicMock(source_id="y3")]),
|
||
]
|
||
scraper = _yandex_scraper(combos)
|
||
save_mock = MagicMock(side_effect=[(2, 0), (1, 0)])
|
||
cfg = _config()
|
||
kwargs = dict(
|
||
run_id=1,
|
||
anchors=None,
|
||
pages_per_anchor=1,
|
||
request_delay_sec=0.0,
|
||
enrich_address=False,
|
||
)
|
||
if new:
|
||
enrichment = MagicMock()
|
||
enrichment.record_yandex_price_history = MagicMock(return_value=5)
|
||
with (
|
||
patch(f"{NEW_PFX}.YandexRealtyScraper", return_value=scraper),
|
||
patch(f"{NEW_PFX}.save_listings", save_mock),
|
||
patch(f"{NEW_PFX}.runs", recorder),
|
||
):
|
||
counters = await new_yandex_city_sweep(
|
||
db, config=cfg, matcher=MagicMock(), enrichment=enrichment, **kwargs
|
||
)
|
||
else:
|
||
with (
|
||
patch(f"{OLD_PFX}.YandexRealtyScraper", return_value=scraper),
|
||
patch(f"{OLD_PFX}.save_listings", save_mock),
|
||
patch(f"{OLD_PFX}.record_yandex_price_history", MagicMock(return_value=5)),
|
||
patch(f"{OLD_PFX}.scrape_runs", recorder),
|
||
patch(f"{OLD_PFX}.settings", cfg),
|
||
patch(f"{OLD_PFX}.shutdown_requested", return_value=False),
|
||
):
|
||
counters = await old_yandex_city_sweep(db, **kwargs)
|
||
return counters.to_dict(), _normalize(recorder.calls)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_yandex_city_sweep_parity() -> None:
|
||
"""1 центральный anchor, 2 combos, save + price-history, enrich_address=False."""
|
||
counters = await _assert_parity(
|
||
await _drive_yandex_city(new=False), await _drive_yandex_city(new=True)
|
||
)
|
||
assert counters["lots_fetched"] == 3
|
||
assert counters["lots_inserted"] == 3
|
||
assert counters["price_history_rows"] == 5
|
||
assert counters["anchors_done"] == 1
|
||
|
||
|
||
# ── Cian city sweep ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _cian_lot(segment: str) -> MagicMock:
|
||
return MagicMock(listing_segment=segment, house_source=None, house_ext_id=None)
|
||
|
||
|
||
async def _drive_cian_city(*, new: bool) -> _DriveResult:
|
||
recorder = _RunsRecorder()
|
||
db = MagicMock()
|
||
# 3 novostroyki + 2 secondary → newbuilding_only оставит 3.
|
||
lots = [_cian_lot("novostroyki")] * 3 + [_cian_lot("vtorichnaya")] * 2
|
||
scraper = _ctx_scraper(fetch_around_multi_room=AsyncMock(return_value=lots))
|
||
save_mock = MagicMock(side_effect=[(3, 0)])
|
||
cfg = _config()
|
||
kwargs = dict(
|
||
run_id=1,
|
||
anchors=[(56.84, 60.60, "A1")],
|
||
radius_m=1000,
|
||
pages_per_anchor=1,
|
||
request_delay_sec=0.0,
|
||
detail_top_n=0,
|
||
enrich_houses=False,
|
||
newbuilding_only=True,
|
||
)
|
||
if new:
|
||
with (
|
||
patch(f"{NEW_PFX}.CianScraper", return_value=scraper),
|
||
patch(f"{NEW_PFX}.save_listings", save_mock),
|
||
patch(f"{NEW_PFX}.runs", recorder),
|
||
):
|
||
counters = await new_cian_city_sweep(db, config=cfg, matcher=MagicMock(), **kwargs)
|
||
else:
|
||
with (
|
||
patch("app.services.scrapers.cian.CianScraper", return_value=scraper),
|
||
patch(f"{OLD_PFX}.save_listings", save_mock),
|
||
patch(f"{OLD_PFX}.scrape_runs", recorder),
|
||
patch(f"{OLD_PFX}.settings", cfg),
|
||
patch(f"{OLD_PFX}.shutdown_requested", return_value=False),
|
||
):
|
||
counters = await old_cian_city_sweep(db, **kwargs)
|
||
return counters.to_dict(), _normalize(recorder.calls)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cian_city_sweep_parity() -> None:
|
||
"""1 anchor, newbuilding_only отбрасывает вторичку, save новостроек, mark_done."""
|
||
counters = await _assert_parity(
|
||
await _drive_cian_city(new=False), await _drive_cian_city(new=True)
|
||
)
|
||
assert counters["lots_fetched"] == 5
|
||
assert counters["lots_dropped_secondary"] == 2
|
||
assert counters["lots_inserted"] == 3
|
||
assert counters["anchors_done"] == 1
|
||
|
||
|
||
# ── DomClick city sweep ───────────────────────────────────────────────────────
|
||
|
||
|
||
async def _drive_domclick(*, new: bool, lots_n: int, blocked: bool) -> _DriveResult:
|
||
recorder = _RunsRecorder()
|
||
db = MagicMock()
|
||
lots = [MagicMock() for _ in range(lots_n)]
|
||
scraper = _ctx_scraper(
|
||
fetch_city=AsyncMock(return_value=lots),
|
||
blocked=blocked,
|
||
geo_filtered=0,
|
||
fetch_errors=0,
|
||
)
|
||
save_mock = MagicMock(side_effect=[(lots_n, 0)] if lots_n else [])
|
||
cfg = _config()
|
||
kwargs = dict(run_id=1, city_id=4, pages=1, request_delay_sec=0.0)
|
||
if new:
|
||
with (
|
||
patch(f"{NEW_PFX}.DomClickScraper", return_value=scraper),
|
||
patch(f"{NEW_PFX}.save_listings", save_mock),
|
||
patch(f"{NEW_PFX}.runs", recorder),
|
||
):
|
||
counters = await new_domclick_city_sweep(db, config=cfg, matcher=MagicMock(), **kwargs)
|
||
else:
|
||
with (
|
||
patch("app.services.scrapers.domclick.DomClickScraper", return_value=scraper),
|
||
patch(f"{OLD_PFX}.save_listings", save_mock),
|
||
patch(f"{OLD_PFX}.scrape_runs", recorder),
|
||
patch(f"{OLD_PFX}.settings", cfg),
|
||
patch(f"{OLD_PFX}.shutdown_requested", return_value=False),
|
||
):
|
||
counters = await old_domclick_city_sweep(db, **kwargs)
|
||
return counters.to_dict(), _normalize(recorder.calls)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_domclick_city_sweep_parity_done() -> None:
|
||
"""Citywide fetch_city → 4 lots saved → mark_done."""
|
||
counters = await _assert_parity(
|
||
await _drive_domclick(new=False, lots_n=4, blocked=False),
|
||
await _drive_domclick(new=True, lots_n=4, blocked=False),
|
||
)
|
||
assert counters["lots_fetched"] == 4
|
||
assert counters["lots_inserted"] == 4
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_domclick_city_sweep_parity_blocked_failed() -> None:
|
||
"""QRATOR-блок + 0 lots → mark_failed (честный статус #1968)."""
|
||
counters = await _assert_parity(
|
||
await _drive_domclick(new=False, lots_n=0, blocked=True),
|
||
await _drive_domclick(new=True, lots_n=0, blocked=True),
|
||
)
|
||
assert counters["lots_fetched"] == 0
|
||
assert counters["blocked"] == 1
|
||
|
||
|
||
# ── Avito newbuilding sweep ───────────────────────────────────────────────────
|
||
|
||
|
||
async def _drive_nb_sweep(*, new: bool) -> _DriveResult:
|
||
recorder = _RunsRecorder()
|
||
db = MagicMock()
|
||
lots = [MagicMock() for _ in range(6)]
|
||
scraper = MagicMock()
|
||
scraper._cffi = None
|
||
scraper._browser = None
|
||
scraper.fetch_newbuildings = AsyncMock(return_value=lots)
|
||
save_mock = MagicMock(side_effect=[(5, 1)])
|
||
cfg = _config()
|
||
kwargs = dict(run_id=1, pages=2, request_delay_sec=0.0)
|
||
if new:
|
||
with (
|
||
patch(f"{NEW_PFX}.AvitoScraper", return_value=scraper),
|
||
patch(f"{NEW_PFX}.save_listings", save_mock),
|
||
patch(f"{NEW_PFX}.runs", recorder),
|
||
patch(f"{NEW_PFX}.AsyncSession", return_value=_async_session_cm()),
|
||
):
|
||
counters = await new_nb_sweep(db, config=cfg, matcher=MagicMock(), **kwargs)
|
||
else:
|
||
with (
|
||
patch(f"{OLD_PFX}.AvitoScraper", return_value=scraper),
|
||
patch(f"{OLD_PFX}.save_listings", save_mock),
|
||
patch(f"{OLD_PFX}.scrape_runs", recorder),
|
||
patch(f"{OLD_PFX}.settings", cfg),
|
||
patch(f"{OLD_PFX}.shutdown_requested", return_value=False),
|
||
patch(f"{OLD_PFX}.AsyncSession", return_value=_async_session_cm()),
|
||
):
|
||
counters = await old_nb_sweep(db, **kwargs)
|
||
return counters.to_dict(), _normalize(recorder.calls)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_avito_newbuilding_sweep_parity() -> None:
|
||
"""Citywide novostroyka SERP → save → heartbeat → mark_done."""
|
||
counters = await _assert_parity(
|
||
await _drive_nb_sweep(new=False), await _drive_nb_sweep(new=True)
|
||
)
|
||
assert counters["lots_fetched"] == 6
|
||
assert counters["lots_inserted"] == 5
|
||
assert counters["lots_updated"] == 1
|
||
|
||
|
||
# ── Full loads (smoke-parity через on_bucket) ─────────────────────────────────
|
||
|
||
|
||
def _full_load_scraper(buckets: list[tuple[str, list[Any]]]) -> MagicMock:
|
||
"""Fake scraper: fetch_all_secondary вызывает on_bucket по каждому бакету."""
|
||
|
||
async def _fetch(*_a: Any, on_bucket: Any = None, on_progress: Any = None, **_k: Any) -> None:
|
||
for key, lots in buckets:
|
||
on_bucket(key, lots)
|
||
|
||
scraper = _ctx_scraper(fetch_all_secondary=_fetch)
|
||
scraper._browser = None
|
||
return scraper
|
||
|
||
|
||
async def _drive_full_load(*, new: bool, source: str) -> _DriveResult:
|
||
recorder = _RunsRecorder()
|
||
db = MagicMock()
|
||
buckets = [
|
||
("2к:0-5m", [MagicMock(source_id=f"{source}1"), MagicMock(source_id=f"{source}2")]),
|
||
("1к:0-5m", [MagicMock(source_id=f"{source}3")]),
|
||
]
|
||
scraper = _full_load_scraper(buckets)
|
||
save_mock = MagicMock(side_effect=[(2, 0), (1, 0)])
|
||
cfg = _config()
|
||
|
||
old_map = {
|
||
"avito": (old_avito_full_load, new_avito_full_load, f"{OLD_PFX}.AvitoScraper"),
|
||
"cian": (old_cian_full_load, new_cian_full_load, "app.services.scrapers.cian.CianScraper"),
|
||
"yandex": (
|
||
old_yandex_full_load,
|
||
new_yandex_full_load,
|
||
"app.services.scrapers.yandex_realty.YandexRealtyScraper",
|
||
),
|
||
}
|
||
old_fn, new_fn, old_scraper_target = old_map[source]
|
||
|
||
if new:
|
||
new_scraper_target = {
|
||
"avito": f"{NEW_PFX}.AvitoScraper",
|
||
"cian": f"{NEW_PFX}.CianScraper",
|
||
"yandex": f"{NEW_PFX}.YandexRealtyScraper",
|
||
}[source]
|
||
extra: dict[str, Any] = {}
|
||
if source == "yandex":
|
||
enrichment = MagicMock()
|
||
enrichment.record_yandex_price_history = MagicMock(return_value=0)
|
||
extra["enrichment"] = enrichment
|
||
with (
|
||
patch(new_scraper_target, return_value=scraper),
|
||
patch(f"{NEW_PFX}.save_listings", save_mock),
|
||
patch(f"{NEW_PFX}.runs", recorder),
|
||
):
|
||
counters = await new_fn(db, run_id=1, config=cfg, matcher=MagicMock(), **extra)
|
||
else:
|
||
import contextlib
|
||
|
||
ctx = [
|
||
patch(old_scraper_target, return_value=scraper),
|
||
patch(f"{OLD_PFX}.save_listings", save_mock),
|
||
patch(f"{OLD_PFX}.scrape_runs", recorder),
|
||
patch(f"{OLD_PFX}.settings", cfg),
|
||
patch(f"{OLD_PFX}.shutdown_requested", return_value=False),
|
||
]
|
||
if source == "yandex":
|
||
ctx.append(patch(f"{OLD_PFX}.record_yandex_price_history", MagicMock(return_value=0)))
|
||
with contextlib.ExitStack() as stack:
|
||
for cm in ctx:
|
||
stack.enter_context(cm)
|
||
counters = await old_fn(db, run_id=1)
|
||
return counters.to_dict(), _normalize(recorder.calls)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("source", ["avito", "cian", "yandex"])
|
||
async def test_full_load_smoke_parity(source: str) -> None:
|
||
"""Full load: 2 бакета через on_bucket → save + heartbeat → mark_done. Counters ≡."""
|
||
counters = await _assert_parity(
|
||
await _drive_full_load(new=False, source=source),
|
||
await _drive_full_load(new=True, source=source),
|
||
)
|
||
assert counters["unique_fetched"] == 3
|
||
assert counters["saved_inserted"] == 3
|
||
assert counters["saved_updated"] == 0
|