lekss361
f2cbd76ae0
All checks were successful
Deploy Trade-In / changes (push) Successful in 12s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 3m12s
Deploy Trade-In / build-backend (push) Successful in 1m1s
Deploy Trade-In / deploy (push) Successful in 1m27s
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""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)
|