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

"""#2830: хвосты инцидента 2026-08-10 — пути, которые ещё ходили мимо пула прокси.

Корень (issue #2830, разбор владельца): `SCRAPER_PROXY_URL` — один статичный узел,
общий для всех источников и НЕ знающий про `scrape_proxy_source_bans`. Три места
продолжали брать egress именно оттуда:

  1. `cian_price_history` (ручка POST /admin/scrape/cian-price-history);
  2. `_provider_proxy_url` / GET /admin/scraper/health — ops-панель показывала статичный
     узел, пока трафик уже выбирался пулом (#2825/#2831);
  3. `resolve_cian_zhk_url_via_search` — вторая нога обогащения ЖК: #2767 перевёл на пул
     только `fetch_newbuilding`, резолв ЖК-url остался на `config.cian_proxy_url`.

Тесты проверяют ПОВЕДЕНИЕ пула (какой узел взят и какой вердикт вернулся), а не наличие
kwarg'а в вызове. Красные на старом коде:

  * (1) `fetch_detail` вызывался без `proxy_provider` → lease не брался, 403 никому не
    сообщался: `mark_banned_calls == []`. Ловушка «правка без эффекта» была в том, что
    `USE_PROXY_POOL_CURL` задавался только контейнеру `scraper`, а ручка живёт в
    `backend`, где флага не было — один `proxy_provider=` пул бы не включил. Отсюда взялся
    подкласс конфига с зашитым `use_proxy_pool_curl = True`; #3387 задал переменную и
    сервису `backend`, костыль снят (#3386 хвост), флаг снова управляет путём в обе
    стороны — см. `test_price_history_honours_flag_off`.
  * (2) `_provider_proxy_url(source)` возвращал `settings.scraper_proxy_url` и не имел
    параметра `db` — вызов из теста падал бы на сигнатуре, а исход «пул исчерпан»
    выражения не имел вообще.
  * (3) `resolve_cian_zhk_url_via_search` на 403 возвращал None: узел получал
    `mark_health(ok=True)` и оставался в выдаче Циану (механика #2700/#2821).
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

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

from scraper_kit.cian_exceptions import CianBlockedError
from scraper_kit.contracts import ProxyLease
from scraper_kit.providers.cian import detail as cian_detail
from scraper_kit.providers.cian import newbuilding as cian_newbuilding
from scraper_kit.proxy_errors import NoProxyAvailableError

# Узел 9 (asocks-mobile-1) — тот, что 2026-08-10 отдавал Циану 200, пока статичный
# SCRAPER_PROXY_URL смотрел на забаненный узел 1 (asocks-residential-1).
_LEASE = ProxyLease(id=9, url="http://user:pass@pool-node-9:10313", kind="http", rotate_url=None)


class _SpyProvider:
    """ProxyProvider-заглушка: запоминает вердикты, которые путь вернул пулу."""

    def __init__(self, lease: ProxyLease | None = _LEASE) -> None:
        self._lease = lease
        self.acquire_calls: list[str] = []
        self.mark_health_calls: list[tuple[int, bool]] = []
        self.mark_banned_calls: list[tuple[int, str]] = []
        self.release_calls: list[int] = []

    def acquire(self, provider: str) -> ProxyLease | None:
        self.acquire_calls.append(provider)
        return self._lease

    def release(self, lease: ProxyLease) -> None:
        self.release_calls.append(lease.id)

    def mark_health(
        self, lease: ProxyLease, ok: bool, *, exit_ip: Any = None, latency_ms: Any = None
    ) -> None:
        self.mark_health_calls.append((lease.id, ok))

    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:
        self.mark_banned_calls.append((lease.id, source))


class _SpyPool:
    """Заглушка МОДУЛЯ `app.services.proxy_pool` — подставляется ПОД `RealProxyProvider`.

    Так тест краснеет на поведении (пул не получил вердикт), а не на отсутствии нового
    имени в модуле: старый код проходит ровно тот же путь, просто ничего пулу не говорит.
    """

    def __init__(self, lease: Any = _LEASE) -> None:
        self._lease = lease
        self.acquire_calls: list[str] = []
        self.mark_health_calls: list[tuple[int, bool]] = []
        self.mark_banned_calls: list[tuple[int, str]] = []
        self.release_calls: list[int] = []

    def acquire(self, db: Any, provider: str, **kw: Any) -> Any:
        self.acquire_calls.append(provider)
        return self._lease

    def release(self, db: Any, lease_id: int) -> None:
        self.release_calls.append(lease_id)

    def mark_health(self, db: Any, lease_id: int, ok: bool, **kw: Any) -> None:
        self.mark_health_calls.append((lease_id, ok))

    def mark_banned(self, db: Any, lease_id: int, *, source: str) -> None:
        self.mark_banned_calls.append((lease_id, source))


def _session_returning(status_code: int, text: str = "") -> MagicMock:
    session = MagicMock()
    session.get = AsyncMock(return_value=MagicMock(status_code=status_code, text=text))
    session.close = AsyncMock()
    return session


# ── 1. cian_price_history: узел из пула + вердикт обратно ────────────────────


def _price_history_db(n_listings: int) -> MagicMock:
    """Фейковая сессия: SELECT листингов, дальше COUNT'ы (до fetch дело не дойдёт)."""
    rows = [
        {"id": 100 + i, "source_url": f"https://ekb.cian.ru/sale/flat/{100 + i}/"}
        for i in range(n_listings)
    ]
    listings_result = MagicMock()
    listings_result.mappings.return_value.all.return_value = rows
    db = MagicMock()
    db.execute.return_value = listings_result
    return db


async def _run_price_history(
    pool: _SpyPool, *, status_code: int, n: int = 1, use_pool: bool = True
) -> Any:
    from app.core.config import settings
    from app.services import scraper_adapters
    from app.services.cian_price_history import backfill_cian_price_history

    with (
        patch.object(
            cian_detail, "build_curl_cffi_session", return_value=_session_returning(status_code)
        ),
        patch("app.services.cian_price_history.get_scraper_delay", return_value=0.0),
        patch.object(settings, "use_proxy_pool_curl", use_pool),
        patch.object(scraper_adapters, "_proxy_pool", pool),
        patch.object(scraper_adapters, "_SessionLocal", MagicMock()),
    ):
        return await backfill_cian_price_history(_price_history_db(n), batch_size=n)


@pytest.mark.asyncio
async def test_price_history_takes_pool_node() -> None:
    """Узел берётся из пула, вердикт возвращается, lease не течёт.

    Красный на коде до #2830: не было `proxy_provider=` — `curl_proxy_url` уходил на
    статичный env-узел и `acquire` не звал.
    """
    pool = _SpyPool()
    await _run_price_history(pool, status_code=200)
    assert pool.acquire_calls == ["cian"]
    assert pool.release_calls == [9]  # lease не течёт


@pytest.mark.asyncio
async def test_price_history_honours_flag_off() -> None:
    """USE_PROXY_POOL_CURL=false → честный env-путь, а не пул через зашитую константу.

    #3386 (хвост): пока в сервисе жил подкласс `RealScraperConfig` с
    `use_proxy_pool_curl = True`, рубильник на этом пути был неотключаем — тест красный
    на main (`acquire_calls == ["cian"]`). После #3387 переменная задана контейнеру
    `backend` в compose, костыль лишний, и флаг снова управляет обеими сторонами.
    """
    pool = _SpyPool()
    result = await _run_price_history(pool, status_code=200, use_pool=False)
    assert pool.acquire_calls == [], "при выключенном флаге пул не трогаем"
    assert result.checked == 1, "запрос всё равно идёт — просто env-прокси, как раньше"


@pytest.mark.asyncio
async def test_price_history_403_bans_the_node_for_cian() -> None:
    """403 от Циана снимает узел с выдачи ИМЕННО Циану. Красный: было `mark_banned` = []."""
    pool = _SpyPool()
    result = await _run_price_history(pool, status_code=403)
    assert pool.mark_banned_calls == [(9, "cian")]
    assert pool.mark_health_calls == [(9, False)]
    assert result.errors == 1  # прогон честен: отказ посчитан


@pytest.mark.asyncio
async def test_price_history_404_does_not_ban_the_node() -> None:
    """Снятое объявление — не бан площадки: чужой вердикт узлу не присваиваем (#2805)."""
    pool = _SpyPool()
    await _run_price_history(pool, status_code=404)
    assert pool.mark_banned_calls == []
    assert pool.mark_health_calls == [(9, True)]


@pytest.mark.asyncio
async def test_price_history_aborts_batch_when_pool_empty() -> None:
    """Пул пуст в проде → отказ ДО HTTP и разрыв батча, а не 3 попытки подряд."""
    from app.core.config import settings

    pool = _SpyPool(lease=None)
    # Прод-признак читает `curl_proxy_url` через config.environment (#2616 шаг 1):
    # только в проде пустой пул = отказ, в dev это легитимный env/direct-путь.
    with patch.object(settings, "environment", "production"):
        result = await _run_price_history(pool, status_code=200, n=3)
    assert pool.acquire_calls == ["cian"], "после отказа пула батч продолжаться не должен"
    assert result.errors == 1
    assert result.checked == 3  # выбрали 3, но оборвались на первом


def test_no_proxy_available_is_not_a_platform_ban() -> None:
    """Наш отказ инфраструктуры не должен читаться как бан площадки (#2616)."""
    from scraper_kit.proxy_errors import ProxyBanError

    assert not issubclass(NoProxyAvailableError, ProxyBanError)


# ── 2. /admin/scraper/health: показывает узел, который реально получит трафик ──


def _health_db(*, candidate: dict[str, Any] | None, diag: dict[str, int]) -> MagicMock:
    """Сессия для `proxy_egress`: SELECT кандидата, при пустом — SELECT-разбивка причин."""

    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
        row = diag if "count(*)" in str(stmt).lower() else candidate
        res = MagicMock()
        res.mappings.return_value.fetchone.return_value = row
        return res

    db = MagicMock()
    db.execute.side_effect = _execute
    return db


def _get_health(db: MagicMock) -> dict[str, Any]:
    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

    def _fake_db() -> Any:
        yield db

    app = FastAPI()
    app.include_router(admin_module.router, prefix="/api/v1/admin")
    app.dependency_overrides[get_db] = _fake_db

    async def _no_browser() -> Any:
        return admin_module.BrowserHealth(reachable=False, browsers={})

    async def _no_ip(proxy_url: str | None) -> str | None:
        return None

    with (
        patch.object(admin_module, "_probe_browser_health", _no_browser),
        patch.object(admin_module, "_probe_current_ip", _no_ip),
        # Статичный env-узел ОТЛИЧАЕТСЯ от пулового — иначе тест не различит источники.
        patch.object(
            admin_module.settings.__class__,
            "scraper_proxy_url",
            property(lambda _self: "http://u:p@static-env-node:10423"),
        ),
    ):
        r = TestClient(app).get("/api/v1/admin/scraper/health")
    assert r.status_code == 200
    return {p["source"]: p for p in r.json()["providers"]}


def test_health_shows_pool_node_not_static_env() -> None:
    """Панель показывает узел, который получит трафик, а не SCRAPER_PROXY_URL.

    Красный на старом коде поведенчески: возвращался `static-env-node` — ровно то
    расхождение «панель показывает один узел, трафик идёт через другой», из-за
    которого инцидент 2026-08-10 месяц оставался невидимым.
    """
    db = _health_db(
        candidate={"id": 9, "url": _LEASE.url, "label": "asocks-mobile-1"},
        diag={"pool_total": 4, "banned_for_source": 0, "unhealthy_or_disabled": 0},
    )
    by_source = _get_health(db)
    assert {p["proxy_host"] for p in by_source.values()} == {"pool-node-9"}
    assert by_source["cian"]["proxy_port"] == 10313


def test_health_shows_nothing_when_pool_exhausted_for_source() -> None:
    """Все узлы забанены для источника → «—», а не статичный env-узел.

    Зелёная строка на месте отказа хуже пустой: она читается как «прокси есть».
    """
    db = _health_db(
        candidate=None,
        diag={"pool_total": 4, "banned_for_source": 4, "unhealthy_or_disabled": 0},
    )
    by_source = _get_health(db)
    assert by_source["cian"]["proxy_host"] is None
    assert by_source["cian"]["proxy_port"] is None


# ── 3. resolve ЖК-url: вторая нога обогащения (#2767 чинил только первую) ─────


@dataclass
class _FakeConfig:
    use_proxy_pool_curl: bool = True
    cian_proxy_url: str | None = "http://static-env-node:10423"
    environment: str = "production"


async def _resolve(status_code: int, spy: _SpyProvider, html: str = "") -> Any:
    with patch.object(
        cian_newbuilding,
        "build_curl_cffi_session",
        return_value=_session_returning(status_code, html),
    ):
        return await cian_newbuilding.resolve_cian_zhk_url_via_search(
            48853, config=_FakeConfig(), proxy_provider=spy
        )


@pytest.mark.asyncio
async def test_zhk_resolve_uses_pool_node() -> None:
    """Резолв ЖК-url берёт узел из пула, а не статичный cian_proxy_url."""
    spy = _SpyProvider()
    html = '<h1 data-name="Title"><a href="https://zhk-parkovyy-kvartal-ekb-i.cian.ru">ЖК</a></h1>'
    assert await _resolve(200, spy, html) == "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"
    assert spy.acquire_calls == ["cian"]
    assert spy.mark_health_calls == [(9, True)]
    assert spy.release_calls == [9]


@pytest.mark.asyncio
async def test_zhk_resolve_403_reaches_the_pool() -> None:
    """Красный на старом коде: 403 гасился в `return None`, узел оставался в выдаче."""
    spy = _SpyProvider()
    with pytest.raises(CianBlockedError):
        await _resolve(403, spy)
    assert spy.mark_banned_calls == [(9, "cian")]
    assert spy.mark_health_calls == [(9, False)]
    assert spy.release_calls == [9]


@pytest.mark.asyncio
async def test_zhk_resolve_404_is_a_soft_miss_not_a_ban() -> None:
    """404 cat.php — не бан: здоровый узел за отсутствующий ЖК не наказываем."""
    spy = _SpyProvider()
    assert await _resolve(404, spy) is None
    assert spy.mark_banned_calls == []
    assert spy.mark_health_calls == [(9, True)]


@pytest.mark.asyncio
async def test_zhk_resolve_empty_serp_is_a_soft_miss_not_a_ban() -> None:
    """200 без слага (дрейф вёрстки / пустая выдача) — тоже не повод банить узел."""
    spy = _SpyProvider()
    assert await _resolve(200, spy, "<html><body>ничего</body></html>") is None
    assert spy.mark_banned_calls == []
    assert spy.mark_health_calls == [(9, True)]