gendesign/tradein-mvp/backend/tests/test_domclick_sweep.py

"""Offline unit tests for DomClickCitySweepCounters + DomClick parser.

Legacy `run_domclick_city_sweep` orchestration behavior (sweep phases, cooperative
cancel, honest mark_done/mark_failed/mark_banned statuses) удалён вместе с
`app.services.scrape_pipeline` (#2397 Part E1) — эквивалентная regression-coverage
теперь в `test_scraper_kit_pipeline_parity2.py::test_domclick_city_sweep_*`
(kit `run_domclick_city_sweep`).

`app.services.scrapers.domclick` удалён (#2397 Part E2, 0 runtime-импортёров) —
`_map_item` ретаргетирован на kit `scraper_kit.providers.domclick.serp.DomClickScraper`
(единственный живой путь).

Остались:
  - DomClickCitySweepCounters (dataclass, из scraper_kit) — defaults + to_dict
  - DomClickScraper._map_item (kit) — BFF offer-item → ScrapedLot маппинг (не
    зависит от оркестрации)
"""

from __future__ import annotations

import os
from types import SimpleNamespace

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from scraper_kit.providers.domclick.serp import DomClickScraper

# ── DomClickCitySweepCounters ───────────────────────────────────────────────


def test_domclick_sweep_counters_defaults() -> None:
    from scraper_kit.orchestration.pipeline import DomClickCitySweepCounters

    c = DomClickCitySweepCounters()
    assert c.lots_fetched == 0
    assert c.lots_inserted == 0
    assert c.lots_updated == 0
    assert c.pages_fetched == 0
    assert c.errors_count == 0
    assert c.blocked == 0
    assert c.geo_filtered == 0


def test_domclick_sweep_counters_to_dict_all_keys() -> None:
    from dataclasses import fields

    from scraper_kit.orchestration.pipeline import DomClickCitySweepCounters

    c = DomClickCitySweepCounters()
    d = c.to_dict()
    expected = {f.name for f in fields(c)}
    assert set(d.keys()) == expected


# ── _map_item ─────────────────────────────────────────────────────────────────


def test_map_item_basic_mapping() -> None:
    """DomClickScraper._map_item маппит BFF offer-item → ScrapedLot корректно."""
    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    scraper = DomClickScraper(config, delay_provider=lambda _name: 8.0)
    item = {
        "id": 111222333,
        "path": "https://domclick.ru/card/sale__flat__111222333",
        "location": {"lat": 56.838, "lon": 60.612},
        "address": {"displayName": "Екатеринбург, улица Ленина, 10"},
        "objectInfo": {"area": 52.3, "rooms": 2, "floor": 5},
        "house": {"floors": 16, "buildYear": 2010},
        "price": 6_500_000,
        "squarePrice": 124282,
        "flatComplex": None,
        "isRosreestrApproved": False,
        "publishedDate": "2026-06-01T10:00:00+05:00",
        "updatedDate": None,
        "lastPriceHistoryState": None,
        "offerRegionName": "Екатеринбург",
    }
    lot = scraper._map_item(item)
    assert lot is not None
    assert lot.source == "domklik"
    assert lot.source_id == "111222333"
    assert lot.rooms == 2
    assert lot.area_m2 == pytest.approx(52.3)
    assert lot.floor == 5
    assert lot.total_floors == 16
    assert lot.price_rub == 6_500_000
    assert lot.listing_segment == "vtorichka"
    assert lot.lat == pytest.approx(56.838)
    assert lot.lon == pytest.approx(60.612)


def test_map_item_puts_complex_slug_into_newbuilding_id() -> None:
    """flatComplex.slug доезжает до ScrapedLot.newbuilding_id (#3064).

    Слаг, а не id: докстринг ScrapedLot задаёт формат «federatsiya-ekaterinburg»,
    и avito — единственный источник, который это поле заполняет, кладёт туда слаг.

    ЭТО НЕ привязка к дому. house_source/house_ext_id намеренно не трогаются:
    все активные domklik-листинги — вторичка, а flatComplex — идентификатор ЖК,
    а не здания (116 значений на 737 листингов по проду). Ключ здания
    (domclick_building_guid) приходит detail-путём — отдельная развилка в #3064.
    """
    import json
    from pathlib import Path

    fixture_path = Path(__file__).parent / "fixtures" / "domclick_bff_offers_sample.json"
    items = json.loads(fixture_path.read_text(encoding="utf-8"))["result"]["items"]

    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    scraper = DomClickScraper(config, delay_provider=lambda _name: 8.0)

    # item 1001: flatComplex.slug='malysheva' -> слаг доезжает до колонки.
    lot_with_complex = scraper._map_item(items[0])
    assert lot_with_complex is not None
    assert lot_with_complex.newbuilding_id == "malysheva"
    # Привязка к дому НЕ появилась — это стережёт границу правки.
    assert lot_with_complex.house_source is None
    assert lot_with_complex.house_ext_id is None

    # item 1002: flatComplex отсутствует -> None, а не пустая строка: пустая
    # строка попала бы в колонку как значение и читалась бы как «ЖК есть, но без
    # имени», что неотличимо от настоящего ЖК с пустым слагом.
    lot_no_complex = scraper._map_item(items[1])
    assert lot_no_complex is not None
    assert lot_no_complex.newbuilding_id is None


def test_map_item_promotes_rosreestr_from_bff_fixture() -> None:
    """isRosreestrApproved из BFF-фикстуры доезжает до
    ScrapedLot.is_rosreestr_checked (#3064 п.1).

    На проде у всех 1338 активных domklik-листингов колонка была NULL, хотя поле
    уже парсилось и оседало в raw_payload. Соответствие 1:1 с cian
    (`offer.get("isRosreestrChecked")`, cian/serp.py:1014) — без трансформаций.

    is_pro_seller здесь НЕ проверяется намеренно: колонка уже несёт два разных
    смысла (cian — платная PRO-подписка, `019_listings_alter_cian.sql:43`;
    yandex — категория продавца AGENCY/AGENT, `yandex/serp.py:203-220`), и
    выводить её из наличия agency_name значило бы завести третий. См. #3064.
    """
    import json
    from pathlib import Path

    fixture_path = Path(__file__).parent / "fixtures" / "domclick_bff_offers_sample.json"
    items = json.loads(fixture_path.read_text(encoding="utf-8"))["result"]["items"]

    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    scraper = DomClickScraper(config, delay_provider=lambda _name: 8.0)

    # item 1001: isRosreestrApproved=true -> ЕГРН-сверка пройдена.
    lot_checked = scraper._map_item(items[0])
    assert lot_checked is not None
    assert lot_checked.is_rosreestr_checked is True

    # item 1003: ключа isRosreestrApproved в payload нет вовсе -> None, а НЕ
    # False. Отсутствие данных не то же самое, что "проверено — не сходится":
    # False поехал бы в оценщик как утверждение о квартире.
    lot_unknown = scraper._map_item(items[2])
    assert lot_unknown is not None
    assert lot_unknown.is_rosreestr_checked is None


# ── fetch_city: report_ban на QRATOR-блок (#2600 п.1) ───────────────────────────


class _FakeFetcher:
    """Заглушка BrowserFetcher: async context manager + report_ban recorder."""

    def __init__(self) -> None:
        self.banned: list[str] = []

    async def __aenter__(self) -> _FakeFetcher:
        return self

    async def __aexit__(self, *args: object) -> None:
        return None

    def report_ban(self, reason: str) -> None:
        self.banned.append(reason)


async def test_fetch_city_reports_ban_on_qrator_block(monkeypatch: pytest.MonkeyPatch) -> None:
    """QRATOR-блок в fetch_city → fetcher.report_ban вызывается ВНУТРИ `async with
    build_browser_fetcher(...) as fetcher:` (lease ещё держится) — #2600 п.1.

    no-op сегодня (domclick SERP собирается БЕЗ proxy_provider — #2160 P4 wiring
    gap), но сам вызов должен произойти корректно на нужном (живом) fetcher'е.
    """
    from scraper_kit.domclick_exceptions import DomClickBlockedError

    fake_fetcher = _FakeFetcher()

    def _fake_build_browser_fetcher(config: object, source: str, **_kw: object) -> _FakeFetcher:
        assert source == "domclick"
        return fake_fetcher

    monkeypatch.setattr(
        "scraper_kit.providers._base.build_browser_fetcher", _fake_build_browser_fetcher
    )

    async def _raise_blocked(self: DomClickScraper, **_: object) -> None:
        raise DomClickBlockedError("QRATOR block page")

    monkeypatch.setattr(DomClickScraper, "_sweep_bucket", _raise_blocked)

    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    scraper = DomClickScraper(config)

    lots = await scraper.fetch_city(city_id=1)

    assert lots == []
    assert scraper.blocked is True
    assert fake_fetcher.banned  # report_ban был вызван на ЖИВОМ fetcher'е
    assert "QRATOR" in fake_fetcher.banned[0]


# ── #2854: точка входа в ROOM_BUCKETS сдвигается от прогона к прогону ───────────
#
# Блок QRATOR прилетает ВНУТРИ первой корзины — на проде buckets_completed=0 во всех
# прогонах, включая тот, где собрано 389 лотов. Обход при этом всегда стартовал со
# студий, поэтому комнатности 2+ не собирались никогда. Замер 19.08 по активным
# объявлениям: у Домклика 575 студий, 408 однушек, 1 двушка, 0 трёшек; у Циана в тех
# же категориях 6 461 / 7 352 / 4 581.


class _OrderRecordingScraper:
    """Ничего не делает, только запоминает порядок корзин."""


async def _make_scraper_recording(monkeypatch: pytest.MonkeyPatch) -> tuple[DomClickScraper, list]:
    seen: list[str] = []

    def _fake_build_browser_fetcher(config: object, source: str, **_kw: object) -> _FakeFetcher:
        return _FakeFetcher()

    monkeypatch.setattr(
        "scraper_kit.providers._base.build_browser_fetcher", _fake_build_browser_fetcher
    )

    async def _record(self: DomClickScraper, *, rooms: str, **_: object) -> None:
        seen.append(rooms)

    monkeypatch.setattr(DomClickScraper, "_sweep_bucket", _record)
    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    return DomClickScraper(config), seen


async def test_fetch_city_rotates_bucket_order(monkeypatch: pytest.MonkeyPatch) -> None:
    """Сдвиг меняет ПОРЯДОК корзин, а не их состав — ни одна не теряется."""
    from scraper_kit.providers.domclick.serp import ROOM_BUCKETS

    scraper, seen = await _make_scraper_recording(monkeypatch)
    await scraper.fetch_city(city_id=1, start_bucket_index=2)

    assert seen == list(ROOM_BUCKETS[2:]) + list(ROOM_BUCKETS[:2])
    assert sorted(seen) == sorted(ROOM_BUCKETS), "состав корзин изменился — потеряли охват"


async def test_fetch_city_start_index_is_normalized(monkeypatch: pytest.MonkeyPatch) -> None:
    """Вызывающий передаёт остаток от run_id — он ничем не ограничен сверху."""
    from scraper_kit.providers.domclick.serp import ROOM_BUCKETS

    scraper, seen = await _make_scraper_recording(monkeypatch)
    await scraper.fetch_city(city_id=1, start_bucket_index=len(ROOM_BUCKETS) + 1)

    assert seen[0] == ROOM_BUCKETS[1]
    assert scraper.bucket_start_index == 1


async def test_blocked_run_collects_shifted_bucket_not_studios(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Главный сценарий: блок в ПЕРВОЙ же корзине — собранной оказывается сдвинутая.

    Ровно так ведёт себя прод: одна корзина за прогон и обрыв. До #2854 этой
    единственной корзиной всегда были студии, из-за чего двушки и трёшки не
    появлялись в listings вовсе.
    """
    from scraper_kit.domclick_exceptions import DomClickBlockedError
    from scraper_kit.providers.domclick.serp import ROOM_BUCKETS

    attempted: list[str] = []

    def _fake_build_browser_fetcher(config: object, source: str, **_kw: object) -> _FakeFetcher:
        return _FakeFetcher()

    monkeypatch.setattr(
        "scraper_kit.providers._base.build_browser_fetcher", _fake_build_browser_fetcher
    )

    async def _block(self: DomClickScraper, *, rooms: str, **_: object) -> None:
        attempted.append(rooms)
        raise DomClickBlockedError("QRATOR block page")

    monkeypatch.setattr(DomClickScraper, "_sweep_bucket", _block)
    config = SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
    scraper = DomClickScraper(config)

    await scraper.fetch_city(city_id=1, start_bucket_index=ROOM_BUCKETS.index("2"))

    assert attempted == ["2"], "обход всё ещё начинается не со сдвинутой корзины"
    assert scraper.blocked is True
    assert scraper.bucket_start_index == ROOM_BUCKETS.index("2")


async def test_default_start_index_keeps_previous_behaviour(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Контроль: без явного сдвига порядок прежний — правка не меняет вызовы без него."""
    from scraper_kit.providers.domclick.serp import ROOM_BUCKETS

    scraper, seen = await _make_scraper_recording(monkeypatch)
    await scraper.fetch_city(city_id=1)

    assert seen == list(ROOM_BUCKETS)


# ── проводка: pipeline обязан ПЕРЕДАВАТЬ сдвиг, а не просто уметь его принимать ──


async def test_pipeline_passes_rotating_start_index() -> None:
    """Без этого теста передачу сдвига можно снять, и проверки выше останутся зелёными:
    скрейпер по-прежнему умеет сдвигать, просто его об этом никто не просит.
    """
    from unittest.mock import AsyncMock, MagicMock, patch

    from scraper_kit.orchestration.pipeline import run_domclick_city_sweep
    from scraper_kit.providers.domclick.serp import ROOM_BUCKETS

    pfx = "scraper_kit.orchestration.pipeline"
    run_id = 17
    scraper = MagicMock()
    scraper.__aenter__ = AsyncMock(return_value=scraper)
    scraper.__aexit__ = AsyncMock(return_value=None)
    scraper.fetch_city = AsyncMock(return_value=[])
    scraper.blocked = False
    scraper.geo_filtered = 0
    scraper.fetch_errors = 0
    scraper.buckets_completed = 0
    scraper.buckets_total = len(ROOM_BUCKETS)
    scraper.bucket_start_index = run_id % len(ROOM_BUCKETS)

    runs_stub = MagicMock()
    runs_stub.is_cancelled.return_value = False

    with (
        patch(f"{pfx}.DomClickScraper", return_value=scraper),
        patch(f"{pfx}.save_listings", MagicMock(return_value=(0, 0))),
        patch(f"{pfx}.runs", runs_stub),
    ):
        await run_domclick_city_sweep(
            MagicMock(),
            config=SimpleNamespace(browser_http_endpoint="http://x:9000"),
            matcher=MagicMock(),
            run_id=run_id,
            city_id=4,
            pages=1,
            request_delay_sec=0.0,
        )

    passed = scraper.fetch_city.await_args.kwargs.get("start_bucket_index")
    assert passed == run_id % len(ROOM_BUCKETS), (
        f"pipeline не передал сдвиг корзины (получено {passed!r}) — "
        "обход снова всегда начнётся со студий"
    )