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) → fallback env, не падаем;
  - флаг on + lease → fetch через lease.url, mark_health вызван, release в finally;
  - исключение внутри блока → mark_health(ok=False) + release всё равно (lease не течёт);
  - acquire кинул → fallback env (сбор не ломаем).
"""

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

from scraper_kit.contracts import ProxyLease
from scraper_kit.providers._proxy import curl_proxy_url

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


@dataclass
class _FakeConfig:
    use_proxy_pool_curl: bool = False
    scraper_proxy_url: str | None = None


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]] = []

    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))


_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_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 == []


# ── 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]