gendesign/tradein-mvp/backend/tests/test_support.py

"""Offline-тесты веб-чата поддержки (#tgsupport-web) —
POST/GET /api/v1/trade-in/support/{messages,unread,read}.

Storage-слой (`app.services.tgbot.web_support_storage`) мокается целиком (как
mock'ается DB в test_trade_in_lead.py) — эти тесты проверяют РОУТЕР: изоляцию
тредов (нет параметра, которым можно адресовать чужой тред), валидацию
входа, поведение при несконфигурированном боте, rate-limit и то, что реплай
парсится в `bridge.py` в web-ветку (см. tests/services/tgbot/test_bridge.py —
маршрутизация реплая тестируется ТАМ; здесь — только HTTP-контракт отправки/
чтения).

NEVER touches real DB / real Telegram API.
"""

from __future__ import annotations

import os
import time

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

from typing import Any, ClassVar
from unittest.mock import MagicMock

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy.exc import SQLAlchemyError

from app.api.v1 import support as support_module
from app.core.db import get_db
from app.core.ratelimit import SlidingWindowLimiter
from app.services.tgbot.client import TelegramApiError, TelegramNetworkError


@pytest.fixture(autouse=True)
def _bot_configured(monkeypatch: pytest.MonkeyPatch) -> None:
    """По умолчанию бот считается сконфигурированным — отдельные тесты переопределяют."""
    monkeypatch.setattr(support_module.settings, "telegram_bot_token", "fake-token")
    monkeypatch.setattr(support_module.settings, "telegram_support_chat_id", -100123456789)
    monkeypatch.setattr(support_module.settings, "telegram_support_topic_id", 42)


@pytest.fixture(autouse=True)
def _fresh_rate_limiter(monkeypatch: pytest.MonkeyPatch) -> None:
    """Модульный `_send_limiter` иначе накапливает состояние МЕЖДУ тестами (один
    процесс pytest) — свежий лимитер на каждый тест, щедрый дефолт (rate-limit
    тестируется отдельно на СВОЁМ, явно узком экземпляре)."""
    monkeypatch.setattr(
        support_module, "_send_limiter", SlidingWindowLimiter(limit=1000, window_s=60.0)
    )


@pytest.fixture(autouse=True)
def _fresh_failure_limiters(monkeypatch: pytest.MonkeyPatch) -> None:
    """То же, что `_fresh_rate_limiter`, но для счётчиков ОТКАЗОВ отправки: они
    модульные, и тесты, специально роняющие Telegram, иначе оставляли бы cooldown
    следующим тестам в том же процессе."""
    for name in ("_send_failure_limiter", "_anon_ip_failure_limiter"):
        monkeypatch.setattr(support_module, name, SlidingWindowLimiter(limit=1000, window_s=30.0))


class _FakeTelegramClient:
    """Подменяет `TelegramClient` внутри `support` модуля — никакого httpx/сети."""

    calls: ClassVar[list[dict[str, Any]]] = []
    _response: ClassVar[dict[str, Any] | Exception] = {"message_id": 555}

    def __init__(self, _token: str) -> None:
        pass

    async def send_message(self, **kwargs: Any) -> dict[str, Any]:
        _FakeTelegramClient.calls.append(kwargs)
        if isinstance(_FakeTelegramClient._response, Exception):
            raise _FakeTelegramClient._response
        return _FakeTelegramClient._response


@pytest.fixture(autouse=True)
def _fake_telegram_client(monkeypatch: pytest.MonkeyPatch) -> Any:
    _FakeTelegramClient.calls = []
    _FakeTelegramClient._response = {"message_id": 555}
    # Ручка берёт ОБЩИЙ клиент приложения (#tg-connection-resilience), а не
    # создаёт свой на запрос — подменяем аксессор, а не класс.
    monkeypatch.setattr(
        support_module, "get_telegram_client", lambda: _FakeTelegramClient("fake-token")
    )
    return _FakeTelegramClient


@pytest.fixture
def db() -> MagicMock:
    return MagicMock()


@pytest.fixture
def client(db: MagicMock) -> TestClient:
    app = FastAPI()
    app.include_router(support_module.router, prefix="/api/v1/trade-in")

    def fake_db() -> Any:
        yield db

    app.dependency_overrides[get_db] = fake_db
    # https, а не дефолтный http: анонимная ветка ставит идентити-куку с
    # `secure=True` (как session-cookie), и по http httpx её не вернул бы в
    # следующем запросе — тесты «тот же тред / тот же бюджет лимита» тихо
    # проверяли бы каждый раз НОВОГО анонима. Прод и так только https.
    return TestClient(app, base_url="https://testserver")


def _auth(username: str = "alice") -> dict[str, str]:
    return {"x-authenticated-user": username}


# ── auth guard ────────────────────────────────────────────────────────────────


def test_send_message_without_auth_header_401(client: TestClient) -> None:
    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"})
    assert r.status_code == 401


def test_list_messages_without_auth_header_401(client: TestClient) -> None:
    r = client.get("/api/v1/trade-in/support/messages")
    assert r.status_code == 401


def test_whitespace_only_auth_header_401(client: TestClient) -> None:
    """review L4: заголовок из одних пробелов после `.strip()` пуст — не должен
    считаться валидным auth (не проваливается тихо в "" как username)."""
    r = client.get("/api/v1/trade-in/support/messages", headers={"x-authenticated-user": "   "})
    assert r.status_code == 401


def test_auth_header_with_surrounding_whitespace_is_stripped(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """review L4: лишний пробел от прокси не должен заводить ВТОРОЙ тред для, по
    сути, того же пользователя — username резолвится в тред по .strip()-нутому
    значению."""
    seen_usernames = []

    def fake_find_thread_id(db, username):
        seen_usernames.append(username)
        return None

    monkeypatch.setattr(support_module.storage, "find_thread_id", fake_find_thread_id)

    r = client.get(
        "/api/v1/trade-in/support/messages", headers={"x-authenticated-user": "  alice  "}
    )
    assert r.status_code == 200
    assert seen_usernames == ["alice"]


# ── validation ────────────────────────────────────────────────────────────────


def test_send_message_blank_text_422(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    called = []
    monkeypatch.setattr(
        support_module.storage, "get_or_create_thread", lambda *a, **kw: called.append(1) or 1
    )
    r = client.post("/api/v1/trade-in/support/messages", json={"text": "   "}, headers=_auth())
    assert r.status_code == 422, r.text
    assert called == []  # ничего не персистится на невалидном вводе


def test_send_message_empty_text_422(client: TestClient) -> None:
    r = client.post("/api/v1/trade-in/support/messages", json={"text": ""}, headers=_auth())
    assert r.status_code == 422


def test_send_message_too_long_422(client: TestClient) -> None:
    too_long = "a" * (support_module.MAX_MESSAGE_LENGTH + 1)
    r = client.post("/api/v1/trade-in/support/messages", json={"text": too_long}, headers=_auth())
    assert r.status_code == 422


def test_send_message_at_max_length_ok(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda *a, **kw: 1)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-07-26T00:00:00+00:00",
        },
    )
    at_limit = "a" * support_module.MAX_MESSAGE_LENGTH
    r = client.post("/api/v1/trade-in/support/messages", json={"text": at_limit}, headers=_auth())
    assert r.status_code == 200, r.text


# ── bot not configured ──────────────────────────────────────────────────────


def test_send_message_bot_token_empty_returns_503_not_500(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(support_module.settings, "telegram_bot_token", "")
    thread_created = []
    monkeypatch.setattr(
        support_module.storage, "get_or_create_thread", lambda *a, **kw: thread_created.append(1)
    )

    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())

    assert r.status_code == 503
    assert r.status_code != 500
    assert thread_created == []  # ничего не создаём/не пишем, если зеркалировать некуда


def test_send_message_support_chat_id_unset_returns_503(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(support_module.settings, "telegram_support_chat_id", 0)
    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
    assert r.status_code == 503


# ── happy path + mirror content ──────────────────────────────────────────────


def test_send_message_happy_path_mirrors_with_website_marker(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    thread_calls: list[str] = []

    def fake_get_or_create_thread(db, username):
        thread_calls.append(username)
        return 7

    monkeypatch.setattr(support_module.storage, "get_or_create_thread", fake_get_or_create_thread)
    recorded = {}

    def fake_record_inbound(db, *, thread_id, text_body, topic_message_id, support_chat_id):
        recorded.update(
            thread_id=thread_id,
            text_body=text_body,
            topic_message_id=topic_message_id,
            support_chat_id=support_chat_id,
        )
        return {
            "id": 100,
            "direction": "in",
            "text_body": text_body,
            "operator_tg_id": None,
            "created_at": "2026-07-26T00:00:00+00:00",
        }

    monkeypatch.setattr(support_module.storage, "record_inbound", fake_record_inbound)

    r = client.post(
        "/api/v1/trade-in/support/messages",
        json={"text": "У меня вопрос про trade-in"},
        headers=_auth("kopylov"),
    )

    assert r.status_code == 200, r.text
    body = r.json()
    assert body["id"] == 100
    assert body["direction"] == "in"
    # Успешный путь ничего не знает про флаг — дефолт `True` (#persisted).
    assert body["persisted"] is True

    # Зеркало помечено "С САЙТА" + username — оператор не путает с TG-клиентом.
    assert len(_fake_telegram_client.calls) == 1
    mirror_call = _fake_telegram_client.calls[0]
    assert "С САЙТА" in mirror_call["text"]
    assert "kopylov" in mirror_call["text"]
    assert "У меня вопрос про trade-in" in mirror_call["text"]
    assert mirror_call["chat_id"] == support_module.settings.telegram_support_chat_id
    assert mirror_call["message_thread_id"] == 42
    # review H1: интерактивный узкий бюджет ретраев/timeout, не воркерный дефолт.
    assert mirror_call["max_retries"] == support_module._INTERACTIVE_SEND_MAX_RETRIES
    assert mirror_call["timeout"] == support_module._INTERACTIVE_SEND_TIMEOUT_S

    assert recorded["thread_id"] == 7
    assert recorded["topic_message_id"] == 555  # из FakeTelegramClient.send_message result
    assert recorded["support_chat_id"] == support_module.settings.telegram_support_chat_id
    assert thread_calls == ["kopylov"]
    assert db.commit.called


def test_send_message_telegram_failure_returns_502_and_does_not_persist(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    _fake_telegram_client._response = TelegramApiError("sendMessage", 400, "chat not found")
    thread_created = []
    monkeypatch.setattr(
        support_module.storage,
        "get_or_create_thread",
        lambda db, username: thread_created.append(1),
    )
    record_called = []
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: record_called.append(1),
    )

    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())

    assert r.status_code == 502
    assert record_called == []  # неотправленное сообщение не персистится
    # review H1: thread создаётся ПОСЛЕ успешной отправки — на неудаче до БД не доходит вообще.
    assert thread_created == []


def test_send_message_telegram_unreachable_returns_502_not_500(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Пользователь не должен получить 500 из-за таймаута до Telegram (#3456).

    Отказ площадки уже отдавал задуманный 502, а вот её недоступность летела
    сырым `httpx.ConnectTimeout` мимо `except TelegramApiError` — FastAPI
    превращал это в 500, и в GlitchTip падала ошибка сервера вместо внятного
    «сервис временно недоступен».
    """
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
    record_called = []
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: record_called.append(1),
    )

    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())

    assert r.status_code == 502, "недоступный Telegram снова отдаёт 500"
    assert record_called == []


# ── rate limit ────────────────────────────────────────────────────────────────


def test_send_message_rate_limited_429(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        support_module, "_send_limiter", SlidingWindowLimiter(limit=1, window_s=60.0)
    )
    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda db, username: 1)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-07-26T00:00:00+00:00",
        },
    )

    first = client.post("/api/v1/trade-in/support/messages", json={"text": "one"}, headers=_auth())
    assert first.status_code == 200, first.text

    second = client.post("/api/v1/trade-in/support/messages", json={"text": "two"}, headers=_auth())
    assert second.status_code == 429
    assert "Retry-After" in second.headers


def test_send_message_failed_attempts_do_not_consume_rate_limit(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """review L3: неудачная отправка НЕ должна расходовать rate-limit бюджет —
    иначе клиент, которому не повезло с транзиентной Telegram-ошибкой, терял бы
    попытки, не доставив НИ ОДНОГО сообщения."""
    monkeypatch.setattr(
        support_module, "_send_limiter", SlidingWindowLimiter(limit=1, window_s=60.0)
    )
    _fake_telegram_client._response = TelegramApiError("sendMessage", 500, "boom")

    for _ in range(3):
        r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
        assert r.status_code == 502

    # "Телеграм" снова работает — бюджет (лимит=1) всё ещё цел, ни одна неудачная
    # попытка выше его не тронула.
    _fake_telegram_client._response = {"message_id": 555}
    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda db, username: 1)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-07-26T00:00:00+00:00",
        },
    )
    ok = client.post("/api/v1/trade-in/support/messages", json={"text": "ok"}, headers=_auth())
    assert ok.status_code == 200, ok.text


def test_send_message_rate_limit_is_per_user(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(
        support_module, "_send_limiter", SlidingWindowLimiter(limit=1, window_s=60.0)
    )
    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda db, username: 1)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-07-26T00:00:00+00:00",
        },
    )

    alice = client.post(
        "/api/v1/trade-in/support/messages", json={"text": "one"}, headers=_auth("alice")
    )
    assert alice.status_code == 200

    bob = client.post(
        "/api/v1/trade-in/support/messages", json={"text": "one"}, headers=_auth("bob")
    )
    assert bob.status_code == 200  # свой ключ лимита — не затронут alice-лимитом


# ── thread isolation (own thread only, no thread_id param exists) ───────────


def test_list_messages_resolves_by_own_username_ignores_foreign_params(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Никакого способа адресовать чужой тред: даже если запрос несёт посторонний
    `thread_id`/`username` в query — эндпоинт их не читает, резолвит ИСКЛЮЧИТЕЛЬНО
    по X-Authenticated-User."""
    seen_usernames = []

    def fake_find_thread_id(db, username):
        seen_usernames.append(username)
        return None

    monkeypatch.setattr(support_module.storage, "find_thread_id", fake_find_thread_id)

    r = client.get(
        "/api/v1/trade-in/support/messages",
        params={"thread_id": 999, "username": "someone-else", "since": 0},
        headers=_auth("alice"),
    )
    assert r.status_code == 200
    assert r.json() == []
    assert seen_usernames == ["alice"]  # username из заголовка — параметры query проигнорированы


def test_list_messages_returns_thread_scoped_rows(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: 7)
    monkeypatch.setattr(
        support_module.storage,
        "list_messages",
        lambda db, *, thread_id, since_id, limit: (
            [
                {
                    "id": 1,
                    "direction": "in",
                    "text_body": "hi",
                    "operator_tg_id": None,
                    "created_at": "2026-07-26T00:00:00+00:00",
                }
            ]
            if thread_id == 7
            else []
        ),
    )
    r = client.get("/api/v1/trade-in/support/messages", params={"since": 0}, headers=_auth("alice"))
    assert r.status_code == 200
    body = r.json()
    assert len(body) == 1
    assert body[0]["text_body"] == "hi"
    # GET строит модель из storage-строки и про флаг не знает — дефолт `True`.
    assert body[0]["persisted"] is True


def test_list_messages_passes_bounded_limit_to_storage(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """review M5: без LIMIT каждое монтирование виджета отдавало бы весь тред."""
    seen_limits = []

    def fake_list_messages(db, *, thread_id, since_id, limit):
        seen_limits.append(limit)
        return []

    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: 7)
    monkeypatch.setattr(support_module.storage, "list_messages", fake_list_messages)

    r = client.get("/api/v1/trade-in/support/messages", headers=_auth())
    assert r.status_code == 200
    assert seen_limits == [support_module._LIST_MESSAGES_LIMIT]


def test_two_users_get_independent_threads(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    thread_by_user = {"alice": 1, "bob": 2}
    monkeypatch.setattr(
        support_module.storage, "find_thread_id", lambda db, username: thread_by_user.get(username)
    )
    monkeypatch.setattr(
        support_module.storage,
        "list_messages",
        lambda db, *, thread_id, since_id, limit: [
            {
                "id": 1,
                "direction": "in",
                "text_body": f"secret-of-thread-{thread_id}",
                "operator_tg_id": None,
                "created_at": "2026-07-26T00:00:00+00:00",
            }
        ],
    )

    alice_resp = client.get("/api/v1/trade-in/support/messages", headers=_auth("alice")).json()
    bob_resp = client.get("/api/v1/trade-in/support/messages", headers=_auth("bob")).json()

    assert alice_resp[0]["text_body"] == "secret-of-thread-1"
    assert bob_resp[0]["text_body"] == "secret-of-thread-2"
    assert alice_resp != bob_resp


# ── unread / read ─────────────────────────────────────────────────────────────


def test_unread_no_thread_returns_zero_without_querying_count(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: None)
    count_called = []
    monkeypatch.setattr(
        support_module.storage, "count_unread", lambda *a, **kw: count_called.append(1)
    )

    r = client.get("/api/v1/trade-in/support/unread", headers=_auth())
    assert r.status_code == 200
    assert r.json() == {"unread": 0}
    assert count_called == []


def test_unread_delegates_to_storage(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: 7)
    monkeypatch.setattr(support_module.storage, "count_unread", lambda db, *, thread_id: 3)

    r = client.get("/api/v1/trade-in/support/unread", headers=_auth())
    assert r.status_code == 200
    assert r.json() == {"unread": 3}


def test_mark_read_noop_when_no_thread(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: None)
    mark_called = []
    monkeypatch.setattr(support_module.storage, "mark_read", lambda *a, **kw: mark_called.append(1))

    r = client.post("/api/v1/trade-in/support/read", headers=_auth())
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}
    assert mark_called == []


def test_mark_read_calls_storage_when_thread_exists(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(support_module.storage, "find_thread_id", lambda db, username: 7)
    mark_called = []
    monkeypatch.setattr(
        support_module.storage, "mark_read", lambda db, *, thread_id: mark_called.append(thread_id)
    )

    r = client.post("/api/v1/trade-in/support/read", headers=_auth())
    assert r.status_code == 200
    assert mark_called == [7]
    assert db.commit.called


# ── анонимная ветка: поддержка без входа (инцидент 2026-07-31) ────────────────


@pytest.fixture(autouse=True)
def _fresh_anon_ip_limiter(monkeypatch: pytest.MonkeyPatch) -> None:
    """Как `_fresh_rate_limiter`, но для per-IP бюджета анонимной ветки — иначе
    состояние течёт между тестами в одном процессе pytest."""
    monkeypatch.setattr(
        support_module, "_anon_ip_limiter", SlidingWindowLimiter(limit=1000, window_s=60.0)
    )


def _patch_anon_storage(monkeypatch: pytest.MonkeyPatch) -> list[str]:
    """Мокает storage для send-пути и возвращает список ключей тредов, с которыми
    его позвали (проверяем, что аноним адресуется `anon:<token>`, а не логином)."""
    seen_keys: list[str] = []

    def fake_get_or_create(db: Any, username: str) -> int:
        seen_keys.append(username)
        return 1

    monkeypatch.setattr(support_module.storage, "get_or_create_thread", fake_get_or_create)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-07-31T00:00:00+00:00",
        },
    )
    return seen_keys


def test_anon_send_without_any_auth_succeeds_and_sets_cookie(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Суть фичи: залогиниться нельзя, а написать в поддержку — можно."""
    seen_keys = _patch_anon_storage(monkeypatch)

    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "не могу войти"})
    assert r.status_code == 200, r.text
    assert r.json()["text_body"] == "не могу войти"
    assert r.json()["persisted"] is True

    token = client.cookies.get(support_module._ANON_COOKIE_NAME)
    assert token is not None
    assert support_module._ANON_TOKEN_RE.match(token)
    # Тред адресован анонимным ключом, не голым токеном и не чьим-то логином.
    assert seen_keys == [f"anon:{token}"]


def test_anon_cookie_reused_across_messages_same_thread(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    seen_keys = _patch_anon_storage(monkeypatch)

    client.post("/api/v1/trade-in/support/anon/messages", json={"text": "первое"})
    token_after_first = client.cookies.get(support_module._ANON_COOKIE_NAME)
    client.post("/api/v1/trade-in/support/anon/messages", json={"text": "второе"})

    assert client.cookies.get(support_module._ANON_COOKIE_NAME) == token_after_first
    assert seen_keys == [f"anon:{token_after_first}"] * 2


def test_anon_mirror_is_labelled_and_never_leaks_token(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Оператор видит, что это НЕзалогиненный посетитель, но bearer треда в
    Telegram-топик не уходит (топик читают люди и пересылают дальше)."""
    _patch_anon_storage(monkeypatch)

    client.post("/api/v1/trade-in/support/anon/messages", json={"text": "помогите"})
    token = client.cookies.get(support_module._ANON_COOKIE_NAME)
    sent_text = _fake_telegram_client.calls[-1]["text"]

    assert sent_text.startswith("[С САЙТА · БЕЗ ВХОДА] anon-")
    assert "помогите" in sent_text
    assert token not in sent_text
    assert support_module._anon_display_id(token) in sent_text


def test_anon_read_paths_without_cookie_are_empty_not_401(client: TestClient) -> None:
    """Виджет поллит эти ручки ДО первого сообщения — 401 там был бы ложной ошибкой."""
    assert client.get("/api/v1/trade-in/support/anon/messages").status_code == 200
    assert client.get("/api/v1/trade-in/support/anon/messages").json() == []
    assert client.get("/api/v1/trade-in/support/anon/unread").json() == {"unread": 0}
    assert client.post("/api/v1/trade-in/support/anon/read").json() == {"status": "ok"}


def test_anon_malformed_cookie_ignored_and_never_reaches_storage(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Кука клиент-контролируема: мусор из браузера не должен становиться ключом
    треда. Считаем куку отсутствующей и выдаём новую."""
    seen_keys = _patch_anon_storage(monkeypatch)
    bogus = "not-a-valid-token!@#$%^"
    client.cookies.set(support_module._ANON_COOKIE_NAME, bogus)

    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "hi"})
    assert r.status_code == 200

    assert len(seen_keys) == 1
    assert bogus not in seen_keys[0]
    assert seen_keys[0].startswith("anon:")
    assert support_module._ANON_TOKEN_RE.match(seen_keys[0].removeprefix("anon:"))


def test_anon_read_path_with_malformed_cookie_returns_empty(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    find_calls = []
    monkeypatch.setattr(
        support_module.storage,
        "find_thread_id",
        lambda db, username: find_calls.append(username),
    )
    client.cookies.set(support_module._ANON_COOKIE_NAME, "!!not-a-token!!")

    assert client.get("/api/v1/trade-in/support/anon/messages").json() == []
    assert find_calls == []  # до storage мусор не доехал вообще


def test_anon_per_ip_rate_limit_429(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    """Главный анти-абуз: per-token бюджет обходится сбросом куки, per-IP — нет."""
    _patch_anon_storage(monkeypatch)
    monkeypatch.setattr(
        support_module, "_anon_ip_limiter", SlidingWindowLimiter(limit=1, window_s=60.0)
    )

    assert (
        client.post("/api/v1/trade-in/support/anon/messages", json={"text": "1"}).status_code == 200
    )
    # Ротация куки НЕ спасает — бюджет привязан к IP.
    client.cookies.delete(support_module._ANON_COOKIE_NAME)
    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "2"})
    assert r.status_code == 429
    assert "Retry-After" in r.headers


def test_anon_per_token_rate_limit_429(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    _patch_anon_storage(monkeypatch)
    monkeypatch.setattr(
        support_module, "_send_limiter", SlidingWindowLimiter(limit=1, window_s=60.0)
    )

    assert (
        client.post("/api/v1/trade-in/support/anon/messages", json={"text": "1"}).status_code == 200
    )
    assert (
        client.post("/api/v1/trade-in/support/anon/messages", json={"text": "2"}).status_code == 429
    )


def test_anon_failed_send_sets_no_cookie_and_writes_nothing(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Неудачная отправка не должна закреплять за посетителем пустой тред."""
    seen_keys = _patch_anon_storage(monkeypatch)
    _fake_telegram_client._response = TelegramApiError("sendMessage", 500, "boom")

    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "hi"})
    assert r.status_code == 502
    assert seen_keys == []
    assert client.cookies.get(support_module._ANON_COOKIE_NAME) is None


def test_anon_telegram_unreachable_returns_502_not_500(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Тот же контракт на анонимной ручке — она открыта наружу без авторизации."""
    seen_keys = _patch_anon_storage(monkeypatch)
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)

    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "hi"})

    assert r.status_code == 502, "недоступный Telegram снова отдаёт 500"
    assert seen_keys == []
    assert client.cookies.get(support_module._ANON_COOKIE_NAME) is None


def test_anon_bot_not_configured_503(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(support_module.settings, "telegram_bot_token", "")
    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "hi"})
    assert r.status_code == 503
    assert client.cookies.get(support_module._ANON_COOKIE_NAME) is None


def test_anon_blank_text_422(client: TestClient) -> None:
    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "   "})
    assert r.status_code == 422


def test_anon_thread_key_cannot_collide_with_real_username() -> None:
    """Инвариант изоляции: `anon:` невозможен в реальном логине (CHECK миграции
    193 + Pydantic `^[A-Za-z0-9._-]{3,64}$`), значит аноним структурно не может
    попасть в тред существующего пользователя."""
    from app.schemas.team import _USERNAME_RE

    key = support_module._anon_thread_key(support_module.secrets.token_urlsafe(18))
    assert key.startswith("anon:")
    assert _USERNAME_RE.match(key) is None


# ── отказ БД ПОСЛЕ успешной доставки в топик ─────────────────────────────────
#
# Порядок «сначала Telegram, потом БД» (H1) означает, что SQLAlchemyError здесь
# = сообщение оператору уже доставлено. 500 на это — худший вариант: клиент
# повторяет, в топике дубль, а на осиротевшее зеркало оператор отвечает в пустоту.


def _patch_storage_ok(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda db, username: 7)
    monkeypatch.setattr(
        support_module.storage,
        "record_inbound",
        lambda *a, **kw: {
            "id": 1,
            "direction": "in",
            "text_body": kw["text_body"],
            "operator_tg_id": None,
            "created_at": "2026-09-12T00:00:00+00:00",
        },
    )


def test_send_message_db_failure_after_delivery_returns_success_and_warns_operator(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    _patch_storage_ok(monkeypatch)
    db.commit.side_effect = SQLAlchemyError("connection lost")

    r = client.post(
        "/api/v1/trade-in/support/messages", json={"text": "секретный вопрос"}, headers=_auth()
    )

    # Клиенту НЕ врём про отказ: сообщение оператору правда доставлено, повтор
    # создал бы дубль в топике.
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["text_body"] == "секретный вопрос"
    assert body["id"] == 0  # сентинел «не персистировано», курсор не завышаем
    # Контракт для клиента — именно флаг: без него 200 неотличим от тишины
    # (фронт рендерит транскрипт только из GET, где сообщения нет) и клиент
    # шлёт повтор → дубль в топике, ровно то, против чего вся правка.
    assert body["persisted"] is False
    assert db.rollback.called

    # Оператор предупреждён реплаем к только что доставленному зеркалу.
    assert len(_fake_telegram_client.calls) == 2
    warning = _fake_telegram_client.calls[-1]
    assert warning["reply_to_message_id"] == 555
    assert "НЕ записано в базу" in warning["text"]
    # Оператор не должен ждать повтора: клиенту сказано, что сообщение принято.
    assert "Повтора тоже не ждите" in warning["text"]
    # Служебное уведомление не дублирует текст обращения (ПДн).
    assert "секретный вопрос" not in warning["text"]


def test_send_message_db_failure_survives_failed_operator_warning(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Отказ уведомления оператору поверх отказа БД не должен становиться 500
    поверх уже доставленного сообщения."""
    _patch_storage_ok(monkeypatch)
    db.commit.side_effect = SQLAlchemyError("connection lost")

    class _FlakyClient:
        calls = 0

        async def send_message(self, **kwargs: Any) -> dict[str, Any]:
            _FlakyClient.calls += 1
            if _FlakyClient.calls == 1:
                return {"message_id": 555}  # зеркало доставлено
            raise TelegramNetworkError("sendMessage", "ConnectTimeout", 4)

    monkeypatch.setattr(support_module, "get_telegram_client", _FlakyClient)

    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())

    assert r.status_code == 200, r.text
    assert _FlakyClient.calls == 2


def test_anon_db_failure_after_delivery_returns_success_and_sets_cookie(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Аноним: кука обязана пережить сбой БД — иначе следующее сообщение заведёт
    ВТОРОЙ тред и переписка разъедется на два."""
    _patch_anon_storage(monkeypatch)
    db.commit.side_effect = SQLAlchemyError("connection lost")

    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "не могу войти"})

    assert r.status_code == 200, r.text
    assert r.json()["id"] == 0
    assert r.json()["persisted"] is False
    token = client.cookies.get(support_module._ANON_COOKIE_NAME)
    assert token is not None and support_module._ANON_TOKEN_RE.match(token)

    warning = _fake_telegram_client.calls[-1]
    assert warning["reply_to_message_id"] == 555
    assert "НЕ записано в базу" in warning["text"]
    assert token not in warning["text"]  # bearer треда в топик не уходит


def test_send_message_db_failure_without_topic_message_id_still_warns(
    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Нет message_id → реплай невозможен, но уведомление всё равно уходит
    отдельным сообщением в топик: хуже реплая, но лучше тишины."""
    _patch_storage_ok(monkeypatch)
    _fake_telegram_client._response = {}  # sendMessage без message_id
    db.commit.side_effect = SQLAlchemyError("connection lost")

    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())

    assert r.status_code == 200, r.text
    assert _fake_telegram_client.calls[-1]["reply_to_message_id"] is None


# ── cooldown по отказам отправки ─────────────────────────────────────────────
#
# Дефект: `retry_after()` — peek, `record()` только на успехе, значит пока
# Telegram лежит, лимита нет ВООБЩЕ, и каждый повтор стоит до 4 попыток к
# api.telegram.org, не расходуя ни один бюджет.


def test_send_failures_trigger_cooldown_and_stop_reaching_telegram(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    monkeypatch.setattr(
        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=2, window_s=30.0)
    )
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)

    for _ in range(2):
        assert (
            client.post(
                "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
            ).status_code
            == 502
        )
    assert len(_fake_telegram_client.calls) == 2

    blocked = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
    assert blocked.status_code == 429
    assert int(blocked.headers["Retry-After"]) >= 1
    # Главное: до Telegram запрос не дошёл вообще.
    assert len(_fake_telegram_client.calls) == 2


def test_send_failure_cooldown_is_per_key(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Cooldown одного пользователя не должен задевать другого."""
    monkeypatch.setattr(
        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=1, window_s=30.0)
    )
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("alice")
        ).status_code
        == 502
    )

    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("alice")
        ).status_code
        == 429
    )
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("bob")
        ).status_code
        == 502
    )


def test_send_failure_cooldown_expires_with_window(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """После окончания окна cooldown снимается сам — пользователя не наказывают
    после восстановления Telegram."""
    monkeypatch.setattr(
        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=1, window_s=0.2)
    )
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
        ).status_code
        == 502
    )
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
        ).status_code
        == 429
    )

    time.sleep(0.25)
    _fake_telegram_client._response = {"message_id": 555}
    _patch_storage_ok(monkeypatch)
    ok = client.post("/api/v1/trade-in/support/messages", json={"text": "ok"}, headers=_auth())
    assert ok.status_code == 200, ok.text


def test_successful_send_resets_failure_counter(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Счётчик считает ПОДРЯД идущие отказы: успех стирает историю, иначе редкие
    транзиентные отказы копились бы в cooldown на живом канале."""
    monkeypatch.setattr(
        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=2, window_s=30.0)
    )
    _patch_storage_ok(monkeypatch)

    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
        ).status_code
        == 502
    )
    _fake_telegram_client._response = {"message_id": 555}
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "ok"}, headers=_auth()
        ).status_code
        == 200
    )

    # Ещё один отказ — всего второй в окне, но ПЕРВЫЙ подряд: cooldown не встаёт.
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
    assert (
        client.post(
            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
        ).status_code
        == 502
    )


def test_anon_send_failures_trigger_cooldown_per_ip(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None:
    """Обход ротацией куки: новый токен — новый per-token счётчик, но per-IP
    счётчик отказов тот же."""
    _patch_anon_storage(monkeypatch)
    monkeypatch.setattr(
        support_module, "_anon_ip_failure_limiter", SlidingWindowLimiter(limit=1, window_s=30.0)
    )
    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)

    assert (
        client.post("/api/v1/trade-in/support/anon/messages", json={"text": "1"}).status_code == 502
    )
    client.cookies.clear()  # «сбросил куку» — по токену бюджет снова пуст
    calls_before = len(_fake_telegram_client.calls)

    blocked = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "2"})
    assert blocked.status_code == 429
    assert len(_fake_telegram_client.calls) == calls_before  # до Telegram не дошло