"""Тесты аналитики: каталог событий, буферизация и запись в Postgres. Postgres не требуется: writer подменяется фейком. """ from __future__ import annotations import re from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace from typing import Any import pytest from common.analytics import ( ALL_EVENT_TYPES, EventType, close_analytics, get_analytics, init_analytics, record, record_message, ) from common.analytics.client import PAYLOAD_MAX_LENGTH, AnalyticsClient from common.analytics.messages import describe_command, describe_message from common.config import Settings from common.version import __version__ _DOCS_PATH = Path(__file__).resolve().parents[1] / "docs" / "events.md" #: Строка каталога событий: первый столбец в обратных кавычках. _EVENT_ROW = re.compile(r"^\|\s*`([a-z_]+)`\s*\|", re.MULTILINE) class _FakeEnum: """Заглушка enum-полей aiogram (у значений есть ``.value``).""" def __init__(self, value: str) -> None: self.value = value class FakeWriter: """Заглушка записи в Postgres: складывает батчи в списки.""" def __init__(self, *, fail: bool = False) -> None: self.events: list[list[Any]] = [] self.messages: dict[tuple[Any, int, int], list[Any]] = {} self._fail = fail async def __call__(self, events: list[list[Any]], messages: dict[tuple[Any, int, int], list[Any]]) -> None: if self._fail: raise RuntimeError("Postgres недоступен") self.events.extend(events) for key, value in messages.items(): if key in self.messages: self.messages[key][1] += value[1] self.messages[key][0] = value[0] else: self.messages[key] = list(value) def _settings(**overrides: Any) -> Settings: return Settings(_env_file=None, bot_token="1:a", **overrides) @pytest.fixture(autouse=True) async def _clean_singleton(): """Тесты не должны оставлять активный singleton аналитики.""" await close_analytics() yield await close_analytics() def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeWriter]: fake = FakeWriter() return AnalyticsClient(_settings(**overrides), writer=fake), fake # --- каталог событий --- def test_event_types_are_unique_and_nonempty() -> None: assert len(ALL_EVENT_TYPES) == len(EventType) assert ALL_EVENT_TYPES def test_event_types_match_documentation() -> None: documented = set(_EVENT_ROW.findall(_DOCS_PATH.read_text(encoding="utf-8"))) assert documented == set(ALL_EVENT_TYPES), "Список событий в common/analytics/events.py и docs/events.md разошёлся" # --- метаданные сообщения (без aiogram) --- def _fake_message(**overrides: Any) -> Any: """Минимальная заглушка aiogram Message для describe_message/describe_command.""" fields: dict[str, Any] = { "message_id": 77, "text": "привет", "caption": None, "content_type": _FakeEnum("text"), "chat": SimpleNamespace(type=_FakeEnum("supergroup")), } fields.update(overrides) return SimpleNamespace(**fields) def test_describe_message_without_text() -> None: payload = describe_message(_fake_message(text="привет")) assert payload == { "message_id": 77, "chat_type": "supergroup", "content_type": "text", "text_len": 6, "is_command": False, } def test_describe_message_uses_caption_for_media() -> None: payload = describe_message(_fake_message(text=None, caption="подпись", content_type=_FakeEnum("photo"))) assert payload["content_type"] == "photo" assert payload["text_len"] == len("подпись") assert payload["is_command"] is False def test_describe_command_bare_and_with_args() -> None: assert describe_command(_fake_message(text="/очищение")) == {"command": "очищение", "args_count": 0} assert describe_command(_fake_message(text="/ник Вася")) == {"command": "ник", "args_count": 1} def test_describe_command_strips_bot_mention() -> None: assert describe_command(_fake_message(text="/Ник@DispelEvilBot Вася")) == {"command": "ник", "args_count": 1} def test_describe_command_returns_none_for_plain_message() -> None: assert describe_command(_fake_message(text="просто текст")) is None assert describe_command(_fake_message(text=None, caption="подпись")) is None # --- буферизация и запись --- async def test_record_without_init_is_noop() -> None: assert get_analytics() is None record(EventType.COMMAND, user_id=1, chat_id=2, payload={"a": 1}) # не должно бросать record_message(user_id=1, username="vasya", chat_id=2) async def test_client_buffers_and_flushes_rows() -> None: client, fake = _client() client.record(EventType.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "инфо"}) client.record(EventType.PURIFICATION, user_id=11, username="vasya", chat_id=100, payload={"success": True}) assert client.pending == 2 await client.flush() assert client.pending == 0 assert len(fake.events) == 2 row = fake.events[0] assert isinstance(row[0], datetime) assert row[1] == "command" assert row[2] == 11 assert row[3] == "vasya" assert row[4] == 100 assert row[5] == {"command": "инфо"} assert row[6] == __version__ assert row[7] is False async def test_message_counter_is_aggregated_per_day() -> None: client, fake = _client() client.record_message(user_id=11, username="vasya", chat_id=100) client.record_message(user_id=11, username="vasya", chat_id=100) client.record_message(user_id=12, username=None, chat_id=100) assert client.pending == 3 await client.flush() assert client.pending == 0 assert len(fake.messages) == 2 vasya = next(value for key, value in fake.messages.items() if key[1] == 11) assert vasya == ["vasya", 2] async def test_remember_user_enriches_username() -> None: client, fake = _client() client.remember_user(42, "nickname") client.record(EventType.PURIFICATION, user_id=42, chat_id=1) await client.flush() assert fake.events[0][3] == "nickname" async def test_record_triggers_flush_at_batch_size() -> None: client, _ = _client(analytics_batch_size=3) client.record(EventType.COMMAND, user_id=1) client.record(EventType.COMMAND, user_id=1) assert not client._flush_now.is_set() client.record(EventType.COMMAND, user_id=1) assert client._flush_now.is_set() async def test_overflow_drops_oldest_events() -> None: client, _ = _client(analytics_buffer_max=2, analytics_batch_size=100) for user_id in range(1, 5): client.record(EventType.COMMAND, user_id=user_id) assert client.pending == 2 assert [row[2] for row in client._rows] == [3, 4] async def test_disabled_analytics_ignores_events() -> None: client, fake = _client(analytics_enabled=False) client.record(EventType.COMMAND, user_id=1) client.record_message(user_id=1, username="vasya", chat_id=1) await client.flush() assert client.pending == 0 assert fake.events == [] assert fake.messages == {} # --- деградация без Postgres --- async def test_flush_failure_logs_fallback_and_drops(caplog: pytest.LogCaptureFixture) -> None: fake = FakeWriter(fail=True) client = AnalyticsClient(_settings(), writer=fake) client.record(EventType.COMMAND, user_id=1, username="vasya", chat_id=100) with caplog.at_level("WARNING"): await client.flush() # не должно бросать assert client.pending == 0 assert any("Fallback" in message for message in caplog.messages) async def test_payload_is_truncated() -> None: client, fake = _client() client.record(EventType.COMMAND, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)}) await client.flush() payload = fake.events[0][5] assert "_truncated" in payload assert len(payload["_truncated"]) == PAYLOAD_MAX_LENGTH # --- жизненный цикл --- async def test_init_and_close_flush_pending_events() -> None: fake = FakeWriter() client = await init_analytics(_settings(), writer=fake) record(EventType.BOT_START, payload={"app_env": "dev"}) record(EventType.BOT_STOP, payload={"app_env": "dev"}) assert get_analytics() is client assert client.pending == 2 await close_analytics() assert get_analytics() is None assert len(fake.events) == 2 # --- middleware (транспортные события) --- def _message(**overrides: Any) -> Any: """Реальное сообщение aiogram для проверки middleware.""" from aiogram.types import Chat, Message, User fields: dict[str, Any] = { "message_id": 5, "date": datetime.now(UTC), "chat": Chat(id=-100, type="supergroup", title="Клан"), "from_user": User(id=7, is_bot=False, first_name="Вася", username="vasya"), "text": "/инфо", } fields.update(overrides) return Message(**fields) async def _handler(event: Any, data: dict[str, Any]) -> str: return "handled" async def test_middleware_counts_message_and_records_command(monkeypatch: pytest.MonkeyPatch) -> None: from bot.middlewares.analytics import AnalyticsMiddleware settings = _settings(chat_id=-100) monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings) fake = FakeWriter() client = await init_analytics(settings, writer=fake) assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled" await client.flush() assert [row[1] for row in fake.events] == ["command"] assert fake.events[0][5] == {"command": "инфо", "args_count": 0} assert fake.events[0][3] == "vasya" assert len(fake.messages) == 1 assert next(iter(fake.messages.values())) == ["vasya", 1] async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyPatch) -> None: from bot.middlewares.analytics import AnalyticsMiddleware settings = _settings(chat_id=-100) monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings) fake = FakeWriter() client = await init_analytics(settings, writer=fake) from aiogram.types import Chat, User await AnalyticsMiddleware()(_handler, _message(chat=Chat(id=-999, type="supergroup")), {}) await AnalyticsMiddleware()( _handler, _message(from_user=User(id=7, is_bot=True, first_name="Bot")), {}, ) await client.flush() assert fake.events == [] assert fake.messages == {} async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) -> None: from bot.middlewares.analytics import AnalyticsMiddleware settings = _settings(chat_id=-100) monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings) def boom(*args: Any, **kwargs: Any) -> None: raise RuntimeError("аналитика сломалась") monkeypatch.setattr("bot.middlewares.analytics.record_message", boom) assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"