gendesign/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py

from __future__ import annotations

import asyncio
import os
import sys
from unittest.mock import AsyncMock, MagicMock, patch

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

_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)

import pytest  # noqa: E402

from app.core import shutdown as _sd  # noqa: E402
from app.tasks.avito_detail_backfill import (  # noqa: E402
    AvitoDetailBackfillResult,
    run_avito_detail_backfill,
)


@pytest.fixture(autouse=True)
def _reset_shutdown() -> None:
    """shutdown — module-global Event: чистим вокруг каждого теста (изоляция #1182)."""
    _sd.reset_shutdown()
    yield
    _sd.reset_shutdown()


@pytest.fixture(autouse=True)
def _patch_build_warmed() -> object:
    """use_curl-ветка (#1551) строит прогретую сессию через build_warmed_session и
    освежает куки через research_in_session (реальный yandex/avito GET + sleep).
    fetch_detail в этих тестах всё равно замокан — мокаем обе не-сетевыми AsyncMock,
    чтобы setup / periodic re-warm / on-block cooldown не ходили в сеть."""
    with (
        patch(
            "app.tasks.avito_detail_backfill.build_warmed_session",
            AsyncMock(return_value=AsyncMock()),
        ),
        patch(
            "app.tasks.avito_detail_backfill.research_in_session",
            AsyncMock(return_value=True),
        ) as research,
    ):
        yield research


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_snapshot(n: int) -> list[dict]:
    return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)]


def _mock_db(snapshot: list[dict]) -> MagicMock:
    """Fake Session: first execute() returns snapshot via .mappings().all()."""
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
_SETTINGS = "app.tasks.avito_detail_backfill.settings"
_SHUTDOWN = "app.tasks.avito_detail_backfill.shutdown_requested"
_BUILD_WARM = "app.tasks.avito_detail_backfill.build_warmed_session"
_RESEARCH = "app.tasks.avito_detail_backfill.research_in_session"

# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_empty_snapshot_marks_done() -> None:
    """Empty snapshot -> mark_done immediately, no fetch calls."""
    db = _mock_db([])
    runs = MagicMock()
    fake_session = AsyncMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=fake_session),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH) as mock_fetch,
    ):
        result = await run_avito_detail_backfill(
            db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
        )

    assert isinstance(result, AvitoDetailBackfillResult)
    assert result.attempted == 0
    assert result.enriched == 0
    mock_fetch.assert_not_called()
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_processes_snapshot_to_completion() -> None:
    """3 listings -> all fetched and enriched, mark_done called."""
    from app.services.scraper_adapters import RealScraperConfig

    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER) as mock_scraper_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.attempted == 3
    assert result.enriched == 3
    assert result.blocked == 0
    assert result.failed == 0
    assert mock_fetch.call_count == 3
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()
    # #2310 regression guard: kit fetch_detail silently drops the backconnect-
    # on-403 retry (and kit AvitoScraper can't read avito_proxy_rotate_url at
    # all) unless config=RealScraperConfig() is passed/injected at the call
    # site — assert_called()/call_count alone wouldn't catch someone dropping
    # that kwarg later (mirrors #2306's test_backfill_wave2.py:282-286 pattern).
    _, fetch_call_kwargs = mock_fetch.call_args
    assert isinstance(fetch_call_kwargs.get("config"), RealScraperConfig)
    assert isinstance(mock_scraper_cls.call_args.args[0], RealScraperConfig)


@pytest.mark.asyncio
async def test_backfill_build_warmed_session_receives_config() -> None:
    """#2397 Part D1 / #2330 regression guard: kit's build_warmed_session() now accepts
    config=ScraperConfig and silently drops settings.scraper_proxy_url (sticky МГТС-прокси)
    -- NO crash, just a proxy-less session -- unless config=RealScraperConfig() is passed
    at the call site. use_curl=True is the prod-default path (avito_detail_backfill_use_curl),
    so this is the setup-time build_warmed_session() call (~line 177). assert_awaited_once()
    alone wouldn't catch someone dropping the kwarg later (mirrors the fetch_detail config=
    guard above, lines 141-148)."""
    from app.services.scraper_adapters import RealScraperConfig

    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_BUILD_WARM, AsyncMock(return_value=AsyncMock())) as mock_build,
    ):
        result = await run_avito_detail_backfill(
            db, run_id=15, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.enriched == 1
    mock_build.assert_awaited_once()
    _, build_call_kwargs = mock_build.call_args
    assert isinstance(build_call_kwargs.get("config"), RealScraperConfig)
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_blocked_abort_after_max_consecutive() -> None:
    """5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed).

    #1950 abort-reorder: abort-check ПЕРЕД recovery → на 5-м (аборт-)блоке rotate_ip
    НЕ дёргается (не тратим recovery на финальном блоке). rotate_ip x4 (блоки 1-4).
    """
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(10)
    db = _mock_db(snapshot)
    runs = MagicMock()
    blocked_exc = AvitoBlockedError("ip blocked")
    mock_fetch = AsyncMock(side_effect=blocked_exc)
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
        )

    assert result.blocked == 5
    assert result.attempted == 5
    assert result.enriched == 0
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()
    # abort-check до recovery → 5-й блок абортит без rotate; rotate только на блоках 1-4.
    assert mock_scraper.return_value._rotate_ip.call_count == 4


@pytest.mark.asyncio
async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None:
    """#1182 Phase 2: shutdown_requested() True → loop выходит на границе карточки,
    mark_done вызывается с ЧАСТИЧНЫМИ счётчиками (не mark_failed, не mark_cancelled).

    side_effect [False, True]: 1-я карточка обрабатывается (attempted=1, enriched=1),
    перед 2-й приходит SIGTERM-drain → break. Snapshot — pending-query, остаток
    до-резюмит следующий run (resume_cursor не нужен).
    """
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_SHUTDOWN, side_effect=[False, True]),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=13, params={"batch_size": 10, "budget_sec": 3600}
        )

    # Обработана только 1-я карточка, на 2-й — drain-break.
    assert result.attempted == 1
    assert result.enriched == 1
    assert mock_fetch.call_count == 1
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()
    runs.mark_cancelled.assert_not_called()
    # mark_done получил ЧАСТИЧНЫЕ счётчики (attempted=1, а не весь snapshot=3).
    done_counters = runs.mark_done.call_args.args[2]
    assert done_counters["attempted"] == 1


@pytest.mark.asyncio
async def test_backfill_budget_guard_stops_loop() -> None:
    """Budget expired before first listing -> fetch_detail not called."""
    snapshot = _make_snapshot(5)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mono_values = iter([0.0, 999.0, 999.0])
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch("app.tasks.avito_detail_backfill.time.monotonic", side_effect=mono_values),
        patch(_FETCH) as mock_fetch,
    ):
        await run_avito_detail_backfill(db, run_id=4, params={"batch_size": 5, "budget_sec": 1})

    mock_fetch.assert_not_called()
    runs.mark_done.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_top_level_exception_marks_failed() -> None:
    """db.execute raises -> mark_failed called, exception re-raised."""
    db = MagicMock()
    db.execute.side_effect = RuntimeError("DB connection lost")
    runs = MagicMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
    ):
        with pytest.raises(RuntimeError, match="DB connection lost"):
            await run_avito_detail_backfill(
                db, run_id=5, params={"batch_size": 5, "budget_sec": 60}
            )

    runs.mark_failed.assert_called_once()
    runs.mark_done.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_rotate_ip_called_on_each_block() -> None:
    """1 block + 1 success -> rotate_ip called once, enriched=1."""
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    blocked_exc = AvitoBlockedError("x")
    mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=6, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
        )

    assert result.enriched == 1
    assert result.blocked == 1
    assert mock_scraper.return_value._rotate_ip.call_count == 1
    runs.mark_done.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_snapshot_filters_ekb_active_only() -> None:
    """Снапшот-SELECT (#1814) фильтрует только активные ЕКБ-листинги.

    Проверяем, что текст запроса содержит `is_active = TRUE` и
    `LIKE '%/ekaterinburg/%'` — legacy не-ЕКБ (moskva/spb/tyumen) и мёртвые
    листинги не попадают в фетч, иначе browser спотыкается → curl-бан 429.
    """
    db = _mock_db([])
    runs = MagicMock()
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH),
    ):
        await run_avito_detail_backfill(db, run_id=8, params={"batch_size": 10, "budget_sec": 60})

    # Первый (и единственный при пустом снапшоте) execute — это SELECT-снапшот.
    snapshot_call = db.execute.call_args_list[0]
    sql_text = str(snapshot_call.args[0])
    assert "is_active = TRUE" in sql_text
    assert "/ekaterinburg/" in sql_text
    assert "detail_enriched_at IS NULL" in sql_text
    assert "(lat IS NULL) DESC" in sql_text


@pytest.mark.asyncio
async def test_backfill_fetch_exception_continues() -> None:
    """RuntimeError on one listing -> failed++, db.rollback(), loop continues for next."""
    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment])
    fake_settings = MagicMock(scraper_fetch_mode="cffi")
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=7, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.failed == 1
    assert result.enriched == 1
    assert result.attempted == 2
    db.rollback.assert_called()
    runs.mark_done.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_fetch_timeout_skips_and_continues() -> None:
    """#1950: один fetch_detail зависает дольше hard-timeout → этот листинг failed,
    loop НЕ зависает, переходит к следующему, run завершается mark_done (не zombie).

    Регрессия run 423 (завис 7.7ч → reaped): fetch_detail без timeout блокировал
    loop навсегда. asyncio.wait_for(timeout) отменяет зависший fetch → TimeoutError.
    """
    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()

    call_urls: list[str] = []

    async def _fetch(
        url: str,
        *,
        cffi_session: object = None,
        browser_fetcher: object = None,
        referer: object = None,
        reconnect_on_block: bool = True,
        config: object = None,
    ):
        call_urls.append(url)
        if len(call_urls) == 1:
            await asyncio.Event().wait()  # висит вечно → wait_for отменит по timeout
        return mock_enrichment

    # Короткий timeout (50ms) — тест не ждёт реальные 90s; rotate-settle тоже 0.
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_fetch_timeout_s=0.05,
        avito_proxy_rotate_settle_s=0.0,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER),
        patch(_RUNS, runs),
        patch(_FETCH, _fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=11, params={"batch_size": 10, "budget_sec": 3600}
        )

    # Первый листинг — failed (timeout), второй — enriched. Run завершён mark_done.
    assert result.failed == 1
    assert result.enriched == 1
    assert result.attempted == 2
    assert len(call_urls) == 2, "loop должен дойти до второго листинга, а не зависнуть"
    db.rollback.assert_called()
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()


@pytest.mark.asyncio
async def test_backfill_listing_gone_marks_inactive_no_breaker() -> None:
    """#2034: AvitoListingGoneError (мёртвый 404-листинг) → counters.gone++, blocked
    НЕ растёт, consecutive-block breaker НЕ абортит, листинг помечается is_active=FALSE.

    Регрессия run 458 (attempted=5 enriched=0 blocked=5 → abort): lat-null очередь
    состоит из dead-листингов; раньше 404 ловился как soft-block → breaker абортил run
    до live-листингов. Теперь 404 нейтрален к breaker'у и метит листинг inactive.
    """
    from scraper_kit.avito_exceptions import AvitoListingGoneError

    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
    mock_scraper = MagicMock()
    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
    # use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
    # use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION, return_value=AsyncMock()),
        patch(_SCRAPER, mock_scraper),
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=12,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
        )

    assert result.gone == 1
    assert result.blocked == 0
    assert result.attempted == 1
    assert result.enriched == 0
    # breaker НЕ абортил: rotate_ip НЕ дёргался (gone ≠ block), run завершён mark_done.
    mock_scraper.return_value._rotate_ip.assert_not_called()
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()
    # UPDATE listings SET is_active = FALSE по row id=1 выполнен (мок db.execute).
    update_calls = [
        c
        for c in db.execute.call_args_list
        if "UPDATE listings SET is_active = FALSE" in str(c.args[0])
    ]
    assert len(update_calls) == 1
    assert update_calls[0].args[1] == {"id": 1}


_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"


@pytest.mark.asyncio
async def test_backfill_use_curl_flag_skips_browser_fetcher() -> None:
    """avito_detail_backfill_use_curl=True: BrowserFetcher не создаётся,
    fetch_detail вызывается с browser_fetcher=None (curl/backconnect путь).
    """
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    # Флаг use_curl=True, fetch_mode=browser (но флаг перекрывает)
    fake_settings = MagicMock(
        scraper_fetch_mode="browser",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_BROWSER_FETCHER) as mock_bf_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=9, params={"batch_size": 10, "budget_sec": 3600}
        )

    # BrowserFetcher не должен быть создан
    mock_bf_cls.assert_not_called()
    # fetch_detail вызван с browser_fetcher=None (curl путь)
    assert mock_fetch.call_count == 1
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("browser_fetcher") is None
    assert result.enriched == 1
    runs.mark_done.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_use_curl_false_creates_browser_fetcher() -> None:
    """avito_detail_backfill_use_curl=False + scraper_fetch_mode='browser':
    BrowserFetcher создаётся (legacy browser/auv путь).
    """
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    mock_fetch = AsyncMock(return_value=mock_enrichment)
    mock_save = MagicMock(return_value=True)
    fake_settings = MagicMock(
        scraper_fetch_mode="browser",
        avito_detail_backfill_use_curl=False,
    )
    mock_bf_instance = AsyncMock()
    mock_bf_instance.__aenter__ = AsyncMock(return_value=mock_bf_instance)
    mock_bf_instance.__aexit__ = AsyncMock(return_value=False)
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION),
        patch(_SCRAPER),
        patch(_BROWSER_FETCHER, return_value=mock_bf_instance) as mock_bf_cls,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, mock_save),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_avito_detail_backfill(
            db, run_id=10, params={"batch_size": 10, "budget_sec": 3600}
        )

    # BrowserFetcher должен быть создан (source="avito", endpoint из settings — #2310
    # kit BrowserFetcher требует endpoint= обязательным keyword-only параметром)
    mock_bf_cls.assert_called_once_with(
        source="avito", endpoint=fake_settings.browser_http_endpoint
    )
    # fetch_detail вызван с browser_fetcher установленным (не None)
    assert mock_fetch.call_count == 1
    _, kwargs = mock_fetch.call_args
    assert kwargs.get("browser_fetcher") is not None
    assert result.enriched == 1
    runs.mark_done.assert_called_once()


@pytest.mark.asyncio
async def test_backfill_use_curl_block_cooldown_research_no_rebuild() -> None:
    """#1551 sticky-IP: on-block (use_curl) даёт cooldown + in-session research, НЕ
    пересоздаёт прогретую сессию и НЕ дёргает changeip-ротацию.

    МГТС sticky — один фикс. exit-IP (rebuild != новый IP), уйти на свежий IP софтом
    нельзя. Блок = rate-limit текущего IP → cooldown + re-search той же сессией.
    fetch: 1-й вызов BLOCKED, 2-й — успех.
    """
    from scraper_kit.avito_exceptions import AvitoBlockedError

    snapshot = _make_snapshot(2)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_enrichment = MagicMock()
    blocked_exc = AvitoBlockedError("rate-limited")
    mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
    warmed_session = AsyncMock()
    fake_settings = MagicMock(
        scraper_fetch_mode="cffi",
        avito_detail_backfill_use_curl=True,
    )
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SCRAPER) as mock_scraper,
        patch(_RUNS, runs),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
        patch(_BUILD_WARM, AsyncMock(return_value=warmed_session)) as mock_build,
        patch(_RESEARCH, new_callable=AsyncMock) as mock_research,
    ):
        result = await run_avito_detail_backfill(
            db,
            run_id=14,
            params={
                "batch_size": 10,
                "budget_sec": 3600,
                "max_consecutive_blocks": 5,
                "block_cooldown_sec": 0.0,
            },
        )

    assert result.blocked == 1
    assert result.enriched == 1
    assert result.attempted == 2
    # on-block: research_in_session вызван (освежить куки in-session)...
    mock_research.assert_awaited_once()
    # ...а build_warmed_session НЕ дёргался повторно (вызван только 1× в setup —
    # прогретая сессия НЕ пересоздаётся на блоке, sticky-IP rebuild бесполезен).
    assert mock_build.await_count == 1
    # changeip-ротация (legacy путь) под use_curl НЕ дёргается.
    mock_scraper.return_value._rotate_ip.assert_not_called()
    runs.mark_done.assert_called_once()
    runs.mark_failed.assert_not_called()