"""Offline-тесты резолвера egress-прокси по источнику (#2825, fail-closed #2616,
ban-history ранжирование доп. #2825 от 2026-08-13).

Покрытие БЕЗ live-сети/БД: FakeSession эмулирует ДВА запроса над scrape_proxies +
scrape_proxy_source_bans — основной SELECT кандидата (`_pick_candidate`) и, только
когда он вернул пусто, diagnostic-агрегат (`_diagnose_no_candidate`) для различения
"пул пуст" от "пул не пуст, все отсеяны".

  - выбирается небанненный прокси;
  - забаненный ДЛЯ ИСТОЧНИКА (АКТИВНО, banned_until > now()) не выбирается;
  - забаненный для ДРУГОГО источника — выбирается (суть #2600 п.2: Авито банит IP,
    Яндекс через тот же IP ходит чисто) — включая случай, когда у него накопилась
    ИСТОРИЯ банов по другому source: на ранжирование ДЛЯ ТЕКУЩЕГО source это не влияет;
  - узел с историей банов (даже истёкшей) по ЭТОМУ source уступает чистому узлу без
    истории, даже когда у чистого узла хуже consecutive_fails/last_ok_at;
  - при равной истории (ban_count) — работает прежний tie-break: меньший
    consecutive_fails, затем более свежий last_ok_at;
  - активный бан (banned_until > now()) по-прежнему полностью исключает узел, вне
    зависимости от ban_count;
  - пул ПУСТ (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 _ban_count(self, pid: int, source: str) -> int:
        """COALESCE(b.ban_count, 0) семантика LEFT JOIN — история учитывается ДАЖЕ
        если сама строка бана уже истекла (banned_until <= now(), ещё не спурженная)."""
        for b in self.bans:
            if b["proxy_id"] == pid and b["source"] == source:
                return int(b.get("ban_count", 1))
        return 0

    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 — LEFT JOIN на bans по source: активный бан
        # по-прежнему исключает узел (WHERE), а ban_count (в т.ч. от истёкшего бана)
        # ранжирует прошедших фильтр: сначала без истории (0), затем по возрастанию.
        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: (
                self._ban_count(r["id"], source),
                r["consecutive_fails"],
                -(r["last_ok_at"] or datetime.min.replace(tzinfo=UTC)).timestamp(),
                r["id"],
            )
        )
        return _FakeResult(
            [{**r, "ban_count": self._ban_count(r["id"], source)} 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_clean_node_beats_node_with_expired_ban_history_for_source() -> None:
    """Замер на проде 2026-08-10: asocks-residential-1 отдавал 403 и cian, и avito, но
    после истечения TTL всплывал первым, потому что consecutive_fails=0 у ОБОИХ узлов
    и решал только свежий healthcheck. История (ban_count) ДОЛЖНА перевешивать даже
    когда у банившегося узла лучше consecutive_fails/last_ok_at."""
    now = datetime.now(UTC)
    db = FakeSession(
        [
            _proxy(
                1,
                consecutive_fails=0,
                last_ok_at=now,  # свежее всех — раньше выиграл бы по старому правилу
                url="http://u:p@chronic.local:8080",
            ),
            _proxy(
                2,
                consecutive_fails=1,
                last_ok_at=now - timedelta(hours=3),
                url="http://u:p@clean.local:8080",
            ),
        ],
        bans=[
            {
                "proxy_id": 1,
                "source": "avito",
                "banned_until": now - timedelta(hours=1),  # ИСТЁК, но ban_count остаётся
                "ban_count": 4,
            }
        ],
    )
    result = resolve_proxy_url(db, "avito")
    assert result == "http://u:p@clean.local:8080"


def test_no_history_node_beats_node_with_ban_count_one() -> None:
    """Узел БЕЗ ЕДИНОЙ строки истории (ban_count трактуется как 0) выигрывает у узла с
    ban_count=1, даже при равном consecutive_fails/last_ok_at."""
    now = datetime.now(UTC)
    db = FakeSession(
        [
            _proxy(1, consecutive_fails=0, last_ok_at=now, url="http://u:p@once-banned.local:8080"),
            _proxy(
                2, consecutive_fails=0, last_ok_at=now, url="http://u:p@never-banned.local:8080"
            ),
        ],
        bans=[
            {
                "proxy_id": 1,
                "source": "cian",
                "banned_until": now - timedelta(hours=2),
                "ban_count": 1,
            }
        ],
    )
    result = resolve_proxy_url(db, "cian")
    assert result == "http://u:p@never-banned.local:8080"


def test_equal_ban_history_falls_back_to_prior_tiebreak() -> None:
    """При РАВНОМ ban_count у обоих узлов -- прежний порядок tie-break (consecutive_fails,
    затем last_ok_at) без изменений."""
    now = datetime.now(UTC)
    db = FakeSession(
        [
            _proxy(
                1,
                consecutive_fails=2,
                last_ok_at=now,
                url="http://u:p@flaky-history.local:8080",
            ),
            _proxy(
                2,
                consecutive_fails=0,
                last_ok_at=now - timedelta(hours=1),
                url="http://u:p@solid-history.local:8080",
            ),
        ],
        bans=[
            {
                "proxy_id": 1,
                "source": "yandex",
                "banned_until": now - timedelta(hours=5),
                "ban_count": 2,
            },
            {
                "proxy_id": 2,
                "source": "yandex",
                "banned_until": now - timedelta(hours=5),
                "ban_count": 2,
            },
        ],
    )
    result = resolve_proxy_url(db, "yandex")
    # Равный ban_count=2 у обоих -- решает consecutive_fails (0 < 2).
    assert result == "http://u:p@solid-history.local:8080"


def test_ban_history_on_other_source_does_not_affect_ranking() -> None:
    """Высокий ban_count по source=cian у узла НЕ влияет на его ранжирование для
    source=avito -- история строго per-source, ровно как активный бан (#2600 п.2)."""
    now = datetime.now(UTC)
    db = FakeSession(
        [
            _proxy(
                1,
                consecutive_fails=0,
                last_ok_at=now,
                url="http://u:p@cian-history-only.local:8080",
            ),
            _proxy(
                2,
                consecutive_fails=0,
                last_ok_at=now - timedelta(hours=2),
                url="http://u:p@clean-everywhere.local:8080",
            ),
        ],
        bans=[
            {
                "proxy_id": 1,
                "source": "cian",  # ДРУГОЙ source, не avito
                "banned_until": now - timedelta(hours=1),
                "ban_count": 9,
            }
        ],
    )
    result = resolve_proxy_url(db, "avito")
    # Для avito у узла 1 ban_count=0 (истории по avito нет) -- выигрывает по last_ok_at.
    assert result == "http://u:p@cian-history-only.local:8080"


def test_active_ban_still_excludes_regardless_of_ban_count() -> None:
    """Активный бан по-прежнему полный фильтр -- ban_count=1 (низкий) не спасает узел
    с АКТИВНЫМ баном от исключения."""
    now = datetime.now(UTC)
    db = FakeSession(
        [_proxy(1, url="http://u:p@actively-banned.local:8080")],
        bans=[
            {
                "proxy_id": 1,
                "source": "avito",
                "banned_until": now + timedelta(hours=6),
                "ban_count": 1,
            }
        ],
    )
    with pytest.raises(ProxyPoolExhaustedError):
        resolve_proxy_url(db, "avito")


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)