"""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 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 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 @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) ) 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} monkeypatch.setattr(support_module, "TelegramClient", _FakeTelegramClient) 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" # Зеркало помечено "С САЙТА" + 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 == [] # ── 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" 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:`, а не логином).""" 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"] == "не могу войти" 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_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