"""Offline-тесты ротации exit-IP ASocks (#2600 п.5).

Покрытие БЕЗ live-сети/БД: httpx.AsyncClient подменён предсказуемым фейком,
FakeSession эмулирует scrape_proxies (одна строка) + scrape_proxy_rotations
(append-only список), pytest-asyncio (asyncio_mode=auto, см. pyproject.toml).

  - rotate_url пуст → «не поддерживается», НЕ ошибка, HTTP не дёргается.
  - ASOCKS_API_TOKEN не задан → внятный отказ, HTTP не дёргается.
  - 4-я попытка за сутки отклоняется БЕЗ обращения к API (лимит 3/сутки).
  - Успешная ротация пишет запись в scrape_proxy_rotations (success=True).
  - 401 → logger.error (громкий отказ) + sentry_sdk.capture_message (мониторинг),
    аудит-запись пишется, но НЕ считается против суточного лимита.
  - Токен не появляется ни в RotationResult.reason, ни в note аудит-записи, ни в
    тексте log-сообщений (caplog.getMessage()), ни в тексте, ушедшем в Sentry —
    ни в одном из сценариев (сеть-ошибка, 401, provider 5xx, success).
  - rotate_url на ЧУЖОМ хосте (не ALLOWED_ROTATE_HOST) → отказ ДО HTTP-вызова —
    scrape_proxies.rotate_url колонка неоднородна (несёт и mobileproxy changeip-
    ссылки), наш ASOCKS_API_TOKEN не должен уйти на них (security review PR #2611).
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import logging
from datetime import UTC, datetime, timedelta
from typing import Any

import httpx
import pytest

from app.services import proxy_rotation

SECRET_TOKEN = "asocks-super-secret-token-must-never-leak-1a2b3c"

# ── stateful fakes ────────────────────────────────────────────────────────────


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

    def fetchall(self) -> list[Any]:
        # RETURNING source у clear_source_bans — код читает r.source (attribute access)
        return [type("Row", (), r)() for r in self._rows]


class FakeSession:
    """Эмуляция Session: одна строка scrape_proxies + append-only
    scrape_proxy_rotations, интерпретирует SQL по ключевым фрагментам (тот же
    паттерн, что tests/services/test_proxy_pool.py)."""

    def __init__(
        self,
        proxy_row: dict[str, Any] | None,
        rotations: list[dict[str, Any]] | None = None,
        source_bans: list[dict[str, Any]] | None = None,
    ):
        self.proxy_row = proxy_row
        self.rotations: list[dict[str, Any]] = rotations or []
        # #2600 п.2: успешная ротация снимает баны узла (IP сменился — бан старого
        # адреса недействителен), см. proxy_pool.clear_source_bans.
        self.source_bans: list[dict[str, Any]] = source_bans or []
        self.commits = 0
        self.exit_ip_updates: list[tuple[int, str]] = []

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        sql = str(stmt)
        p = params or {}

        if "SELECT id, rotate_url FROM scrape_proxies" in sql:
            if self.proxy_row is None or self.proxy_row["id"] != p["id"]:
                return _FakeResult([])
            return _FakeResult([dict(self.proxy_row)])

        if "SELECT count(*) AS n" in sql and "scrape_proxy_rotations" in sql:
            cutoff = datetime.now(UTC) - timedelta(hours=24)
            n = sum(
                1
                for r in self.rotations
                if r["proxy_id"] == p["proxy_id"]
                and r["rotated_at"] > cutoff
                and r["http_status"] is not None
                and r["http_status"] != 401
            )
            return _FakeResult([{"n": n}])

        if "INSERT INTO scrape_proxy_rotations" in sql:
            self.rotations.append(
                {
                    "proxy_id": p["proxy_id"],
                    "success": p["success"],
                    "http_status": p["http_status"],
                    "note": p["note"],
                    "rotated_at": datetime.now(UTC),
                }
            )
            return _FakeResult([])

        if "UPDATE scrape_proxy_source_bans" in sql and "cleared_reason" in sql:
            # clear_source_bans (#3404): гасит строку (cleared_reason=reason), а НЕ
            # удаляет — строка остаётся для трассируемости. Фейк мутирует найденные
            # записи в месте, не убирая их из self.source_bans.
            target_source = p.get("source")
            cleared = [
                b
                for b in self.source_bans
                if b["proxy_id"] == p["proxy_id"]
                and (target_source is None or b["source"] == target_source)
            ]
            for b in cleared:
                b["cleared_reason"] = p["reason"]
            return _FakeResult([{"source": b["source"]} for b in cleared])

        if "UPDATE scrape_proxies" in sql and "exit_ip" in sql:
            self.exit_ip_updates.append((p["proxy_id"], p["exit_ip"]))
            if self.proxy_row is not None and self.proxy_row["id"] == p["proxy_id"]:
                self.proxy_row["exit_ip"] = p["exit_ip"]
            return _FakeResult([])

        raise AssertionError(f"unhandled SQL: {sql}")

    def commit(self) -> None:
        self.commits += 1

    def rollback(self) -> None:
        pass


class _FakeResponse:
    def __init__(self, status_code: int, json_data: dict[str, Any] | None):
        self.status_code = status_code
        self._json_data = json_data

    def json(self) -> dict[str, Any]:
        if self._json_data is None:
            raise ValueError("no json body")
        return self._json_data


def _fake_async_client(
    *,
    response: tuple[int, dict[str, Any] | None] | None,
    exception: Exception | None,
):
    """Строит замену httpx.AsyncClient, никогда не бьющую в реальную сеть.

    Ровно один из (response, exception) задан. calls накапливает (url, headers)
    каждого post() — тест проверяет по ним, был ли вообще HTTP-вызов.
    """
    calls: list[dict[str, Any]] = []

    class _FakeClientImpl:
        def __init__(self, timeout: float | None = None) -> None:
            self.timeout = timeout

        async def __aenter__(self) -> _FakeClientImpl:
            return self

        async def __aexit__(self, *exc: object) -> bool:
            return False

        async def post(self, url: str, headers: dict[str, str] | None = None) -> _FakeResponse:
            calls.append({"url": url, "headers": headers or {}})
            if exception is not None:
                raise exception
            assert response is not None
            status, body = response
            return _FakeResponse(status, body)

    return _FakeClientImpl, calls


def _no_http_allowed():
    """httpx.AsyncClient-заглушка, падающая AssertionError при любом post() —
    для сценариев, где HTTP до провайдера дойти НЕ должно."""

    class _ForbiddenClient:
        def __init__(self, timeout: float | None = None) -> None:
            pass

        async def __aenter__(self) -> _ForbiddenClient:
            return self

        async def __aexit__(self, *exc: object) -> bool:
            return False

        async def post(self, *a: object, **kw: object) -> None:
            raise AssertionError("HTTP call must NOT happen for this scenario")

        async def get(self, *a: object, **kw: object) -> None:
            raise AssertionError("HTTP call must NOT happen for this scenario")

    return _ForbiddenClient


def _fake_async_client_get(
    *,
    response: tuple[int, dict[str, Any] | None] | None,
    exception: Exception | None,
):
    """Аналог _fake_async_client, но для GET (mobileproxy — без Authorization,
    секрет живёт в URL). calls накапливает str(url) каждого get()."""
    calls: list[dict[str, Any]] = []

    class _FakeClientImpl:
        def __init__(self, timeout: float | None = None) -> None:
            self.timeout = timeout

        async def __aenter__(self) -> _FakeClientImpl:
            return self

        async def __aexit__(self, *exc: object) -> bool:
            return False

        async def get(self, url: object) -> _FakeResponse:
            calls.append({"url": str(url)})
            if exception is not None:
                raise exception
            assert response is not None
            status, body = response
            return _FakeResponse(status, body)

    return _FakeClientImpl, calls


_DEFAULT_ROTATE_URL = "https://api.asocks.com/unlimited-proxy/1/refresh-ip"
MOBILEPROXY_KEY = "deadbeef00112233deadbeef00112233"
_MOBILEPROXY_ROTATE_URL = f"https://aproxy.site/?proxy_key={MOBILEPROXY_KEY}"

# (name, rotate_url, response=(status, json_body)|None, exception|None) — ровно один
# из response/exception задан, либо оба None (локальный отказ, HTTP не идёт).
_LogScenario = tuple[str, str | None, tuple[int, dict[str, Any] | None] | None, Exception | None]


def _proxy_row(rotate_url: str | None = _DEFAULT_ROTATE_URL) -> dict[str, Any]:
    return {"id": 1, "rotate_url": rotate_url}


def _quota_rows(proxy_id: int, n: int, *, http_status: int = 200) -> list[dict[str, Any]]:
    now = datetime.now(UTC)
    return [
        {
            "proxy_id": proxy_id,
            "success": http_status < 400,
            "http_status": http_status,
            "note": None,
            "rotated_at": now - timedelta(minutes=i),
        }
        for i in range(n)
    ]


# ── no rotate_url → not an error ────────────────────────────────────────────


async def test_no_rotate_url_is_not_an_error(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    db = FakeSession(_proxy_row(rotate_url=None))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert result.reason is not None
    assert "rotat" in result.reason.lower()  # human-readable, not a crash
    assert db.rotations == []  # ничего не писалось — попытки не было


# ── host pinning (security review PR #2611) ─────────────────────────────────
#
# scrape_proxies.rotate_url колонка неоднородна: прод сейчас несёт mobileproxy
# changeip-ссылки (id 3/4/5) БОК О БОК с ASocks-ссылками (id 1/9/10/11, миграция
# 199). Без host-пиннинга наш Authorization: Bearer <ASOCKS_API_TOKEN> ушёл бы
# на чужой провайдер.


async def test_rotate_url_on_foreign_host_refused_before_http_call(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    foreign_url = "https://changeip.mobileproxy.space/?proxy_key=mobileproxy-own-secret"
    db = FakeSession(_proxy_row(rotate_url=foreign_url))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert result.reason is not None
    # _no_http_allowed() would have raised AssertionError from within rotate_proxy
    # if the code had tried an HTTP call (i.e. sent our token) — reaching this
    # line means it refused first. Belt-and-suspenders: no audit row either
    # (this is a local rejection, same as no-rotate_url/no-token/limit).
    assert db.rotations == []
    assert SECRET_TOKEN not in result.reason


async def test_allowed_host_case_insensitive_still_proceeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Хост сверяется без учёта регистра (urlparse().hostname лоуеркейзит) — тот
    же ALLOWED_ROTATE_HOST в другом регистре ДОЛЖЕН проходить, иначе пиннинг
    превратился бы в ложный отказ на легитимном rotate_url."""
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, calls = _fake_async_client(response=(200, {"ip": "1.2.3.4"}), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(rotate_url="https://API.ASOCKS.COM/unlimited-proxy/1/refresh-ip"))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is True
    assert len(calls) == 1


async def test_allowed_host_over_plain_http_is_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    """http:// (не https://) на тот же хост — отказ (защита от даунгрейда
    транспорта, которым Authorization ушёл бы в открытом виде)."""
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    db = FakeSession(_proxy_row(rotate_url="http://api.asocks.com/unlimited-proxy/1/refresh-ip"))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert db.rotations == []


# ── missing token → neutral refusal, no crash ───────────────────────────────


async def test_missing_token_is_neutral_refusal(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", "")
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    db = FakeSession(_proxy_row())
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert result.reason is not None
    assert db.rotations == []


# ── daily limit ──────────────────────────────────────────────────────────────


async def test_fourth_attempt_today_rejected_without_api_call(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    # 3 quota-consuming попытки уже сегодня (успешные 200 — засчитываются).
    db = FakeSession(_proxy_row(), _quota_rows(1, proxy_rotation.DAILY_ROTATION_LIMIT))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert "limit" in (result.reason or "").lower() or "лимит" in (result.reason or "").lower()
    assert result.rotations_remaining_today == 0
    # _no_http_allowed() would have raised AssertionError from within rotate_proxy
    # if the code had tried an HTTP call — reaching here means it didn't.
    assert len(db.rotations) == proxy_rotation.DAILY_ROTATION_LIMIT  # ничего нового не дописано


# ── success writes history ──────────────────────────────────────────────────


async def test_successful_rotation_writes_history_row(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, calls = _fake_async_client(response=(200, {"ip": "9.9.9.9"}), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row())
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is True
    assert result.new_ip == "9.9.9.9"
    assert result.rotations_remaining_today == proxy_rotation.DAILY_ROTATION_LIMIT - 1
    assert len(calls) == 1
    assert calls[0]["headers"]["Authorization"] == f"Bearer {SECRET_TOKEN}"

    assert len(db.rotations) == 1
    row = db.rotations[0]
    assert row["success"] is True
    assert row["http_status"] == 200
    assert db.commits >= 1


async def test_successful_rotation_clears_source_bans(monkeypatch: pytest.MonkeyPatch) -> None:
    """(#2600 п.2) Сменился exit-IP → баны площадок на СТАРОМ адресе недействительны.

    Строка бана привязана к proxy_id, а не к IP — без снятия узел остался бы вне выдачи
    источнику до 72 часов уже без причины.
    """
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, _calls = _fake_async_client(response=(200, {"ip": "9.9.9.9"}), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(
        _proxy_row(),
        source_bans=[
            {"proxy_id": 1, "source": "avito"},
            {"proxy_id": 1, "source": "cian"},
            {"proxy_id": 2, "source": "avito"},  # чужой узел — не трогаем
        ],
    )
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is True
    # #3404: строки не удаляются — гасятся (cleared_reason проставлен), остаются для
    # трассируемости. Чужой узел (proxy_id=2) не тронут вообще.
    proxy1_bans = [b for b in db.source_bans if b["proxy_id"] == 1]
    assert len(proxy1_bans) == 2
    assert all(b.get("cleared_reason") == "exit ip rotated (status=200)" for b in proxy1_bans)
    other_ban = next(b for b in db.source_bans if b["proxy_id"] == 2)
    assert other_ban.get("cleared_reason") is None


async def test_failed_rotation_keeps_source_bans(monkeypatch: pytest.MonkeyPatch) -> None:
    """Провайдер ответил ошибкой — IP НЕ сменился, баны обязаны остаться."""
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, _calls = _fake_async_client(response=(500, None), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(), source_bans=[{"proxy_id": 1, "source": "avito"}])
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert db.source_bans == [{"proxy_id": 1, "source": "avito"}]


# ── 401 → loud failure ───────────────────────────────────────────────────────


async def test_401_logs_error_and_alerts_monitoring_excluded_from_quota(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, calls = _fake_async_client(
        response=(401, {"success": False, "message": "Unauthenticated"}), exception=None
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    sentry_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(
        "sentry_sdk.capture_message",
        lambda msg, level=None: sentry_calls.append((msg, level)),
    )

    db = FakeSession(_proxy_row())
    with caplog.at_level(logging.ERROR):
        result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert len(calls) == 1  # запрос реально ушёл

    # громкий отказ: и лог, и мониторинг — не молчаливая остановка
    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
    assert any("401" in r.getMessage() for r in error_records)
    assert len(sentry_calls) == 1
    assert sentry_calls[0][1] == "error"

    # аудит записан, но 401 НЕ считается против суточного лимита (см. модуль
    # docstring: auth-отсев до провайдера, лимит на его стороне не тратится).
    assert len(db.rotations) == 1
    assert db.rotations[0]["http_status"] == 401
    assert db.rotations[0]["success"] is False
    assert proxy_rotation._quota_used_today(db, 1) == 0  # type: ignore[arg-type]

    second = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]
    # 401 не съел лимит — снова полный DAILY_ROTATION_LIMIT доступен
    assert second.rotations_remaining_today == proxy_rotation.DAILY_ROTATION_LIMIT


# ── token never leaks ────────────────────────────────────────────────────────


async def test_token_never_appears_in_reason_success(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, _ = _fake_async_client(response=(200, {"ip": "1.1.1.1"}), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(), source_bans=[{"proxy_id": 1, "source": "avito"}])
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]
    assert SECRET_TOKEN not in (result.reason or "")
    assert SECRET_TOKEN not in (result.new_ip or "")
    # #3404: успешная ротация гасит бан и пишет cleared_reason — секрет не должен
    # попасть и туда (та же гигиена, что для reason/note/логов).
    assert SECRET_TOKEN not in (db.source_bans[0].get("cleared_reason") or "")


async def test_token_never_appears_in_reason_on_401(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr("sentry_sdk.capture_message", lambda *a, **kw: None)
    fake_client, _ = _fake_async_client(
        response=(401, {"message": "Unauthenticated"}), exception=None
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row())
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]
    assert SECRET_TOKEN not in (result.reason or "")
    assert all(SECRET_TOKEN not in (r["note"] or "") for r in db.rotations)


async def test_token_never_appears_in_reason_on_network_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """httpx-исключения могут нести полный request-контекст (URL/детали) —
    прецедент утечки: app.api.v1.admin.rotate_proxy_ip (~line 2400). Здесь токен
    живёт только в headers (не в URL), но проверяем end-to-end: даже если
    exception-текст содержит секрет (симулируем это явно), наружу он не идёт."""
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    boom = httpx.ConnectError(f"connection failed while POSTing token={SECRET_TOKEN}")
    fake_client, _ = _fake_async_client(response=None, exception=boom)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row())
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert SECRET_TOKEN not in (result.reason or "")
    assert all(SECRET_TOKEN not in (r["note"] or "") for r in db.rotations)
    # сетевая ошибка не подтверждает, что провайдер обработал попытку → квота не тратится
    assert db.rotations[0]["http_status"] is None
    assert proxy_rotation._quota_used_today(db, 1) == 0  # type: ignore[arg-type]
    # exception class name (не секрет) в note — оператор отличит "не дозвонились"
    # (ConnectError) от "дозвонились, зависли" (ReadTimeout).
    assert "ConnectError" in (db.rotations[0]["note"] or "")


async def test_token_never_appears_on_provider_error_status(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    fake_client, _ = _fake_async_client(
        response=(500, {"message": "internal error"}), exception=None
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row())
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert SECRET_TOKEN not in (result.reason or "")
    # провайдер прошёл auth и ответил своей ошибкой (500) — засчитывается в квоту
    assert db.rotations[0]["http_status"] == 500
    assert proxy_rotation._quota_used_today(db, 1) == 1  # type: ignore[arg-type]


async def test_token_never_appears_in_log_messages_or_sentry_text(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    """Расширенное leak-покрытие (security review PR #2611): предыдущие тесты
    проверяли только reason/note. Здесь — текст, реально уходящий в logging и в
    Sentry (не exc_info-traceback, который по дизайну МОЖЕТ нести детали
    исключения — см. модуль docstring; это осознанно разрешённое место).
    caplog.records[i].getMessage() возвращает форматированный msg %% args, БЕЗ
    exc_text — то есть эта проверка ловит именно "секрет попал в аргумент
    logger.*()", а не в traceback.
    """
    sentry_texts: list[str] = []
    monkeypatch.setattr(
        "sentry_sdk.capture_message",
        lambda msg, level=None: sentry_texts.append(msg),
    )
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)

    scenarios: list[_LogScenario] = [
        ("success", _DEFAULT_ROTATE_URL, (200, {"ip": "1.1.1.1"}), None),
        ("401", _DEFAULT_ROTATE_URL, (401, {"message": "Unauthenticated"}), None),
        ("provider_500", _DEFAULT_ROTATE_URL, (500, {"message": "err"}), None),
        (
            "network_error",
            _DEFAULT_ROTATE_URL,
            None,
            httpx.ConnectError(f"boom token={SECRET_TOKEN}"),
        ),
        ("foreign_host", "https://changeip.mobileproxy.space/?proxy_key=x", None, None),
    ]

    for name, rotate_url, response, exception in scenarios:
        if response is not None or exception is not None:
            fake_client, _ = _fake_async_client(response=response, exception=exception)
            monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)
        else:
            monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

        db = FakeSession(
            _proxy_row(rotate_url=rotate_url),
            source_bans=[{"proxy_id": 1, "source": "avito"}],
        )
        with caplog.at_level(logging.DEBUG):
            caplog.clear()
            await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

        for record in caplog.records:
            assert SECRET_TOKEN not in record.getMessage(), (
                f"scenario={name}: token leaked into log message args"
            )
        if name == "success":
            # #3404: успешная ротация гасит бан и пишет cleared_reason — секрет не
            # должен попасть и туда.
            assert SECRET_TOKEN not in (db.source_bans[0].get("cleared_reason") or ""), (
                f"scenario={name}: token leaked into cleared_reason"
            )

    assert sentry_texts, "expected at least one Sentry capture (401 scenario)"
    assert all(SECRET_TOKEN not in text for text in sentry_texts)


# ── proxy not found ──────────────────────────────────────────────────────────


async def test_unknown_proxy_id_returns_neutral_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    db = FakeSession(None)
    result = await proxy_rotation.rotate_proxy(db, 999)  # type: ignore[arg-type]
    assert result.ok is False
    assert result.reason is not None


# ── mobileproxy.space (aproxy.site) ─────────────────────────────────────────


async def test_mobileproxy_success_writes_history_and_exit_ip(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_client, calls = _fake_async_client_get(
        response=(200, {"status": "OK", "code": 200, "new_ip": "128.71.225.59", "rt": "2.41"}),
        exception=None,
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is True
    assert result.new_ip == "128.71.225.59"
    assert result.reconnect_delay_s == pytest.approx(2.41)
    assert (
        result.rotations_remaining_today
        == proxy_rotation.MOBILEPROXY_DAILY_ROTATION_LIMIT - 1
    )

    assert len(calls) == 1
    assert "format=json" in calls[0]["url"]

    assert len(db.rotations) == 1
    assert db.rotations[0]["success"] is True
    assert db.rotations[0]["http_status"] == 200
    assert db.exit_ip_updates == [(1, "128.71.225.59")]


async def test_mobileproxy_success_clears_source_bans(monkeypatch: pytest.MonkeyPatch) -> None:
    fake_client, _calls = _fake_async_client_get(
        response=(200, {"status": "OK", "new_ip": "9.9.9.9", "rt": "3.1"}), exception=None
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(
        _proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL),
        source_bans=[{"proxy_id": 1, "source": "avito"}],
    )
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is True
    # #3404: гашение, не удаление — строка остаётся, cleared_reason помечает снятие.
    assert len(db.source_bans) == 1
    assert db.source_bans[0]["cleared_reason"] == "exit ip rotated (mobileproxy status=200)"


async def test_mobileproxy_non_ok_status_is_failure_and_consumes_quota(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """HTTP 200, но body.status != "OK" — провал, даже без транспортной ошибки.
    Провайдер ответил (транспортно дошли) → попытка засчитывается в предохранитель."""
    fake_client, _calls = _fake_async_client_get(
        response=(200, {"status": "FAIL", "code": 400}), exception=None
    )
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert result.new_ip is None
    assert db.exit_ip_updates == []
    assert len(db.rotations) == 1
    assert db.rotations[0]["success"] is False
    assert db.rotations[0]["http_status"] == 200
    assert proxy_rotation._quota_used_today(db, 1) == 1  # type: ignore[arg-type]


async def test_mobileproxy_http_error_status_is_failure(monkeypatch: pytest.MonkeyPatch) -> None:
    fake_client, _calls = _fake_async_client_get(response=(503, None), exception=None)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert db.rotations[0]["http_status"] == 503


async def test_mobileproxy_network_error_does_not_consume_quota(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    boom = httpx.ConnectError(f"connection failed for proxy_key={MOBILEPROXY_KEY}")
    fake_client, _calls = _fake_async_client_get(response=None, exception=boom)
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

    db = FakeSession(_proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert db.rotations[0]["http_status"] is None
    assert proxy_rotation._quota_used_today(db, 1) == 0  # type: ignore[arg-type]
    assert MOBILEPROXY_KEY not in (result.reason or "")
    assert MOBILEPROXY_KEY not in (db.rotations[0]["note"] or "")


async def test_mobileproxy_daily_limit_rejected_without_api_call(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    db = FakeSession(
        _proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL),
        _quota_rows(1, proxy_rotation.MOBILEPROXY_DAILY_ROTATION_LIMIT),
    )
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert "limit" in (result.reason or "").lower() or "лимит" in (result.reason or "").lower()
    assert result.rotations_remaining_today == 0
    assert len(db.rotations) == proxy_rotation.MOBILEPROXY_DAILY_ROTATION_LIMIT


async def test_mobileproxy_key_never_leaks_in_reason_or_db_or_logs(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    """Проверка секретной гигиены для proxy_key — тот же класс требований, что и
    для ASOCKS_API_TOKEN (см. модуль docstring): reason, note аудит-записи и
    аргументы logger.*() не должны содержать ключ ни в одном сценарии."""
    scenarios: list[tuple[str, tuple[int, dict[str, Any] | None] | None, Exception | None]] = [
        ("success", (200, {"status": "OK", "new_ip": "1.1.1.1", "rt": "2.0"}), None),
        ("non_ok", (200, {"status": "FAIL"}), None),
        ("http_error", (500, {"message": "err"}), None),
        (
            "network_error",
            None,
            httpx.ConnectError(f"boom proxy_key={MOBILEPROXY_KEY}"),
        ),
    ]

    for name, response, exception in scenarios:
        fake_client, _ = _fake_async_client_get(response=response, exception=exception)
        monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)

        db = FakeSession(
            _proxy_row(rotate_url=_MOBILEPROXY_ROTATE_URL),
            source_bans=[{"proxy_id": 1, "source": "avito"}],
        )
        with caplog.at_level(logging.DEBUG):
            caplog.clear()
            result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

        assert MOBILEPROXY_KEY not in (result.reason or ""), name
        for row in db.rotations:
            assert MOBILEPROXY_KEY not in (row["note"] or ""), name
        for record in caplog.records:
            assert MOBILEPROXY_KEY not in record.getMessage(), (
                f"scenario={name}: proxy_key leaked into log message args"
            )
        if name == "success":
            # #3404: успешная ротация гасит бан и пишет cleared_reason — ключ не
            # должен попасть и туда.
            assert MOBILEPROXY_KEY not in (db.source_bans[0].get("cleared_reason") or ""), name


async def test_mobileproxy_unknown_query_shape_still_masks_url_in_refusal(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Хост-пиннинг: неизвестный хост, похожий на mobileproxy по форме query
    (проверка того, что дефолтный отказ путь тоже не светит секрет из url)."""
    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", _no_http_allowed())

    foreign_url = f"https://changeip.mobileproxy.space/?proxy_key={MOBILEPROXY_KEY}"
    db = FakeSession(_proxy_row(rotate_url=foreign_url))
    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]

    assert result.ok is False
    assert db.rotations == []
    assert MOBILEPROXY_KEY not in (result.reason or "")