376 lines
15 KiB
Python
376 lines
15 KiB
Python
"""
|
||
Тесты для helper_bot.middlewares.metrics_middleware.
|
||
"""
|
||
|
||
import time
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
from aiogram.types import Message
|
||
|
||
from helper_bot.middlewares.metrics_middleware import (
|
||
DatabaseMetricsMiddleware,
|
||
ErrorMetricsMiddleware,
|
||
MetricsMiddleware,
|
||
)
|
||
|
||
|
||
@pytest.mark.unit
|
||
@pytest.mark.asyncio
|
||
class TestMetricsMiddleware:
|
||
"""Тесты для MetricsMiddleware."""
|
||
|
||
@pytest.fixture
|
||
def middleware(self):
|
||
"""Экземпляр middleware с отключённым периодическим обновлением активных пользователей."""
|
||
m = MetricsMiddleware()
|
||
m.last_active_users_update = time.time()
|
||
return m
|
||
|
||
@pytest.fixture
|
||
def mock_handler(self):
|
||
"""Мок handler."""
|
||
|
||
async def sample_handler(event, data):
|
||
return "result"
|
||
|
||
sample_handler.__name__ = "sample_handler"
|
||
return sample_handler
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_handler_success_records_metrics(
|
||
self, mock_metrics, middleware, mock_handler
|
||
):
|
||
"""При успешном выполнении handler вызываются record_method_duration и record_middleware."""
|
||
event = MagicMock(spec=Message)
|
||
event.message = None
|
||
event.callback_query = None
|
||
event.text = "привет"
|
||
event.from_user = MagicMock()
|
||
event.chat = MagicMock()
|
||
event.chat.type = "private"
|
||
event.photo = None
|
||
event.video = None
|
||
event.audio = None
|
||
event.document = None
|
||
event.voice = None
|
||
event.sticker = None
|
||
event.animation = None
|
||
data = {}
|
||
|
||
result = await middleware(mock_handler, event, data)
|
||
|
||
assert result == "result"
|
||
mock_metrics.record_method_duration.assert_called()
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
call_args = mock_metrics.record_middleware.call_args[0]
|
||
assert call_args[0] == "MetricsMiddleware"
|
||
assert call_args[2] == "success"
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_handler_exception_records_error_and_reraises(
|
||
self, mock_metrics, middleware
|
||
):
|
||
"""При исключении в handler записываются метрики ошибки и исключение пробрасывается."""
|
||
|
||
async def failing_handler(event, data):
|
||
raise ValueError("test error")
|
||
|
||
failing_handler.__name__ = "failing_handler"
|
||
|
||
event = MagicMock(spec=Message)
|
||
event.message = None
|
||
event.callback_query = None
|
||
event.text = "text"
|
||
event.from_user = MagicMock()
|
||
event.chat = MagicMock()
|
||
event.chat.type = "private"
|
||
event.photo = None
|
||
event.video = None
|
||
event.audio = None
|
||
event.document = None
|
||
event.voice = None
|
||
event.sticker = None
|
||
event.animation = None
|
||
data = {}
|
||
|
||
with pytest.raises(ValueError, match="test error"):
|
||
await middleware(failing_handler, event, data)
|
||
|
||
mock_metrics.record_error.assert_called_once()
|
||
call_args = mock_metrics.record_error.call_args[0]
|
||
assert call_args[0] == "ValueError"
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
|
||
def test_get_handler_name_returns_function_name(self, middleware):
|
||
"""_get_handler_name возвращает __name__ функции."""
|
||
|
||
def named_handler():
|
||
pass
|
||
|
||
assert middleware._get_handler_name(named_handler) == "named_handler"
|
||
|
||
def test_get_handler_name_for_lambda_returns_qualname_or_unknown(self, middleware):
|
||
"""_get_handler_name для lambda возвращает qualname (содержит 'lambda') или 'unknown'."""
|
||
lambda_handler = lambda e, d: None
|
||
name = middleware._get_handler_name(lambda_handler)
|
||
assert "lambda" in name or name == "unknown"
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_record_comprehensive_message_metrics_photo(
|
||
self, mock_metrics, middleware
|
||
):
|
||
"""_record_comprehensive_message_metrics для сообщения с фото записывает message_type photo."""
|
||
message = MagicMock()
|
||
message.photo = [MagicMock()]
|
||
message.video = None
|
||
message.audio = None
|
||
message.document = None
|
||
message.voice = None
|
||
message.sticker = None
|
||
message.animation = None
|
||
message.chat = MagicMock()
|
||
message.chat.type = "private"
|
||
message.from_user = MagicMock()
|
||
message.from_user.id = 1
|
||
message.from_user.is_bot = False
|
||
|
||
result = await middleware._record_comprehensive_message_metrics(message)
|
||
|
||
mock_metrics.record_message.assert_called_once_with(
|
||
"photo", "private", "message_handler"
|
||
)
|
||
assert result["message_type"] == "photo"
|
||
assert result["chat_type"] == "private"
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_record_comprehensive_message_metrics_voice(
|
||
self, mock_metrics, middleware
|
||
):
|
||
"""_record_comprehensive_message_metrics для voice записывает message_type voice."""
|
||
message = MagicMock()
|
||
message.photo = None
|
||
message.video = None
|
||
message.audio = None
|
||
message.document = None
|
||
message.voice = MagicMock()
|
||
message.sticker = None
|
||
message.animation = None
|
||
message.chat = MagicMock()
|
||
message.chat.type = "supergroup"
|
||
message.from_user = MagicMock()
|
||
message.from_user.id = 2
|
||
message.from_user.is_bot = False
|
||
|
||
result = await middleware._record_comprehensive_message_metrics(message)
|
||
|
||
mock_metrics.record_message.assert_called_once_with(
|
||
"voice", "supergroup", "message_handler"
|
||
)
|
||
assert result["message_type"] == "voice"
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_record_comprehensive_callback_metrics(
|
||
self, mock_metrics, middleware
|
||
):
|
||
"""_record_comprehensive_callback_metrics записывает callback_query и возвращает данные."""
|
||
callback = MagicMock()
|
||
callback.data = "publish"
|
||
callback.from_user = MagicMock()
|
||
callback.from_user.id = 10
|
||
callback.from_user.is_bot = False
|
||
|
||
result = await middleware._record_comprehensive_callback_metrics(callback)
|
||
|
||
mock_metrics.record_message.assert_called_once_with(
|
||
"callback_query", "callback", "callback_handler"
|
||
)
|
||
assert result["callback_data"] == "publish"
|
||
assert result["user_id"] == 10
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_record_unknown_event_metrics(self, mock_metrics, middleware):
|
||
"""_record_unknown_event_metrics записывает unknown и возвращает event_type."""
|
||
event = MagicMock()
|
||
event.__str__ = lambda self: "custom_event"
|
||
|
||
result = await middleware._record_unknown_event_metrics(event)
|
||
|
||
mock_metrics.record_message.assert_called_once_with(
|
||
"unknown", "unknown", "unknown_handler"
|
||
)
|
||
assert "event_type" in result
|
||
|
||
def test_extract_command_info_slash_command_returns_mapping(self, middleware):
|
||
"""_extract_command_info_with_fallback для слеш-команды возвращает command info."""
|
||
message = MagicMock()
|
||
message.text = "/start"
|
||
message.from_user = MagicMock()
|
||
result = middleware._extract_command_info_with_fallback(message)
|
||
assert result is not None
|
||
assert "command" in result
|
||
assert "handler_type" in result
|
||
|
||
def test_extract_command_info_no_text_returns_none(self, middleware):
|
||
"""_extract_command_info_with_fallback при отсутствии text возвращает None."""
|
||
message = MagicMock()
|
||
message.text = None
|
||
result = middleware._extract_command_info_with_fallback(message)
|
||
assert result is None
|
||
|
||
def test_extract_command_info_empty_string_returns_none(self, middleware):
|
||
"""_extract_command_info_with_fallback при пустом text возвращает None или fallback."""
|
||
message = MagicMock()
|
||
message.text = ""
|
||
message.from_user = None
|
||
result = middleware._extract_command_info_with_fallback(message)
|
||
assert result is None or (result is not None and "command" in result)
|
||
|
||
def test_extract_callback_command_info_no_data_returns_none(self, middleware):
|
||
"""_extract_callback_command_info_with_fallback при отсутствии data возвращает None."""
|
||
callback = MagicMock()
|
||
callback.data = None
|
||
callback.from_user = MagicMock()
|
||
result = middleware._extract_callback_command_info_with_fallback(callback)
|
||
assert result is None
|
||
|
||
def test_extract_callback_command_info_ban_pattern_returns_callback_ban(
|
||
self, middleware
|
||
):
|
||
"""_extract_callback_command_info_with_fallback для ban_123 возвращает callback_ban."""
|
||
callback = MagicMock()
|
||
callback.data = "ban_123456"
|
||
callback.from_user = MagicMock()
|
||
result = middleware._extract_callback_command_info_with_fallback(callback)
|
||
assert result is not None
|
||
assert result["command"] == "callback_ban" or "ban" in result["command"]
|
||
assert "handler_type" in result
|
||
|
||
def test_extract_callback_command_info_page_pattern_returns_callback_page(
|
||
self, middleware
|
||
):
|
||
"""_extract_callback_command_info_with_fallback для page_2 возвращает callback_page."""
|
||
callback = MagicMock()
|
||
callback.data = "page_2"
|
||
callback.from_user = MagicMock()
|
||
result = middleware._extract_callback_command_info_with_fallback(callback)
|
||
assert result is not None
|
||
assert result["command"] == "callback_page" or "page" in result["command"]
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
@patch("helper_bot.utils.base_dependency_factory.get_global_instance")
|
||
async def test_update_active_users_metric_sets_metrics(
|
||
self, mock_get_global, mock_metrics, middleware
|
||
):
|
||
"""_update_active_users_metric вызывает fetch_one и устанавливает метрики."""
|
||
mock_bdf = MagicMock()
|
||
mock_db = MagicMock()
|
||
mock_db.fetch_one = AsyncMock(
|
||
side_effect=[
|
||
{"total": 100},
|
||
{"daily": 10},
|
||
]
|
||
)
|
||
mock_bdf.get_db.return_value = mock_db
|
||
mock_get_global.return_value = mock_bdf
|
||
|
||
await middleware._update_active_users_metric()
|
||
|
||
assert mock_metrics.set_active_users.called
|
||
assert mock_metrics.set_total_users.called
|
||
mock_metrics.set_active_users.assert_called_with(10, "daily")
|
||
mock_metrics.set_total_users.assert_called_with(100)
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
@patch("helper_bot.utils.base_dependency_factory.get_global_instance")
|
||
async def test_update_active_users_metric_on_exception_sets_fallback(
|
||
self, mock_get_global, mock_metrics, middleware
|
||
):
|
||
"""_update_active_users_metric при исключении устанавливает fallback 1."""
|
||
mock_get_global.side_effect = RuntimeError("no bdf")
|
||
|
||
await middleware._update_active_users_metric()
|
||
|
||
mock_metrics.set_active_users.assert_called_with(1, "daily")
|
||
mock_metrics.set_total_users.assert_called_with(1)
|
||
|
||
|
||
@pytest.mark.unit
|
||
@pytest.mark.asyncio
|
||
class TestDatabaseMetricsMiddleware:
|
||
"""Тесты для DatabaseMetricsMiddleware."""
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_success_records_middleware(self, mock_metrics):
|
||
"""При успешном handler вызывается record_middleware с success."""
|
||
middleware = DatabaseMetricsMiddleware()
|
||
handler = AsyncMock(return_value="ok")
|
||
handler.__name__ = "test_handler"
|
||
event = MagicMock()
|
||
data = {}
|
||
|
||
result = await middleware(handler, event, data)
|
||
|
||
assert result == "ok"
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
assert mock_metrics.record_middleware.call_args[0][2] == "success"
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_exception_records_error_and_reraises(self, mock_metrics):
|
||
"""При исключении записывается ошибка и исключение пробрасывается."""
|
||
middleware = DatabaseMetricsMiddleware()
|
||
handler = AsyncMock(side_effect=RuntimeError("db error"))
|
||
handler.__name__ = "db_handler"
|
||
event = MagicMock()
|
||
data = {}
|
||
|
||
with pytest.raises(RuntimeError, match="db error"):
|
||
await middleware(handler, event, data)
|
||
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
assert mock_metrics.record_middleware.call_args[0][2] == "error"
|
||
mock_metrics.record_error.assert_called_once()
|
||
|
||
|
||
@pytest.mark.unit
|
||
@pytest.mark.asyncio
|
||
class TestErrorMetricsMiddleware:
|
||
"""Тесты для ErrorMetricsMiddleware."""
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_success_records_middleware(self, mock_metrics):
|
||
"""При успешном handler вызывается record_middleware с success."""
|
||
middleware = ErrorMetricsMiddleware()
|
||
handler = AsyncMock(return_value="ok")
|
||
handler.__name__ = "test_handler"
|
||
event = MagicMock()
|
||
data = {}
|
||
|
||
result = await middleware(handler, event, data)
|
||
|
||
assert result == "ok"
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
assert mock_metrics.record_middleware.call_args[0][2] == "success"
|
||
|
||
@patch("helper_bot.middlewares.metrics_middleware.metrics")
|
||
async def test_exception_records_error_and_reraises(self, mock_metrics):
|
||
"""При исключении записывается ошибка и исключение пробрасывается."""
|
||
middleware = ErrorMetricsMiddleware()
|
||
handler = AsyncMock(side_effect=TypeError("error"))
|
||
handler.__name__ = "err_handler"
|
||
event = MagicMock()
|
||
data = {}
|
||
|
||
with pytest.raises(TypeError, match="error"):
|
||
await middleware(handler, event, data)
|
||
|
||
mock_metrics.record_middleware.assert_called_once()
|
||
assert mock_metrics.record_middleware.call_args[0][2] == "error"
|
||
mock_metrics.record_error.assert_called_once()
|