Andrey
c8c7d50cbb
- Removed the MetricsManager initialization from `run_helper.py` to avoid duplication, as metrics are now handled in `main.py`. - Updated logging levels in `server_prometheus.py` and `metrics_middleware.py` to use debug instead of info for less critical messages. - Added metrics configuration to `BaseDependencyFactory` for better management of metrics settings. - Deleted the obsolete `metrics_exporter.py` file to streamline the codebase. - Updated various tests to reflect changes in the metrics handling and ensure proper functionality.
196 lines
7.2 KiB
Python
196 lines
7.2 KiB
Python
"""Tests for refactored group handlers"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, MagicMock
|
|
from aiogram import types
|
|
from aiogram.fsm.context import FSMContext
|
|
|
|
from helper_bot.handlers.group.group_handlers import (
|
|
create_group_handlers, GroupHandlers
|
|
)
|
|
from helper_bot.handlers.group.services import AdminReplyService
|
|
from helper_bot.handlers.group.exceptions import NoReplyToMessageError, UserNotFoundError
|
|
from helper_bot.handlers.group.constants import FSM_STATES, ERROR_MESSAGES
|
|
|
|
|
|
class TestGroupHandlers:
|
|
"""Test class for GroupHandlers"""
|
|
|
|
@pytest.fixture
|
|
def mock_db(self):
|
|
"""Mock database"""
|
|
db = Mock()
|
|
db.get_user_by_message_id = Mock()
|
|
return db
|
|
|
|
@pytest.fixture
|
|
def mock_keyboard_markup(self):
|
|
"""Mock keyboard markup"""
|
|
return Mock()
|
|
|
|
@pytest.fixture
|
|
def mock_message(self):
|
|
"""Mock Telegram message"""
|
|
message = Mock()
|
|
message.from_user = Mock()
|
|
message.from_user.id = 12345
|
|
message.from_user.full_name = "Test Admin"
|
|
message.text = "test reply message"
|
|
message.chat = Mock()
|
|
message.chat.title = "Test Group"
|
|
message.chat.id = 67890
|
|
message.message_id = 111
|
|
message.answer = AsyncMock()
|
|
message.bot = Mock()
|
|
message.bot.send_message = AsyncMock()
|
|
return message
|
|
|
|
@pytest.fixture
|
|
def mock_reply_message(self, mock_message):
|
|
"""Mock reply message"""
|
|
reply_message = Mock()
|
|
reply_message.message_id = 222
|
|
mock_message.reply_to_message = reply_message
|
|
return mock_message
|
|
|
|
@pytest.fixture
|
|
def mock_state(self):
|
|
"""Mock FSM state"""
|
|
state = Mock(spec=FSMContext)
|
|
state.set_state = AsyncMock()
|
|
return state
|
|
|
|
def test_create_group_handlers(self, mock_db, mock_keyboard_markup):
|
|
"""Test creating group handlers instance"""
|
|
handlers = create_group_handlers(mock_db, mock_keyboard_markup)
|
|
assert isinstance(handlers, GroupHandlers)
|
|
assert handlers.db == mock_db
|
|
assert handlers.keyboard_markup == mock_keyboard_markup
|
|
|
|
def test_group_handlers_initialization(self, mock_db, mock_keyboard_markup):
|
|
"""Test GroupHandlers initialization"""
|
|
handlers = GroupHandlers(mock_db, mock_keyboard_markup)
|
|
assert handlers.db == mock_db
|
|
assert handlers.keyboard_markup == mock_keyboard_markup
|
|
assert handlers.admin_reply_service is not None
|
|
assert handlers.router is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_message_success(self, mock_db, mock_keyboard_markup, mock_reply_message, mock_state):
|
|
"""Test successful message handling"""
|
|
mock_db.get_user_by_message_id = AsyncMock(return_value=99999)
|
|
|
|
handlers = create_group_handlers(mock_db, mock_keyboard_markup)
|
|
|
|
# Mock the send_reply_to_user method
|
|
handlers.admin_reply_service.send_reply_to_user = AsyncMock()
|
|
|
|
await handlers.handle_message(mock_reply_message, mock_state)
|
|
|
|
# Verify database call
|
|
mock_db.get_user_by_message_id.assert_called_once_with(222)
|
|
|
|
# Verify service call
|
|
handlers.admin_reply_service.send_reply_to_user.assert_called_once_with(
|
|
99999, mock_reply_message, "test reply message", mock_keyboard_markup
|
|
)
|
|
|
|
# Verify state was set
|
|
mock_state.set_state.assert_called_once_with(FSM_STATES["CHAT"])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_message_no_reply(self, mock_db, mock_keyboard_markup, mock_message, mock_state):
|
|
"""Test message handling without reply"""
|
|
handlers = create_group_handlers(mock_db, mock_keyboard_markup)
|
|
|
|
# Mock the send_reply_to_user method to prevent it from being called
|
|
handlers.admin_reply_service.send_reply_to_user = AsyncMock()
|
|
|
|
# Ensure reply_to_message is None
|
|
mock_message.reply_to_message = None
|
|
|
|
await handlers.handle_message(mock_message, mock_state)
|
|
|
|
# Verify error message was sent
|
|
mock_message.answer.assert_called_once_with(ERROR_MESSAGES["NO_REPLY_TO_MESSAGE"])
|
|
|
|
# Verify no database calls
|
|
mock_db.get_user_by_message_id.assert_not_called()
|
|
|
|
# Verify send_reply_to_user was not called
|
|
handlers.admin_reply_service.send_reply_to_user.assert_not_called()
|
|
|
|
# Verify state was not set
|
|
mock_state.set_state.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_message_user_not_found(self, mock_db, mock_keyboard_markup, mock_reply_message, mock_state):
|
|
"""Test message handling when user is not found"""
|
|
mock_db.get_user_by_message_id = AsyncMock(return_value=None)
|
|
|
|
handlers = create_group_handlers(mock_db, mock_keyboard_markup)
|
|
|
|
await handlers.handle_message(mock_reply_message, mock_state)
|
|
|
|
# Verify error message was sent
|
|
mock_reply_message.answer.assert_called_once_with(ERROR_MESSAGES["USER_NOT_FOUND"])
|
|
|
|
# Verify database call
|
|
mock_db.get_user_by_message_id.assert_called_once_with(222)
|
|
|
|
# Verify state was not set
|
|
mock_state.set_state.assert_not_called()
|
|
|
|
|
|
class TestAdminReplyService:
|
|
"""Test class for AdminReplyService"""
|
|
|
|
@pytest.fixture
|
|
def mock_db(self):
|
|
"""Mock database"""
|
|
db = Mock()
|
|
db.get_user_by_message_id = Mock()
|
|
return db
|
|
|
|
@pytest.fixture
|
|
def service(self, mock_db):
|
|
"""Create service instance"""
|
|
return AdminReplyService(mock_db)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_id_for_reply_success(self, service, mock_db):
|
|
"""Test successful user ID retrieval"""
|
|
mock_db.get_user_by_message_id = AsyncMock(return_value=12345)
|
|
|
|
result = await service.get_user_id_for_reply(111)
|
|
|
|
assert result == 12345
|
|
mock_db.get_user_by_message_id.assert_called_once_with(111)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_id_for_reply_not_found(self, service, mock_db):
|
|
"""Test user ID retrieval when user not found"""
|
|
mock_db.get_user_by_message_id = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(UserNotFoundError, match="User not found for message_id: 111"):
|
|
await service.get_user_id_for_reply(111)
|
|
|
|
mock_db.get_user_by_message_id.assert_called_once_with(111)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_reply_to_user(self, service, mock_db):
|
|
"""Test sending reply to user"""
|
|
message = Mock()
|
|
message.reply_to_message = Mock()
|
|
message.reply_to_message.message_id = 222
|
|
markup = Mock()
|
|
|
|
# Mock the send_text_message function
|
|
with pytest.MonkeyPatch().context() as m:
|
|
mock_send_text = AsyncMock()
|
|
m.setattr('helper_bot.handlers.group.services.send_text_message', mock_send_text)
|
|
|
|
await service.send_reply_to_user(12345, message, "test reply", markup)
|
|
|
|
mock_send_text.assert_called_once_with(12345, message, "test reply", markup)
|