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

"""P3 (#2163): kit curl-пути берут прокси из пула за флагом USE_PROXY_POOL_CURL.

Покрывает инвариант ship-dark + fallback на уровне helper'а `curl_proxy_url` и
class-based провайдера (YandexValuationScraper):
  - флаг off / proxy_provider=None → env-прокси, пул не трогается (golden-parity);
  - флаг on + пул пуст (acquire→None) + dev → fallback env, не падаем;
  - флаг on + пул пуст (acquire→None) + prod (#2616 шаг 1) → NoProxyAvailableError,
    HTTP-запрос НЕ выполняется, на env НЕ идём;
  - флаг on + lease → fetch через lease.url, mark_health вызван, release в finally
    (в prod и dev одинаково — пул выдал лизу, отказа быть не должно);
  - исключение внутри блока → mark_health(ok=False) + release всё равно (lease не течёт);
  - отмена/прерывание (CancelledError/KeyboardInterrupt/SystemExit — BaseException, не
    Exception) внутри блока → тот же ok=False, исключение пролетает наружу;
  - acquire кинул → fallback env (dev) / NoProxyAvailableError (prod).
"""

from __future__ import annotations

import asyncio
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/test_db")

from scraper_kit.contracts import ProxyLease
from scraper_kit.providers._proxy import curl_proxy_url
from scraper_kit.proxy_errors import NoProxyAvailableError, ProxyBanError

# ── Фейки ─────────────────────────────────────────────────────────────────────


@dataclass
class _FakeConfig:
    use_proxy_pool_curl: bool = False
    scraper_proxy_url: str | None = None
    # #2616 шаг 1: дефолт "dev" — существующие тесты (не задающие поле явно) не
    # затрагиваются новым prod-отказом, ведут себя ровно как до этого изменения.
    environment: str = "dev"


class _SpyProvider:
    """ProxyProvider-заглушка, записывающая вызовы."""

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

    def acquire(self, provider: str) -> ProxyLease | None:
        self.acquire_calls.append(provider)
        if self._acquire_raises:
            raise RuntimeError("boom acquire")
        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))


_LEASE = ProxyLease(id=7, url="http://user:pass@pool-proxy:3128", kind="http", rotate_url=None)


# ── curl_proxy_url: helper-инвариант ──────────────────────────────────────────


def test_flag_off_yields_env_and_never_touches_pool() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=False)
    spy = _SpyProvider(_LEASE)
    with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url:
        assert url == "http://env:3128"
    # golden-parity: пул не задействован при выключенном флаге
    assert spy.acquire_calls == []
    assert spy.mark_health_calls == []
    assert spy.release_calls == []


def test_provider_none_yields_env() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    with curl_proxy_url(cfg, None, "cian", env_fallback_url="http://env:3128") as url:
        assert url == "http://env:3128"


def test_flag_on_empty_pool_falls_back_to_env() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(None)  # acquire → None (пул пуст/выключен)
    with curl_proxy_url(cfg, spy, "avito", env_fallback_url="http://env:3128") as url:
        assert url == "http://env:3128"
    assert spy.acquire_calls == ["avito"]
    # lease не выдан → mark_health/release не зовём
    assert spy.mark_health_calls == []
    assert spy.release_calls == []


def test_flag_on_lease_used_and_released_on_success() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE)
    with curl_proxy_url(cfg, spy, "yandex", env_fallback_url="http://env:3128") as url:
        assert url == _LEASE.url  # fetch идёт через lease.url, НЕ env
    assert spy.acquire_calls == ["yandex"]
    assert spy.mark_health_calls == [(7, True)]
    assert spy.release_calls == [7]  # release в finally


def test_flag_on_exception_marks_fail_and_still_releases() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE)
    with pytest.raises(RuntimeError, match="ban"):
        with curl_proxy_url(cfg, spy, "cian", env_fallback_url=None) as url:
            assert url == _LEASE.url
            raise RuntimeError("ban 403")
    # исключение → ok=False, но lease ОБЯЗАТЕЛЬНО освобождён (не течёт)
    assert spy.mark_health_calls == [(7, False)]
    assert spy.release_calls == [7]


def test_ban_exception_calls_mark_banned_in_addition_to_mark_health() -> None:
    """Исключение — подкласс ProxyBanError (напр. AvitoBlockedError) внутри блока —
    вызывает mark_banned(lease, source=provider) В ДОПОЛНЕНИЕ к mark_health(ok=False)
    (#2600 п.1 curl-путь). Zero изменений в вызывающем коде — сигнал детектируется
    по ТИПУ исключения, а не явным вызовом."""

    class _FakeBlockedError(ProxyBanError):
        pass

    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE)
    with pytest.raises(_FakeBlockedError):
        with curl_proxy_url(cfg, spy, "avito", env_fallback_url=None) as url:
            assert url == _LEASE.url
            raise _FakeBlockedError("firewall page detected")
    assert spy.mark_banned_calls == [(7, "avito")]
    assert spy.mark_health_calls == [(7, False)]  # оба сигнала, не взаимоисключающие
    assert spy.release_calls == [7]  # lease всё равно освобождён


def test_plain_exception_does_not_call_mark_banned() -> None:
    """Обычная (не-ProxyBanError) ошибка — сетевой сбой/таймаут — идёт ТОЛЬКО через
    mark_health(ok=False), mark_banned НЕ вызывается (issue #2600 п.4 — различимость
    «бан площадки» vs «сетевой сбой»)."""
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE)
    with pytest.raises(TimeoutError):
        with curl_proxy_url(cfg, spy, "cian", env_fallback_url=None) as url:
            assert url == _LEASE.url
            raise TimeoutError("connect timed out")
    assert spy.mark_banned_calls == []
    assert spy.mark_health_calls == [(7, False)]
    assert spy.release_calls == [7]


@pytest.mark.parametrize(
    "exc_type", [asyncio.CancelledError, KeyboardInterrupt, SystemExit], ids=lambda t: t.__name__
)
def test_base_exception_marks_fail_and_still_releases(exc_type: type[BaseException]) -> None:
    """Отмена/прерывание внутри блока — тоже ok=False, а не «узел здоров».

    `asyncio.CancelledError` наследует BaseException, а не Exception: при таймауте
    (`asyncio.wait_for` в `_with_budget` эстиматора) отмена приходит внутрь await'а
    ВНУТРИ блока. С `except Exception` она пролетала мимо, `ok` оставался True и
    в finally узел, из-за которого запрос завис до отмены, записывался ЗДОРОВЫМ.
    KeyboardInterrupt/SystemExit — тот же путь: health честный, но исключение
    обязано пролететь наружу (`raise` в блоке сохранён, не проглатываем).
    """
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE)
    with pytest.raises(exc_type):
        with curl_proxy_url(cfg, spy, "avito", env_fallback_url=None) as url:
            assert url == _LEASE.url
            raise exc_type()
    assert spy.mark_health_calls == [(7, False)]
    assert spy.release_calls == [7]  # ровно один раз, lease не течёт
    assert spy.mark_banned_calls == []  # отмена ≠ бан площадки


def test_acquire_raises_falls_back_to_env() -> None:
    cfg = _FakeConfig(use_proxy_pool_curl=True)
    spy = _SpyProvider(_LEASE, acquire_raises=True)
    with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url:
        assert url == "http://env:3128"  # ошибка acquire → env, не падаем
    assert spy.release_calls == []


# ── #2616 шаг 1: пул пуст в prod → отказ, НЕ мёртвый env-фолбэк ────────────────


def test_flag_on_empty_pool_dev_falls_back_to_env() -> None:
    """Пул пуст + явный dev-признак → прежнее поведение (env-фолбэк, не падаем)."""
    cfg = _FakeConfig(use_proxy_pool_curl=True, environment="dev")
    spy = _SpyProvider(None)
    with curl_proxy_url(cfg, spy, "avito", env_fallback_url="http://env:3128") as url:
        assert url == "http://env:3128"
    assert spy.acquire_calls == ["avito"]
    assert spy.mark_health_calls == []
    assert spy.release_calls == []


def test_flag_on_empty_pool_prod_refuses_no_http_request() -> None:
    """Пул пуст + прод-признак → NoProxyAvailableError, HTTP-запрос НЕ выполняется.

    Заглушка `_boom` падает на ЛЮБОМ вызове внутри `with`-блока (там, где в
    реальном коде было бы `session.get(...)`). Если бы curl_proxy_url тихо
    fallback'нулся на env (регрессия), `_boom()` выполнился бы и поднял
    AssertionError, который `pytest.raises(NoProxyAvailableError)` НЕ поймает —
    тест упал бы с несовпадающим типом исключения (falsifiable).
    """
    cfg = _FakeConfig(use_proxy_pool_curl=True, environment="production")
    spy = _SpyProvider(None)  # acquire → None (пул пуст)

    def _boom() -> None:
        raise AssertionError("HTTP request must NOT happen — proxy pool empty in prod")

    with pytest.raises(NoProxyAvailableError):
        with curl_proxy_url(cfg, spy, "avito", env_fallback_url="http://env:3128"):
            _boom()  # НЕ должно достигаться — raise происходит ДО yield

    assert spy.acquire_calls == ["avito"]
    # lease не выдан → mark_health/release не зовём (нечего освобождать)
    assert spy.mark_health_calls == []
    assert spy.release_calls == []


def test_acquire_raises_prod_refuses_no_env_fallback() -> None:
    """acquire() упал + прод-признак → NoProxyAvailableError, не мёртвый env."""
    cfg = _FakeConfig(use_proxy_pool_curl=True, environment="production")
    spy = _SpyProvider(_LEASE, acquire_raises=True)
    with pytest.raises(NoProxyAvailableError):
        with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128"):
            pytest.fail("must not enter with-block body")
    assert spy.release_calls == []


def test_flag_on_lease_prod_unaffected() -> None:
    """Пул выдал прокси в прод — поведение БЕЗ ИЗМЕНЕНИЙ (не наш случай отказа)."""
    cfg = _FakeConfig(use_proxy_pool_curl=True, environment="production")
    spy = _SpyProvider(_LEASE)
    with curl_proxy_url(cfg, spy, "yandex", env_fallback_url="http://env:3128") as url:
        assert url == _LEASE.url
    assert spy.acquire_calls == ["yandex"]
    assert spy.mark_health_calls == [(7, True)]
    assert spy.release_calls == [7]


def test_no_proxy_error_distinguishable_from_site_block() -> None:
    """Причина отказа ("нет прокси", наша инфраструктура) programmatically отличима
    от блокировки площадкой: отдельный exception-тип (не AvitoBlockedError/generic),
    структурный provider-атрибут, текст без "blocked"/"captcha"/"banned".
    """
    cfg = _FakeConfig(use_proxy_pool_curl=True, environment="production")
    spy = _SpyProvider(None)
    with pytest.raises(NoProxyAvailableError) as exc_info:
        with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128"):
            pass
    err = exc_info.value
    assert err.provider == "cian"
    assert not isinstance(err, LookupError)  # не путается с "не найдено"-семантикой
    lowered = str(err).lower()
    assert "blocked" not in lowered
    assert "captcha" not in lowered
    assert "banned" not in lowered
    assert "no proxy available" in lowered


# ── YandexValuationScraper: lease держится на всё время сессии ─────────────────


@pytest.mark.asyncio
async def test_yandex_scraper_flag_off_no_pool() -> None:
    from scraper_kit.providers.yandex.valuation import YandexValuationScraper

    cfg = _FakeConfig(use_proxy_pool_curl=False, scraper_proxy_url="http://env:3128")
    spy = _SpyProvider(_LEASE)
    with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
        sess_cls.return_value = MagicMock(close=AsyncMock())
        async with YandexValuationScraper(cfg, proxy_provider=spy):
            pass
        # флаг off → env-прокси, пул не тронут
        _, kwargs = sess_cls.call_args
        assert kwargs["proxies"] == {"http": "http://env:3128", "https": "http://env:3128"}
    assert spy.acquire_calls == []


@pytest.mark.asyncio
async def test_yandex_scraper_flag_on_lease_lifecycle() -> None:
    from scraper_kit.providers.yandex.valuation import YandexValuationScraper

    cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url="http://env:3128")
    spy = _SpyProvider(_LEASE)
    with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
        sess_cls.return_value = MagicMock(close=AsyncMock())
        async with YandexValuationScraper(cfg, proxy_provider=spy):
            # сессия создана через lease.url, lease ещё удерживается
            _, kwargs = sess_cls.call_args
            assert kwargs["proxies"] == {"http": _LEASE.url, "https": _LEASE.url}
            assert spy.release_calls == []
    # выход из сессии → mark_health(ok) + release
    assert spy.acquire_calls == ["yandex"]
    assert spy.mark_health_calls == [(7, True)]
    assert spy.release_calls == [7]


@pytest.mark.asyncio
async def test_yandex_scraper_exception_releases_lease() -> None:
    from scraper_kit.providers.yandex.valuation import YandexValuationScraper

    cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url=None)
    spy = _SpyProvider(_LEASE)
    with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
        sess_cls.return_value = MagicMock(close=AsyncMock())
        with pytest.raises(RuntimeError, match="scrape blew up"):
            async with YandexValuationScraper(cfg, proxy_provider=spy):
                raise RuntimeError("scrape blew up")
    # исключение в теле → ok=False + release (lease не течёт)
    assert spy.mark_health_calls == [(7, False)]
    assert spy.release_calls == [7]