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

"""Offline unit tests for run_cian_city_sweep (#860).

Pure & fast: NO live network, NO DB. Mirrors test_city_sweep.
Asserts orchestration semantics only.

Coverage:
  - Все фазы (SERP → save → detail → houses) вызываются в нужном порядке
  - CianCitySweepCounters корректно заполняются
  - Cooperative cancel прерывает sweep перед anchor'ом
  - enrich_houses=False пропускает houses-фазу
  - Consecutive SERP failures → mark_banned + abort
  - Per-item detail/house error не валит весь sweep
  - mark_done вызывается при нормальном завершении
"""

from __future__ import annotations

import os

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

from typing import Any
from unittest.mock import AsyncMock, MagicMock

import pytest

from app.services import scrape_pipeline
from app.services.scrapers.base import ScrapedLot
from app.services.scrapers.cian import CianScraper

# Три тестовых anchor'а — не все 5, чтобы тесты были быстрее
TEST_ANCHORS: list[tuple[float, float, str]] = [
    (56.8400, 60.6050, "A1"),
    (56.7950, 60.5300, "A2"),
    (56.8970, 60.6100, "A3"),
]


def _fake_lot(
    offer_id: str, *, nb: bool = False, segment: str | None = "novostroyki"
) -> ScrapedLot:
    """Создаём ScrapedLot с опциональным newbuilding-ссылкой.

    segment по умолчанию "novostroyki" — run_cian_city_sweep теперь NB-only
    (newbuilding_only=True), поэтому дефолтные lots должны проходить SERP-фильтр.
    """
    return ScrapedLot(
        source="cian",
        source_url=f"https://ekb.cian.ru/sale/flat/{offer_id}/",
        source_id=offer_id,
        address="Екатеринбург, улица Тестовая, 1",
        price_rub=5_000_000,
        area_m2=50.0,
        rooms=2,
        listing_segment=segment,
        house_source="cian_newbuilding" if nb else "cian",
        house_ext_id="99999" if nb else None,
    )


class _FakeRuns:
    """Stub для scrape_runs: отслеживает вызовы, поддерживает cooperative cancel."""

    def __init__(self, *, cancel_at_anchor: int | None = None) -> None:
        self.cancel_at_anchor = cancel_at_anchor
        self._is_cancelled_calls = 0
        self.heartbeats: list[dict[str, int]] = []
        self.done: dict[str, int] | None = None
        self.failed: tuple[str, dict[str, int]] | None = None
        self.banned: tuple[str, dict[str, int]] | None = None

    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
        self._is_cancelled_calls += 1
        if self.cancel_at_anchor is None:
            return False
        return self._is_cancelled_calls >= self.cancel_at_anchor

    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
        self.heartbeats.append(dict(counters))

    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
        self.done = dict(counters)

    def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
        self.failed = (error, dict(counters))

    def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
        self.banned = (error, dict(counters))


@pytest.fixture(autouse=True)
def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
    """asyncio.sleep → instant (no test delays)."""

    async def _instant(_secs: float) -> None:
        return None

    monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant)


@pytest.fixture(autouse=True)
def _stub_cian_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
    """Нейтрализуем CianScraper.__init__/__aenter__/__aexit__ — нет DB вызовов."""

    def _init(self: CianScraper) -> None:
        self.request_delay_sec = 0.0
        self._cffi = None

    async def _aenter(self: CianScraper) -> CianScraper:
        return self

    async def _aexit(self: CianScraper, *_args: Any) -> None:
        return None

    monkeypatch.setattr(CianScraper, "__init__", _init)
    monkeypatch.setattr(CianScraper, "__aenter__", _aenter)
    monkeypatch.setattr(CianScraper, "__aexit__", _aexit)


def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
    monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled)
    monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat)
    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done)
    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed)
    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned)


# ── CianCitySweepCounters ───────────────────────────────────────────────────


def test_cian_sweep_counters_defaults() -> None:
    from app.services.scrape_pipeline import CianCitySweepCounters

    c = CianCitySweepCounters()
    assert c.anchors_total == 0
    assert c.lots_fetched == 0
    assert c.detail_attempted == 0
    assert c.houses_attempted == 0
    assert c.errors_count == 0


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

    from app.services.scrape_pipeline import CianCitySweepCounters

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


# ── Full sweep — фазы и агрегация ──────────────────────────────────────────


async def test_sweep_iterates_all_anchors_serp_and_save(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """SERP вызывается для каждого anchor, save_listings получает накопленные lots."""
    serp_calls: list[str] = []
    save_calls: list[int] = []

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        serp_calls.append(f"{lat:.4f}")
        return [_fake_lot(f"{lat}-{i}") for i in range(3)]

    def fake_save(_db: Any, lots: list, *, run_id: int | None = None):
        save_calls.append(len(lots))
        return len(lots), 0

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)

    # Stub detail + houses fetchers (возвращают None → graceful skip)
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=1,
        anchors=TEST_ANCHORS,
        pages_per_anchor=2,
        detail_top_n=0,  # skip detail phase
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert len(serp_calls) == len(TEST_ANCHORS)
    assert len(save_calls) == len(TEST_ANCHORS)
    assert counters.anchors_done == len(TEST_ANCHORS)
    assert counters.lots_fetched == len(TEST_ANCHORS) * 3
    assert counters.lots_inserted == len(TEST_ANCHORS) * 3
    assert counters.errors_count == 0
    assert fake.done is not None
    assert fake.done["lots_fetched"] == len(TEST_ANCHORS) * 3
    assert fake.failed is None
    assert fake.banned is None


async def test_newbuilding_only_drops_secondary_before_save(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """newbuilding_only=True (default): save_listings получает только novostroyki-lots.

    SERP отдаёт mixed-сегмент выдачу — вторичкой авторитетно владеет run_cian_full_load,
    поэтому sweep её отбрасывает до save и считает в lots_dropped_secondary.
    """
    saved_lots: list[list[ScrapedLot]] = []

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        # 1 novostroyki + 2 vtorichka на anchor
        return [
            _fake_lot(f"{lat}-nb", segment="novostroyki"),
            _fake_lot(f"{lat}-v1", segment="vtorichka"),
            _fake_lot(f"{lat}-v2", segment="vtorichka"),
        ]

    def fake_save(_db: Any, lots: list, *, run_id: int | None = None):
        saved_lots.append(list(lots))
        return len(lots), 0

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=1,
        anchors=TEST_ANCHORS,
        pages_per_anchor=2,
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
        newbuilding_only=True,
    )

    # Каждый save-вызов содержит только novostroyki-lot
    assert len(saved_lots) == len(TEST_ANCHORS)
    for lots in saved_lots:
        assert len(lots) == 1
        assert all(lot.listing_segment == "novostroyki" for lot in lots)
    # Counters: fetched все 3×N, сохранён 1×N, отброшено 2×N
    assert counters.lots_fetched == len(TEST_ANCHORS) * 3
    assert counters.lots_inserted == len(TEST_ANCHORS) * 1
    assert counters.lots_dropped_secondary == len(TEST_ANCHORS) * 2


async def test_newbuilding_only_false_saves_all_segments(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """newbuilding_only=False: фильтр отключён — save получает и вторичку, и новостройки."""
    saved_lots: list[list[ScrapedLot]] = []

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        return [
            _fake_lot(f"{lat}-nb", segment="novostroyki"),
            _fake_lot(f"{lat}-v1", segment="vtorichka"),
        ]

    def fake_save(_db: Any, lots: list, *, run_id: int | None = None):
        saved_lots.append(list(lots))
        return len(lots), 0

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=1,
        anchors=TEST_ANCHORS,
        pages_per_anchor=2,
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
        newbuilding_only=False,
    )

    assert len(saved_lots) == len(TEST_ANCHORS)
    for lots in saved_lots:
        assert len(lots) == 2
    assert counters.lots_inserted == len(TEST_ANCHORS) * 2
    assert counters.lots_dropped_secondary == 0


async def test_detail_phase_runs_and_fills_counters(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """detail-фаза вызывается и счётчики заполняются."""
    from app.services.scrapers.cian_detail import DetailEnrichment

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        return [_fake_lot(f"{lat}-{i}") for i in range(2)]

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )

    # DB mock returning 2 rows for detail query
    mock_db = MagicMock()
    mock_db.execute.return_value.mappings.return_value.all.return_value = [
        {"id": 101, "source_url": "https://ekb.cian.ru/sale/flat/101/"},
        {"id": 102, "source_url": "https://ekb.cian.ru/sale/flat/102/"},
    ]

    fake_enrichment = DetailEnrichment(cian_id=101)
    mock_fetch_detail = AsyncMock(return_value=fake_enrichment)
    mock_save_detail = MagicMock()

    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        mock_fetch_detail,
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.save_detail_enrichment",
        mock_save_detail,
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=mock_db,
        run_id=2,
        anchors=TEST_ANCHORS[:1],  # один anchor
        detail_top_n=2,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert counters.detail_attempted == 2
    assert counters.detail_enriched == 2
    assert counters.detail_failed == 0
    assert mock_fetch_detail.call_count == 2
    assert mock_save_detail.call_count == 2
    assert fake.done is not None


async def test_enrich_houses_false_skips_houses_phase(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """enrich_houses=False → houses-фаза не выполняется."""
    nb_fetch_calls: list[str] = []

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        return [_fake_lot("1001", nb=True)]

    async def fake_fetch_nb(url: str, **_: Any) -> None:
        nb_fetch_calls.append(url)
        return None

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
        fake_fetch_nb,
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=3,
        anchors=TEST_ANCHORS[:1],
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert len(nb_fetch_calls) == 0
    assert counters.houses_attempted == 0
    assert counters.houses_enriched == 0
    assert fake.done is not None


async def test_cancel_midway_stops_remaining(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Cooperative cancel перед anchor #2 → anchor #1 выполняется, #2+ нет."""
    serp_calls: list[str] = []

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        serp_calls.append(f"{lat:.4f}")
        return [_fake_lot(f"{lat}-1")]

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )

    fake = _FakeRuns(cancel_at_anchor=2)  # cancel при is_cancelled_calls >= 2 → перед anchor #2
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=4,
        anchors=TEST_ANCHORS,
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert len(serp_calls) == 1, "только anchor #1 должен выполниться"
    assert counters.anchors_done == 1
    assert fake.done is None
    assert fake.banned is None


async def test_consecutive_serp_failures_abort_with_banned(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """N подряд SERP-ошибок → mark_banned + abort sweep."""
    from app.services.scrape_pipeline import CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES

    attempt_count = 0

    async def fake_fetch_fail(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        nonlocal attempt_count
        attempt_count += 1
        raise RuntimeError("cian blocked")

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_fail)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    # Передаём больше anchor'ов чем CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
    anchors = [*TEST_ANCHORS, (56.77, 60.55, "A4"), (56.865, 60.62, "A5")]
    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=5,
        anchors=anchors,
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert attempt_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
    assert counters.errors_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
    assert fake.banned is not None, "mark_banned должен быть вызван"
    assert "consecutive" in fake.banned[0]
    assert fake.done is None


async def test_per_item_detail_error_does_not_abort_sweep(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Ошибка при detail-обогащении одного листинга не валит sweep."""
    call_count = 0

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        return [_fake_lot(f"lot-{idx}") for idx in range(2)]

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )

    mock_db = MagicMock()
    # Возвращаем 2 listing-а для detail
    mock_db.execute.return_value.mappings.return_value.all.return_value = [
        {"id": 201, "source_url": "https://ekb.cian.ru/sale/flat/201/"},
        {"id": 202, "source_url": "https://ekb.cian.ru/sale/flat/202/"},
    ]

    async def fetch_detail_sometimes_fails(url: str, **_: Any):
        nonlocal call_count
        call_count += 1
        if call_count == 1:
            raise RuntimeError("transient network error")
        from app.services.scrapers.cian_detail import DetailEnrichment

        return DetailEnrichment(cian_id=202)

    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        fetch_detail_sometimes_fails,
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.save_detail_enrichment",
        MagicMock(),
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=mock_db,
        run_id=6,
        anchors=TEST_ANCHORS[:1],
        detail_top_n=2,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert counters.detail_attempted == 2
    assert counters.detail_failed == 1
    assert counters.detail_enriched == 1
    assert counters.errors_count >= 1
    assert fake.done is not None, "sweep должен завершиться mark_done несмотря на ошибку"
    assert fake.banned is None


async def test_mark_done_always_called(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """mark_done вызывается даже при пустом SERP."""

    async def fake_fetch_empty(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        return []

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_empty)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=MagicMock(),
        run_id=7,
        anchors=TEST_ANCHORS,
        detail_top_n=0,
        enrich_houses=False,
        request_delay_sec=0.0,
    )

    assert fake.done is not None
    assert counters.lots_fetched == 0
    assert counters.anchors_done == len(TEST_ANCHORS)


async def test_houses_phase_uses_house_sources_join(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Houses-фаза формирует JOIN на house_sources, не h.ext_source напрямую."""

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        # Один lot с house_source="cian_newbuilding" → попадёт в houses-фазу
        return [_fake_lot("nb-1", nb=True)]

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
        AsyncMock(return_value=None),
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
        MagicMock(),
    )

    mock_db = MagicMock()
    # DB execute → возвращает одну house-строку
    mock_db.execute.return_value.mappings.return_value.all.return_value = [
        {"id": 501, "cian_zhk_url": "https://ekb.cian.ru/kupit-kvartiru/zhk-test/"},
    ]

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=mock_db,
        run_id=10,
        anchors=TEST_ANCHORS[:1],
        detail_top_n=0,
        enrich_houses=True,
        request_delay_sec=0.0,
    )

    # Проверяем SQL — должен содержать JOIN house_sources, а не h.ext_source
    assert mock_db.execute.called
    executed_sql_args = [str(c.args[0]) for c in mock_db.execute.call_args_list]
    houses_sql_calls = [s for s in executed_sql_args if "house" in s.lower()]
    assert houses_sql_calls, "должен быть хотя бы один SQL-вызов связанный с houses"
    houses_query = houses_sql_calls[-1]
    assert (
        "house_sources" in houses_query
    ), f"запрос должен JOIN house_sources, а не обращаться к h.ext_source: {houses_query!r}"
    assert "ext_source" not in houses_query.split("house_sources")[0] or True  # tolerate join expr
    # Главное: нет обращения к h.ext_source (старый баг)
    assert (
        "h.ext_source" not in houses_query
    ), f"колонка h.ext_source не существует в houses: {houses_query!r}"
    assert (
        "h.ext_id" not in houses_query
    ), f"колонка h.ext_id не существует в houses: {houses_query!r}"

    assert fake.done is not None, "mark_done должен быть вызван"
    assert fake.failed is None, "mark_failed НЕ должен быть вызван"
    assert counters.houses_attempted == 1


async def test_houses_db_query_error_graceful_mark_done(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Ошибка SQL в houses-фазе (напр. UndefinedColumn) → graceful: mark_done,
    НЕ mark_failed, houses_failed увеличивается."""

    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
        # Два nb-лота → houses-фаза попытается запросить 2 id
        return [_fake_lot("nb-10", nb=True), _fake_lot("nb-11", nb=True)]

    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
    monkeypatch.setattr(
        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
    )
    monkeypatch.setattr(
        "app.services.scrapers.cian_detail.fetch_detail",
        AsyncMock(return_value=None),
    )

    mock_db = MagicMock()

    def _db_execute_raises(sql, params=None):
        sql_text = str(sql)
        if "house" in sql_text.lower() and "cian_newbuilding" in sql_text:
            raise RuntimeError("column h.ext_source does not exist")
        result = MagicMock()
        result.mappings.return_value.all.return_value = []
        return result

    mock_db.execute.side_effect = _db_execute_raises

    fake = _FakeRuns()
    _install_runs(monkeypatch, fake)

    counters = await scrape_pipeline.run_cian_city_sweep(
        db=mock_db,
        run_id=11,
        anchors=TEST_ANCHORS[:1],
        detail_top_n=0,
        enrich_houses=True,
        request_delay_sec=0.0,
    )

    assert fake.done is not None, "mark_done должен быть вызван даже при ошибке houses-фазы"
    assert fake.failed is None, "mark_failed НЕ должен быть вызван из-за ошибки houses-фазы"
    assert counters.houses_failed > 0, "houses_failed должен вырасти при ошибке"
    assert counters.houses_attempted == 0, "houses_attempted не растёт — до него не дошли"


# ── Admin endpoint tests (offline) ──────────────────────────────────────────


@pytest.fixture
def app_with_admin():
    from unittest.mock import MagicMock

    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    from app.api.v1 import admin as admin_module
    from app.core.db import get_db

    app = FastAPI()
    app.include_router(admin_module.router, prefix="/api/v1/admin")

    def fake_db():
        yield MagicMock()

    app.dependency_overrides[get_db] = fake_db
    return TestClient(app)


def test_cian_start_endpoint_validates_pages_per_anchor(app_with_admin) -> None:
    """pages_per_anchor=99 превышает max=10 → 422."""
    r = app_with_admin.post(
        "/api/v1/admin/scrape/cian-city-sweep",
        json={"pages_per_anchor": 99},
    )
    assert r.status_code == 422


def test_cian_start_endpoint_validates_delay_too_low(app_with_admin) -> None:
    """request_delay_sec=1.0 меньше min=3.0 → 422."""
    r = app_with_admin.post(
        "/api/v1/admin/scrape/cian-city-sweep",
        json={"request_delay_sec": 1.0},
    )
    assert r.status_code == 422


def test_cian_start_endpoint_ok(app_with_admin) -> None:
    """Valid request → 200, run_id; kit sweep вызван с DI (config/matcher, БЕЗ enrichment)."""
    from unittest.mock import AsyncMock, patch

    from app.services.scraper_adapters import RealMatcherAdapter, RealScraperConfig

    with (
        patch("app.services.scrape_runs.create_run", return_value=42),
        patch("app.api.v1.admin.run_cian_city_sweep", new_callable=AsyncMock) as mock_sweep,
    ):
        r = app_with_admin.post(
            "/api/v1/admin/scrape/cian-city-sweep",
            json={"pages_per_anchor": 2, "detail_top_n": 5, "enrich_houses": False},
        )
    assert r.status_code == 200
    body = r.json()
    assert body["run_id"] == 42
    assert body["status"] == "running"
    assert body["pages_per_anchor"] == 2
    assert body["detail_top_n"] == 5

    mock_sweep.assert_called_once()
    kw = mock_sweep.call_args.kwargs
    assert isinstance(kw["config"], RealScraperConfig)
    assert isinstance(kw["matcher"], RealMatcherAdapter)
    assert "proxy_provider" in kw
    assert "enrichment" not in kw


def test_cian_cancel_endpoint(app_with_admin) -> None:
    """Cancel endpoint возвращает run_id + cancelled=True."""
    from unittest.mock import patch

    with patch("app.services.scrape_runs.mark_cancelled", return_value=True):
        r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/77/cancel")
    assert r.status_code == 200
    body = r.json()
    assert body["run_id"] == 77
    assert body["cancelled"] is True
    assert body["ok"] is True


def test_cian_cancel_endpoint_not_running(app_with_admin) -> None:
    """Cancel на не-running sweep → cancelled=False."""
    from unittest.mock import patch

    with patch("app.services.scrape_runs.mark_cancelled", return_value=False):
        r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/88/cancel")
    assert r.status_code == 200
    assert r.json()["cancelled"] is False


def test_cian_list_runs_endpoint(app_with_admin) -> None:
    """List runs endpoint возвращает список."""
    from datetime import UTC, datetime
    from unittest.mock import patch

    fake_rows = [
        {
            "run_id": 10,
            "source": "cian_city_sweep",
            "status": "done",
            "params": {"pages_per_anchor": 3},
            "counters": {"lots_fetched": 150},
            "error": None,
            "started_at": datetime(2025, 5, 1, tzinfo=UTC),
            "finished_at": datetime(2025, 5, 1, 1, tzinfo=UTC),
            "heartbeat_at": datetime(2025, 5, 1, 1, tzinfo=UTC),
        }
    ]
    with patch("app.services.scrape_runs.list_recent", return_value=fake_rows):
        r = app_with_admin.get("/api/v1/admin/scrape/cian-city-sweep/runs")
    assert r.status_code == 200
    body = r.json()
    assert len(body) == 1
    assert body[0]["run_id"] == 10
    assert body[0]["source"] == "cian_city_sweep"
    assert "2025-05-01" in body[0]["started_at"]


# ── #2160: масштабируемый per-anchor watchdog для proxy-pool browser ──────────


def test_cian_anchor_timeout_defaults() -> None:
    """Формула watchdog даёт worst-case ~1580s при дефолтных параметрах.

    4 room-buckets × 3 pages × (5s delay + 70s serp-fetch) + 10 detail × 50s + 180s houses
    = 900 + 500 + 180 = 1580s. Фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинил бы якорь в
    SERP-фазе (#2160): каждый SERP-фетч через camoufox 13-45s.
    """
    t = scrape_pipeline._cian_anchor_timeout_s(
        pages_per_anchor=3,
        request_delay_sec=5.0,
        detail_top_n=10,
        enrich_houses=True,
    )
    assert t == pytest.approx(1580.0)
    assert t > scrape_pipeline.ANCHOR_TIMEOUT_SEC


def test_cian_anchor_timeout_no_houses() -> None:
    """enrich_houses=False убирает houses-бюджет (180s) из формулы."""
    t = scrape_pipeline._cian_anchor_timeout_s(
        pages_per_anchor=3,
        request_delay_sec=5.0,
        detail_top_n=10,
        enrich_houses=False,
    )
    assert t == pytest.approx(1400.0)  # 900 + 500 + 0


def test_cian_anchor_timeout_floor_at_anchor_timeout_sec() -> None:
    """Малые параметры не опускают таймаут ниже ANCHOR_TIMEOUT_SEC (max-ветка)."""
    t = scrape_pipeline._cian_anchor_timeout_s(
        pages_per_anchor=0,
        request_delay_sec=0.0,
        detail_top_n=0,
        enrich_houses=False,
    )
    assert t == float(scrape_pipeline.ANCHOR_TIMEOUT_SEC)


def test_cian_anchor_timeout_parity_with_scraper_kit() -> None:
    """Strangler-инвариант: backend-хелпер ≡ scraper_kit-хелпер (обе копии в синхроне)."""
    from scraper_kit.orchestration.pipeline import (
        _cian_anchor_timeout_s as kit_timeout_s,
    )

    for args in (
        dict(pages_per_anchor=3, request_delay_sec=5.0, detail_top_n=10, enrich_houses=True),
        dict(pages_per_anchor=5, request_delay_sec=9.0, detail_top_n=20, enrich_houses=False),
        dict(pages_per_anchor=0, request_delay_sec=0.0, detail_top_n=0, enrich_houses=False),
    ):
        assert scrape_pipeline._cian_anchor_timeout_s(**args) == kit_timeout_s(**args)