352 lines
11 KiB
Python
352 lines
11 KiB
Python
"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse.
|
||
|
||
Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а
|
||
``clickhouse_connect`` импортируется лениво внутри клиента.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
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
|
||
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 FakeClickHouse:
|
||
"""Заглушка клиента ClickHouse: пишет в список или падает."""
|
||
|
||
def __init__(self, *, fail: bool = False) -> None:
|
||
self.rows: list[list[Any]] = []
|
||
self.table: str | None = None
|
||
self.columns: list[str] | None = None
|
||
self.closed = False
|
||
self._fail = fail
|
||
|
||
async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
|
||
if self._fail:
|
||
raise RuntimeError("ClickHouse недоступен")
|
||
self.table = table
|
||
self.columns = list(column_names)
|
||
self.rows.extend(data)
|
||
|
||
async def close(self) -> None:
|
||
self.closed = True
|
||
|
||
|
||
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, FakeClickHouse]:
|
||
fake = FakeClickHouse()
|
||
return AnalyticsClient(_settings(**overrides), client=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.MESSAGE, user_id=1, chat_id=2, payload={"a": 1}) # не должно бросать
|
||
|
||
|
||
async def test_client_buffers_and_flushes_rows() -> None:
|
||
client, fake = _client()
|
||
client.record(EventType.MESSAGE, user_id=11, username="vasya", chat_id=100, payload={"message_id": 5})
|
||
client.record(EventType.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "/инфо"})
|
||
|
||
assert client.pending == 2
|
||
|
||
await client.flush()
|
||
|
||
assert client.pending == 0
|
||
assert fake.table == "analytics.events"
|
||
assert fake.columns == [
|
||
"event_time",
|
||
"event_type",
|
||
"user_id",
|
||
"username",
|
||
"chat_id",
|
||
"payload",
|
||
"bot_version",
|
||
"prod",
|
||
]
|
||
assert len(fake.rows) == 2
|
||
|
||
row = fake.rows[0]
|
||
assert isinstance(row[0], datetime)
|
||
assert row[1] == "message"
|
||
assert row[2] == 11
|
||
assert row[3] == "vasya"
|
||
assert row[4] == 100
|
||
assert '"message_id": 5' in row[5]
|
||
assert row[6] == __version__
|
||
assert row[7] == 0
|
||
|
||
|
||
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.rows[0][3] == "nickname"
|
||
|
||
|
||
async def test_record_triggers_flush_at_batch_size() -> None:
|
||
client, _ = _client(analytics_batch_size=3)
|
||
|
||
client.record(EventType.MESSAGE, user_id=1)
|
||
client.record(EventType.MESSAGE, user_id=1)
|
||
assert not client._flush_now.is_set()
|
||
|
||
client.record(EventType.MESSAGE, 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.MESSAGE, 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(clickhouse_enabled=False)
|
||
|
||
client.record(EventType.MESSAGE, user_id=1)
|
||
await client.flush()
|
||
|
||
assert client.pending == 0
|
||
assert fake.rows == []
|
||
|
||
|
||
# --- деградация без ClickHouse ---
|
||
|
||
|
||
async def test_flush_failure_logs_fallback_and_drops(caplog: pytest.LogCaptureFixture) -> None:
|
||
fake = FakeClickHouse(fail=True)
|
||
client = AnalyticsClient(_settings(), client=fake)
|
||
client.record(EventType.MESSAGE, user_id=1, username="vasya", chat_id=100)
|
||
|
||
with caplog.at_level(logging.WARNING):
|
||
await client.flush() # не должно бросать
|
||
|
||
assert client.pending == 0
|
||
assert any("Fallback" in message for message in caplog.messages)
|
||
|
||
|
||
async def test_unavailable_client_falls_back_to_log(caplog: pytest.LogCaptureFixture) -> None:
|
||
settings = _settings(clickhouse_host="127.0.0.1", clickhouse_port=1, clickhouse_password="x")
|
||
client = AnalyticsClient(settings)
|
||
client.record(EventType.BOT_START, payload={"app_env": "dev"})
|
||
|
||
with caplog.at_level(logging.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.MESSAGE, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)})
|
||
|
||
await client.flush()
|
||
|
||
assert len(fake.rows[0][5]) == PAYLOAD_MAX_LENGTH
|
||
|
||
|
||
# --- жизненный цикл ---
|
||
|
||
|
||
async def test_init_and_close_flush_pending_events() -> None:
|
||
fake = FakeClickHouse()
|
||
client = await init_analytics(_settings(), client=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.rows) == 2
|
||
assert fake.closed is True
|
||
|
||
|
||
# --- 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_records_message_and_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 = FakeClickHouse()
|
||
client = await init_analytics(settings, client=fake)
|
||
|
||
assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"
|
||
await client.flush()
|
||
|
||
assert [row[1] for row in fake.rows] == ["message", "command"]
|
||
assert json.loads(fake.rows[0][5])["is_command"] is True
|
||
assert json.loads(fake.rows[1][5]) == {"command": "инфо", "args_count": 0}
|
||
assert fake.rows[0][3] == "vasya"
|
||
|
||
|
||
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 = FakeClickHouse()
|
||
client = await init_analytics(settings, client=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.rows == []
|
||
|
||
|
||
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", boom)
|
||
|
||
assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"
|