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

"""Тесты проброса browser-пула в SERP-скраперы yandex/cian (#2160).

Зеркало P4 (#2164) для avito: там пул раздавался только в avito-путях, а yandex/cian
SERP создавали BrowserFetcher сами в __aenter__ без пула → весь трафик шёл через
env-прокси браузера. Здесь проверяем, что YandexRealtyScraper / CianScraper прокидывают
proxy_provider + use_pool=config.use_proxy_pool_browser в BrowserFetcher.

Инвариант ship-dark:
  - proxy_provider=None ИЛИ флаг off → BrowserFetcher(..., proxy_provider=None, use_pool=False),
    т.е. env-прокси браузера, поведение байт-в-байт как до фикса.
  - proxy_provider задан + флаг on → BrowserFetcher получает провайдер + use_pool=True.

BrowserFetcher монки-патчится на recording-заглушку в модуле каждого скрапера, так что
реального HTTP к tradein-browser не происходит.

Yandex (#2363, F4c миграция): `yandex/serp.py::__aenter__` больше не конструирует
`BrowserFetcher(...)` напрямую — вызывает shared `build_browser_fetcher()` из
`scraper_kit.providers._base`, который ВНУТРИ себя ссылается на СВОЙ собственный
импорт `BrowserFetcher`. Патч локального имени `yandex_serp.BrowserFetcher` больше
не перехватывает вызов (реальный `BrowserFetcher.__init__` конструировался бы, а
`calls` оставался пуст) — патчим `scraper_kit.providers._base.BrowserFetcher`
(реальную зависимость `build_browser_fetcher`), сохраняя все assertions как есть.
Cian (`cian/serp.py`) пока НЕ мигрирован (F4b, отдельный issue) — там патч
локального имени остаётся корректным.
"""

from __future__ import annotations

import types
from typing import Any

import pytest
from scraper_kit.providers.cian import serp as cian_serp
from scraper_kit.providers.yandex import serp as yandex_serp


class _FakeProxyProvider:
    """ProxyProvider-заглушка: методы no-op (в этих тестах пул не вызывается —
    BrowserFetcher замокан, acquire/release не срабатывают)."""

    def acquire(self, provider: str) -> None:
        return None

    def release(self, lease: object) -> None:
        return None

    def mark_health(self, lease: object, ok: bool) -> None:
        return None


def _make_recorder() -> tuple[type, list[dict[str, Any]]]:
    """Recording-заглушка BrowserFetcher: пишет init-kwargs в общий список."""
    calls: list[dict[str, Any]] = []

    class _RecFetcher:
        def __init__(
            self,
            source: str = "avito",
            fetch_timeout_s: float = 120.0,
            *,
            endpoint: str,
            proxy_provider: object | None = None,
            use_pool: bool = False,
        ) -> None:
            calls.append(
                {
                    "source": source,
                    "endpoint": endpoint,
                    "proxy_provider": proxy_provider,
                    "use_pool": use_pool,
                }
            )

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

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

    return _RecFetcher, calls


def _config(*, use_pool: bool) -> types.SimpleNamespace:
    return types.SimpleNamespace(
        browser_http_endpoint="http://browser:3000",
        use_proxy_pool_browser=use_pool,
    )


# ── Yandex SERP ──────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_yandex_serp_wires_pool_when_flag_on(monkeypatch: pytest.MonkeyPatch) -> None:
    """proxy_provider + флаг on → BrowserFetcher получает провайдер + use_pool=True."""
    rec_fetcher, calls = _make_recorder()
    monkeypatch.setattr("scraper_kit.providers._base.BrowserFetcher", rec_fetcher)
    provider = _FakeProxyProvider()

    scraper = yandex_serp.YandexRealtyScraper(_config(use_pool=True), proxy_provider=provider)
    async with scraper:
        pass

    assert len(calls) == 1
    assert calls[0]["source"] == "yandex"
    assert calls[0]["proxy_provider"] is provider
    assert calls[0]["use_pool"] is True


@pytest.mark.asyncio
async def test_yandex_serp_no_pool_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
    """Без provider/флага → BrowserFetcher(proxy_provider=None, use_pool=False) — ship-dark."""
    rec_fetcher, calls = _make_recorder()
    monkeypatch.setattr("scraper_kit.providers._base.BrowserFetcher", rec_fetcher)

    scraper = yandex_serp.YandexRealtyScraper(_config(use_pool=False))
    async with scraper:
        pass

    assert len(calls) == 1
    assert calls[0]["proxy_provider"] is None
    assert calls[0]["use_pool"] is False


@pytest.mark.asyncio
async def test_yandex_serp_provider_without_flag_stays_dark(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """provider задан, но флаг off → use_pool=False (BrowserFetcher пул не трогает)."""
    rec_fetcher, calls = _make_recorder()
    monkeypatch.setattr("scraper_kit.providers._base.BrowserFetcher", rec_fetcher)
    provider = _FakeProxyProvider()

    scraper = yandex_serp.YandexRealtyScraper(_config(use_pool=False), proxy_provider=provider)
    async with scraper:
        pass

    assert calls[0]["proxy_provider"] is provider
    assert calls[0]["use_pool"] is False


# ── Cian SERP ────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_cian_serp_wires_pool_when_flag_on(monkeypatch: pytest.MonkeyPatch) -> None:
    """proxy_provider + флаг on → BrowserFetcher получает провайдер + use_pool=True."""
    rec_fetcher, calls = _make_recorder()
    monkeypatch.setattr(cian_serp, "BrowserFetcher", rec_fetcher)
    provider = _FakeProxyProvider()

    scraper = cian_serp.CianScraper(_config(use_pool=True), proxy_provider=provider)
    async with scraper:
        pass

    assert len(calls) == 1
    assert calls[0]["source"] == "cian"
    assert calls[0]["proxy_provider"] is provider
    assert calls[0]["use_pool"] is True


@pytest.mark.asyncio
async def test_cian_serp_no_pool_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
    """Без provider/флага → BrowserFetcher(proxy_provider=None, use_pool=False) — ship-dark."""
    rec_fetcher, calls = _make_recorder()
    monkeypatch.setattr(cian_serp, "BrowserFetcher", rec_fetcher)

    scraper = cian_serp.CianScraper(_config(use_pool=False))
    async with scraper:
        pass

    assert len(calls) == 1
    assert calls[0]["proxy_provider"] is None
    assert calls[0]["use_pool"] is False