"""Offline-тесты резолвера egress-прокси по источнику (#2825, fail-closed #2616). Покрытие БЕЗ live-сети/БД: FakeSession эмулирует ДВА запроса над scrape_proxies + scrape_proxy_source_bans — основной SELECT кандидата (`_pick_candidate`) и, только когда он вернул пусто, diagnostic-агрегат (`_diagnose_no_candidate`) для различения "пул пуст" от "пул не пуст, все отсеяны". - выбирается небанненный прокси; - забаненный ДЛЯ ИСТОЧНИКА не выбирается; - забаненный для ДРУГОГО источника — выбирается (суть #2600 п.2: Авито банит IP, Яндекс через тот же IP ходит чисто); - при нескольких кандидатах — меньший consecutive_fails выигрывает; - при равном consecutive_fails — более свежий last_ok_at выигрывает; - пул ПУСТ (0 строк вообще) → легитимный fallback на settings.scraper_proxy_url, logger.WARNING с текстом «пуст»; - пул пуст И SCRAPER_PROXY_URL не задан → None (прямое подключение), WARNING; - пул НЕ пуст, но все кандидаты забанены/нездоровы/выключены → ProxyPoolExhaustedError (fail-closed, #2616), logger.ERROR с разбивкой — env НЕ используется, даже если задан; - тексты "пуст" и "все отсеяны" в логах РАЗНЫЕ (не перепутать при чтении логов/алертов). """ from __future__ import annotations import os os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") from datetime import UTC, datetime, timedelta from typing import Any import pytest from app.core.config import settings from app.services.proxy_egress import ProxyPoolExhaustedError, resolve_proxy_url # ── stateful fake session (эмулирует ОБА read-only запроса proxy_egress) ────────── class _FakeResult: def __init__(self, rows: list[dict[str, Any]]): self._rows = rows def mappings(self) -> _FakeResult: return self def fetchone(self) -> dict[str, Any] | None: return self._rows[0] if self._rows else None class FakeSession: def __init__(self, rows: list[dict[str, Any]], bans: list[dict[str, Any]] | None = None): self.rows = rows self.bans = bans or [] def _has_active_ban(self, pid: int, source: str) -> bool: return any( b["proxy_id"] == pid and b["source"] == source and b["banned_until"] > datetime.now(UTC) for b in self.bans ) def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: sql = str(stmt) p = params or {} assert "FROM scrape_proxies" in sql assert "scrape_proxy_source_bans" in sql max_fails = p["max_fails"] source = p["source"] if "pool_total" in sql: # _diagnose_no_candidate aggregate unhealthy = sum( 1 for r in self.rows if not r["enabled"] or r["consecutive_fails"] >= max_fails ) banned = sum( 1 for r in self.rows if r["enabled"] and r["consecutive_fails"] < max_fails and self._has_active_ban(r["id"], source) ) return _FakeResult( [ { "pool_total": len(self.rows), "unhealthy_or_disabled": unhealthy, "banned_for_source": banned, } ] ) # _pick_candidate primary SELECT cands = [ r for r in self.rows if r["enabled"] and r["consecutive_fails"] < max_fails and not self._has_active_ban(r["id"], source) ] cands.sort( key=lambda r: ( r["consecutive_fails"], -(r["last_ok_at"] or datetime.min.replace(tzinfo=UTC)).timestamp(), r["id"], ) ) return _FakeResult([dict(r) for r in cands[:1]]) def _proxy( id_: int, *, enabled: bool = True, consecutive_fails: int = 0, last_ok_at: datetime | None = None, label: str | None = None, url: str = "", ) -> dict[str, Any]: return { "id": id_, "url": url or f"http://user:pass@proxy{id_}.local:8080", "label": label, "enabled": enabled, "consecutive_fails": consecutive_fails, "last_ok_at": last_ok_at, } @pytest.fixture(autouse=True) def _clear_fallback_env(monkeypatch: pytest.MonkeyPatch) -> None: # Изолируем тесты от реального прод-значения ENV (если случайно унаследовано). monkeypatch.setattr(settings, "scraper_proxy_url_env", None) def test_picks_healthy_unbanned_proxy() -> None: db = FakeSession([_proxy(1, url="http://u:p@good.local:8080")]) result = resolve_proxy_url(db, "avito") assert result == "http://u:p@good.local:8080" def test_banned_for_source_raises_pool_exhausted() -> None: """Пул НЕ пуст (1 узел), но он забанен для ИМЕННО этого источника — fail-closed, НЕ fallback на env (#2616).""" now = datetime.now(UTC) db = FakeSession( [_proxy(1, url="http://u:p@banned.local:8080")], bans=[{"proxy_id": 1, "source": "avito", "banned_until": now + timedelta(hours=6)}], ) with pytest.raises(ProxyPoolExhaustedError) as exc_info: resolve_proxy_url(db, "avito") assert exc_info.value.source == "avito" assert exc_info.value.pool_total == 1 assert exc_info.value.banned_for_source == 1 assert exc_info.value.unhealthy_or_disabled == 0 def test_banned_for_other_source_still_picked() -> None: now = datetime.now(UTC) db = FakeSession( [_proxy(1, url="http://u:p@shared.local:8080")], bans=[{"proxy_id": 1, "source": "cian", "banned_until": now + timedelta(hours=6)}], ) # Забанен только для cian — для avito остаётся первосортным кандидатом. result = resolve_proxy_url(db, "avito") assert result == "http://u:p@shared.local:8080" def test_expired_ban_does_not_block() -> None: now = datetime.now(UTC) db = FakeSession( [_proxy(1, url="http://u:p@revived.local:8080")], bans=[{"proxy_id": 1, "source": "avito", "banned_until": now - timedelta(hours=1)}], ) result = resolve_proxy_url(db, "avito") assert result == "http://u:p@revived.local:8080" def test_tiebreak_lower_consecutive_fails_wins() -> None: db = FakeSession( [ _proxy(1, consecutive_fails=2, url="http://u:p@flaky.local:8080"), _proxy(2, consecutive_fails=0, url="http://u:p@solid.local:8080"), ] ) result = resolve_proxy_url(db, "yandex") assert result == "http://u:p@solid.local:8080" def test_tiebreak_fresher_last_ok_at_wins_on_equal_fails() -> None: now = datetime.now(UTC) db = FakeSession( [ _proxy( 1, consecutive_fails=0, last_ok_at=now - timedelta(hours=2), url="http://u:p@stale.local:8080", ), _proxy( 2, consecutive_fails=0, last_ok_at=now - timedelta(minutes=5), url="http://u:p@fresh.local:8080", ), ] ) result = resolve_proxy_url(db, "yandex") assert result == "http://u:p@fresh.local:8080" def test_empty_pool_falls_back_to_env_with_warning( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: """Сценарий 1 (легитимный): 0 строк в scrape_proxies вообще — dev/staging без БД-пула. fallback на env разрешён.""" monkeypatch.setattr(settings, "scraper_proxy_url_env", "http://static-fallback.local:9999") db = FakeSession([]) with caplog.at_level("WARNING"): result = resolve_proxy_url(db, "cian") assert result == "http://static-fallback.local:9999" warnings = [rec for rec in caplog.records if rec.levelname == "WARNING"] assert any("пуст" in rec.message.lower() for rec in warnings) assert not any(rec.levelname == "ERROR" for rec in caplog.records) def test_empty_pool_and_no_env_returns_none_with_warning( caplog: pytest.LogCaptureFixture, ) -> None: db = FakeSession([]) with caplog.at_level("WARNING"): result = resolve_proxy_url(db, "domclick") assert result is None assert any("прямым подключением" in rec.message for rec in caplog.records) def test_all_candidates_banned_raises_pool_exhausted_not_env_fallback( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: """Сценарий 2 (инцидент 2026-08-10): пул НЕ пуст (2 узла), оба забанены для source — fail-closed. env ЗАДАН, но НЕ используется — это и есть сама суть фикса.""" monkeypatch.setattr(settings, "scraper_proxy_url_env", "http://static-fallback.local:9999") now = datetime.now(UTC) db = FakeSession( [_proxy(1), _proxy(2)], bans=[ {"proxy_id": 1, "source": "cian", "banned_until": now + timedelta(hours=6)}, {"proxy_id": 2, "source": "cian", "banned_until": now + timedelta(hours=6)}, ], ) with caplog.at_level("WARNING"): with pytest.raises(ProxyPoolExhaustedError) as exc_info: resolve_proxy_url(db, "cian") assert exc_info.value.pool_total == 2 assert exc_info.value.banned_for_source == 2 assert exc_info.value.unhealthy_or_disabled == 0 errors = [rec for rec in caplog.records if rec.levelname == "ERROR"] assert any("fail-closed" in rec.message.lower() for rec in errors) # НЕ должно быть "обход пула" / "static-fallback" в логах — env не тронут. assert not any("static-fallback" in rec.message for rec in caplog.records) def test_exhausted_and_empty_pool_log_texts_are_distinct( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: """Регрессия на замечание ревью: "пуст" и "все отсеяны" — РАЗНЫЕ формулировки И разные уровни (WARNING vs ERROR), иначе их нельзя различить в логах/алертах.""" now = datetime.now(UTC) with caplog.at_level("WARNING"): resolve_proxy_url(FakeSession([]), "avito") empty_pool_messages = {rec.levelname: rec.message for rec in caplog.records} caplog.clear() with caplog.at_level("WARNING"): with pytest.raises(ProxyPoolExhaustedError): resolve_proxy_url( FakeSession( [_proxy(1)], bans=[ {"proxy_id": 1, "source": "avito", "banned_until": now + timedelta(hours=6)} ], ), "avito", ) exhausted_messages = {rec.levelname: rec.message for rec in caplog.records} assert "ERROR" not in empty_pool_messages assert "ERROR" in exhausted_messages assert empty_pool_messages.get("WARNING") != exhausted_messages.get("ERROR") def test_disabled_proxy_raises_pool_exhausted() -> None: db = FakeSession([_proxy(1, enabled=False)]) with pytest.raises(ProxyPoolExhaustedError) as exc_info: resolve_proxy_url(db, "avito") assert exc_info.value.pool_total == 1 assert exc_info.value.unhealthy_or_disabled == 1 assert exc_info.value.banned_for_source == 0 def test_unhealthy_proxy_raises_pool_exhausted() -> None: from app.services.proxy_pool import MAX_CONSECUTIVE_FAILS db = FakeSession([_proxy(1, consecutive_fails=MAX_CONSECUTIVE_FAILS)]) with pytest.raises(ProxyPoolExhaustedError) as exc_info: resolve_proxy_url(db, "avito") assert exc_info.value.unhealthy_or_disabled == 1 class _RaisingSession: """db, у которой execute() всегда роняет (DB недоступна) — резолвер не может подтвердить exhaustion, лечит это КАК пустой пул (см. resolve_proxy_url docstring).""" def __init__(self) -> None: self.rollback_called = False def execute(self, *args: Any, **kwargs: Any) -> Any: raise RuntimeError("connection refused") def rollback(self) -> None: self.rollback_called = True def test_db_error_treated_as_empty_pool_falls_back_with_warning( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: monkeypatch.setattr(settings, "scraper_proxy_url_env", "http://static-fallback.local:9999") db = _RaisingSession() with caplog.at_level("WARNING"): result = resolve_proxy_url(db, "avito") # type: ignore[arg-type] assert result == "http://static-fallback.local:9999" assert db.rollback_called assert not any(rec.levelname == "ERROR" for rec in caplog.records)