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

"""Tests для тёплого переиспользуемого browser-context в DomClick detail-бэкфилле (#3118).

Проблема: sidecar's browser.new_page() создавал НОВЫЙ изолированный context на КАЖДЫЙ
/fetch, поэтому кук-инъекция (см. domclick_session), сделанная ОДИН РАЗ в начале
прогона, не видела Set-Cookie от площадки — живой qrator_jsid2 (~2.5ч TTL) протухал
уже после первого запроса. Живой замер: 26 подряд sidecar-фетчей = 100% блоков, те же
карточки в тёплом browser-контексте = 5/5 успешно.

Фикс: BrowserFetcher(reuse_context=True) держит per-provider context живым между
вызовами; куки вливаются один раз при его создании. Сброс сожжённого context'а
оркестратор просит через bf.request_context_reset() — так как fetch_detail()
(scraper_kit, намеренно не в scope того фикса) не прокидывает reset_context.

#3212 сузил условие сброса: НЕ на каждый блок, а один раз за прогон. Пропуск QRATOR
(qrator_jsid2 + qrator_jsr) живёт в контексте, и сброс его выбрасывает; повторная
валидация с того же IP сразу после принятой получает 403. Сброс на каждый блок поэтому
гарантировал следующий блок — приёмка #3118 и дала «1 успех из 10».

Зеркалит конвенции tests/tasks/test_domclick_detail_backfill.py (module-level
patch-target строки, _mock_db/_mock_session_svc helpers).
"""

from __future__ import annotations

import os
import sys
from datetime import UTC, datetime, timedelta
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 scraper_kit.domclick_exceptions import DomClickBlockedError  # noqa: E402

from app.tasks.domclick_detail_backfill import run_domclick_detail_backfill  # noqa: E402

_FETCH = "app.tasks.domclick_detail_backfill.fetch_detail"
_SAVE = "app.tasks.domclick_detail_backfill.save_detail_enrichment"
_RUNS = "app.tasks.domclick_detail_backfill.runs_mod"
_SLEEP = "app.tasks.domclick_detail_backfill.asyncio.sleep"
_SETTINGS = "app.tasks.domclick_detail_backfill.settings"
_BROWSER_FETCHER = "app.tasks.domclick_detail_backfill.BrowserFetcher"
_SESSION_SVC = "app.tasks.domclick_detail_backfill.domclick_session_svc"


def _make_snapshot(n: int) -> list[dict]:
    return [
        {
            "id": i + 1,
            "source_url": f"https://ekaterinburg.domclick.ru/card/sale__flat__{i + 1}",
        }
        for i in range(n)
    ]


def _mock_db(snapshot: list[dict]) -> MagicMock:
    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = snapshot
    db.execute.return_value = sel
    return db


def _mock_session_svc(cookies: dict[str, str] | None) -> MagicMock:
    svc = MagicMock()
    svc.load_session.return_value = cookies
    svc.COOKIE_EXPIRY_WARN_DAYS = 5
    svc.session_expires_at.return_value = datetime.now(tz=UTC) + timedelta(days=30)
    return svc


def _mock_browser_fetcher_cls() -> tuple[MagicMock, AsyncMock]:
    """Возвращает (класс-мок, instance-мок) — instance с реальным request_context_reset."""
    instance = AsyncMock()
    instance.__aenter__ = AsyncMock(return_value=instance)
    instance.__aexit__ = AsyncMock(return_value=False)
    # request_context_reset — синхронный метод (не awaited caller'ом), явный MagicMock
    # вместо auto-AsyncMock-атрибута, чтобы вызов не оставлял неawaited coroutine.
    instance.request_context_reset = MagicMock()
    return MagicMock(return_value=instance), instance


@pytest.mark.asyncio
async def test_backfill_constructs_browser_fetcher_with_reuse_context() -> None:
    """BrowserFetcher поднимается с reuse_context=True (#3118) -- тёплый context sidecar'а."""
    snapshot = _make_snapshot(1)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
    mock_svc = _mock_session_svc({"CAS_ID": "123", "qrator_jsid2": "live"})
    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION_SVC, mock_svc),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        await run_domclick_detail_backfill(
            db, run_id=100, params={"batch_size": 10, "budget_sec": 3600}
        )

    # #3197: конструктор с 08.2026 несёт ещё proxy_provider/use_pool/environment
    # (проводка прокси-пула) -- этот тест проверяет только reuse_context, для полной
    # проводки см. tests/test_3197_domclick_proxy_pool_wiring.py.
    mock_bf_cls.assert_called_once()
    _, call_kwargs = mock_bf_cls.call_args
    assert call_kwargs["source"] == "domclick"
    assert call_kwargs["endpoint"] == fake_settings.browser_http_endpoint
    assert call_kwargs["reuse_context"] is True


@pytest.mark.asyncio
async def test_context_reset_requested_at_most_once_per_run() -> None:
    """#3212: сброс контекста запрашивается РОВНО ОДИН раз за прогон, а не на каждый блок.

    До #3212 сброс шёл на КАЖДЫЙ блок, и это само себя поддерживало: в контексте лежит
    пропуск QRATOR, сброс его выбрасывает, повторная валидация с того же IP сразу после
    принятой получает 403 — то есть следующий фетч блокировался гарантированно. Одна
    осечка превращалась в необратимый каскад (приёмка #3118: 1 успех из 10).
    Здесь 3 подряд блока -> сброс запрошен ровно 1 раз.
    """
    snapshot = _make_snapshot(10)
    db = _mock_db(snapshot)
    runs = MagicMock()
    blocked_exc = DomClickBlockedError("browser fetch failed")
    mock_fetch = AsyncMock(side_effect=blocked_exc)
    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
    mock_svc = _mock_session_svc({"CAS_ID": "123"})
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION_SVC, mock_svc),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_domclick_detail_backfill(
            db,
            run_id=101,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
        )

    assert result.blocked == 3
    assert instance.request_context_reset.call_count == 1, (
        "ожидалась ровно одна попытка начать с чистого листа за весь прогон"
    )


@pytest.mark.asyncio
async def test_successful_fetch_does_not_request_context_reset() -> None:
    """Успешный прогон без блоков -> request_context_reset() ни разу не вызывается."""
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    mock_fetch = AsyncMock(return_value=MagicMock())
    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
    mock_svc = _mock_session_svc({"CAS_ID": "123"})
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION_SVC, mock_svc),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_domclick_detail_backfill(
            db, run_id=102, params={"batch_size": 10, "budget_sec": 3600}
        )

    assert result.blocked == 0
    instance.request_context_reset.assert_not_called()


@pytest.mark.asyncio
async def test_block_then_recovery_requests_reset_only_on_the_block() -> None:
    """Блок -> успех -> успех: сброс запрошен РОВНО один раз (на самом блоке).

    Один сброс на блок, не на каждый последующий запрос (см. docstring задачи) --
    после восстановления (успешный fetch) счётчик подряд-блоков сбрасывается и
    request_context_reset() больше не должен вызываться.
    """
    snapshot = _make_snapshot(3)
    db = _mock_db(snapshot)
    runs = MagicMock()
    blocked_exc = DomClickBlockedError("browser fetch failed")
    mock_fetch = AsyncMock(side_effect=[blocked_exc, MagicMock(), MagicMock()])
    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
    mock_svc = _mock_session_svc({"CAS_ID": "123"})
    mock_bf_cls, instance = _mock_browser_fetcher_cls()
    with (
        patch(_SETTINGS, fake_settings),
        patch(_SESSION_SVC, mock_svc),
        patch(_RUNS, runs),
        patch(_BROWSER_FETCHER, mock_bf_cls),
        patch(_FETCH, mock_fetch),
        patch(_SAVE, return_value=True),
        patch(_SLEEP, new_callable=AsyncMock),
    ):
        result = await run_domclick_detail_backfill(
            db,
            run_id=103,
            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
        )

    assert result.blocked == 1
    assert result.enriched == 2
    instance.request_context_reset.assert_called_once()


# ── BrowserFetcher: сама логика отложенного флага (#3118) ──────────────────────
# Тесты выше мокают BrowserFetcher целиком и проверяют ОРКЕСТРАТОР («на блок зовём
# request_context_reset ровно один раз»). Ниже — реальный BrowserFetcher с моком
# _post_fetch: проверяем потребление флага, ради которого метод и заведён.


def _fetcher(*, reuse_context: bool):
    from scraper_kit.browser_fetcher import BrowserFetcher

    bf = BrowserFetcher(
        source="domclick", endpoint="http://browser:3000", reuse_context=reuse_context
    )
    bf._client = MagicMock()  # обходим assert «используй как async context manager»
    return bf


@pytest.mark.asyncio
async def test_3118_pending_reset_consumed_exactly_once():
    """request_context_reset() влияет РОВНО на один следующий fetch, дальше — обычные."""
    bf = _fetcher(reuse_context=True)
    seen: list[bool] = []

    async def _fake_post(url, origin=None, cookies=None, reset_context=False):
        seen.append(reset_context)
        return "<html></html>"

    with patch.object(bf, "_post_fetch", side_effect=_fake_post):
        await bf.fetch("https://ekaterinburg.domclick.ru/card/sale__flat__1")
        bf.request_context_reset()
        await bf.fetch("https://ekaterinburg.domclick.ru/card/sale__flat__2")
        await bf.fetch("https://ekaterinburg.domclick.ru/card/sale__flat__3")

    assert seen == [False, True, False], seen
    assert bf._context_reset_pending is False


@pytest.mark.asyncio
async def test_3118_pending_reset_survives_transport_retry():
    """Transport-ошибка → retry идёт с тем же reset (сброс на сайдкаре идемпотентен)."""
    import httpx

    bf = _fetcher(reuse_context=True)
    seen: list[bool] = []

    async def _fake_post(url, origin=None, cookies=None, reset_context=False):
        seen.append(reset_context)
        if len(seen) == 1:
            raise httpx.ConnectError("boom")
        return "<html></html>"

    with (
        patch.object(bf, "_post_fetch", side_effect=_fake_post),
        patch("scraper_kit.browser_fetcher.asyncio.sleep", new=AsyncMock()),
    ):
        bf.request_context_reset()
        await bf.fetch("https://ekaterinburg.domclick.ru/card/sale__flat__4")

    assert seen == [True, True], seen
    assert bf._context_reset_pending is False


@pytest.mark.asyncio
async def test_3118_payload_unchanged_when_reuse_context_off():
    """reuse_context=False (дефолт) → в payload /fetch новых ключей нет вовсе."""
    bf = _fetcher(reuse_context=False)
    captured: dict = {}

    class _Resp:
        @staticmethod
        def json():
            return {"html": "<html></html>"}

    async def _fake_client_post(url, json=None):
        captured.update(json or {})
        return _Resp()

    bf._client.post = _fake_client_post
    with (
        patch("scraper_kit.browser_fetcher._raise_for_sidecar_status", new=MagicMock()),
        patch.object(bf, "_report_fetch_result", new=MagicMock()),
    ):
        bf.request_context_reset()  # no-op при выключенном флаге
        await bf.fetch("https://ekaterinburg.domclick.ru/card/sale__flat__5")

    assert "reuse_context" not in captured
    assert "reset_context" not in captured