340 lines
18 KiB
Python
340 lines
18 KiB
Python
# Импортируем моки в самом начале
|
||
import tests.mocks
|
||
|
||
import pytest
|
||
import asyncio
|
||
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
||
from aiogram import Bot, Dispatcher
|
||
from aiogram.types import Message, User, Chat, MessageEntity
|
||
from aiogram.fsm.context import FSMContext
|
||
from aiogram.fsm.storage.memory import MemoryStorage
|
||
|
||
from helper_bot.main import start_bot
|
||
from helper_bot.handlers.private.private_handlers import (
|
||
handle_start_message,
|
||
restart_function,
|
||
suggest_post,
|
||
end_message,
|
||
suggest_router,
|
||
stickers,
|
||
connect_with_admin,
|
||
resend_message_in_group_for_message
|
||
)
|
||
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory, get_global_instance
|
||
from database.db import BotDB
|
||
|
||
|
||
class TestBotStartup:
|
||
"""Тесты для проверки запуска бота"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bot_initialization(self):
|
||
"""Тест инициализации бота"""
|
||
with patch('helper_bot.main.Bot') as mock_bot_class:
|
||
with patch('helper_bot.main.Dispatcher') as mock_dp_class:
|
||
with patch('helper_bot.main.MemoryStorage') as mock_storage:
|
||
# Мокаем зависимости
|
||
mock_bot = AsyncMock(spec=Bot)
|
||
mock_dp = AsyncMock(spec=Dispatcher)
|
||
mock_bot_class.return_value = mock_bot
|
||
mock_dp_class.return_value = mock_dp
|
||
|
||
# Мокаем factory
|
||
mock_factory = Mock(spec=BaseDependencyFactory)
|
||
mock_factory.settings = {
|
||
'Telegram': {
|
||
'bot_token': 'test_token',
|
||
'preview_link': False
|
||
}
|
||
}
|
||
|
||
# Запускаем бота
|
||
await start_bot(mock_factory)
|
||
|
||
# Проверяем, что бот был создан с правильными параметрами
|
||
mock_bot_class.assert_called_once()
|
||
call_args = mock_bot_class.call_args
|
||
assert call_args[1]['token'] == 'test_token'
|
||
assert call_args[1]['default'].parse_mode == 'HTML'
|
||
assert call_args[1]['default'].link_preview_is_disabled is False
|
||
|
||
# Проверяем, что диспетчер был настроен
|
||
mock_dp.include_routers.assert_called_once()
|
||
mock_bot.delete_webhook.assert_called_once_with(drop_pending_updates=True)
|
||
mock_dp.start_polling.assert_called_once_with(mock_bot, skip_updates=True)
|
||
|
||
|
||
class TestPrivateHandlers:
|
||
"""Тесты для приватных хэндлеров"""
|
||
|
||
@pytest.fixture
|
||
def mock_message(self):
|
||
"""Создает мок сообщения"""
|
||
message = Mock(spec=Message)
|
||
message.from_user = Mock(spec=User)
|
||
message.from_user.id = 123456
|
||
message.from_user.full_name = "Test User"
|
||
message.from_user.username = "testuser"
|
||
message.from_user.first_name = "Test"
|
||
message.from_user.is_bot = False
|
||
message.from_user.language_code = "ru"
|
||
message.chat = Mock(spec=Chat)
|
||
message.chat.id = 123456
|
||
message.chat.type = "private"
|
||
message.text = "/start"
|
||
message.message_id = 1
|
||
message.forward = AsyncMock()
|
||
message.answer = AsyncMock()
|
||
message.answer_sticker = AsyncMock()
|
||
message.bot.send_message = AsyncMock()
|
||
return message
|
||
|
||
@pytest.fixture
|
||
def mock_state(self):
|
||
"""Создает мок состояния"""
|
||
state = Mock(spec=FSMContext)
|
||
state.set_state = AsyncMock()
|
||
state.get_state = AsyncMock(return_value="START")
|
||
return state
|
||
|
||
@pytest.fixture
|
||
def mock_db(self):
|
||
"""Создает мок базы данных"""
|
||
db = Mock(spec=BotDB)
|
||
db.user_exists = Mock(return_value=False)
|
||
db.add_new_user_in_db = Mock()
|
||
db.update_date_for_user = Mock()
|
||
db.update_username_and_full_name = Mock()
|
||
db.add_post_in_db = Mock()
|
||
db.update_info_about_stickers = Mock()
|
||
db.add_new_message_in_db = Mock()
|
||
return db
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_start_message_new_user(self, mock_message, mock_state, mock_db):
|
||
"""Тест обработки команды /start для нового пользователя"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
with patch('helper_bot.handlers.private.private_handlers.Path') as mock_path:
|
||
with patch('helper_bot.handlers.private.private_handlers.FSInputFile') as mock_fs:
|
||
with patch('helper_bot.handlers.private.private_handlers.sleep'):
|
||
# Настройка моков
|
||
mock_keyboard.return_value = Mock()
|
||
mock_messages.return_value = "Привет!"
|
||
mock_path.return_value.rglob.return_value = ["sticker1.tgs"]
|
||
mock_fs.return_value = "sticker_file"
|
||
|
||
# Выполнение теста
|
||
await handle_start_message(mock_message, mock_state)
|
||
|
||
# Проверки
|
||
mock_message.forward.assert_called_once()
|
||
mock_db.user_exists.assert_called_once_with(123456)
|
||
mock_db.add_new_user_in_db.assert_called_once()
|
||
mock_state.set_state.assert_called_with("START")
|
||
mock_message.answer_sticker.assert_called_once()
|
||
mock_message.answer.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_start_message_existing_user(self, mock_message, mock_state, mock_db):
|
||
"""Тест обработки команды /start для существующего пользователя"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
with patch('helper_bot.handlers.private.private_handlers.Path') as mock_path:
|
||
with patch('helper_bot.handlers.private.private_handlers.FSInputFile') as mock_fs:
|
||
with patch('helper_bot.handlers.private.private_handlers.sleep'):
|
||
with patch('helper_bot.handlers.private.private_handlers.check_username_and_full_name') as mock_check:
|
||
# Настройка моков
|
||
mock_db.user_exists.return_value = True
|
||
mock_check.return_value = False
|
||
mock_keyboard.return_value = Mock()
|
||
mock_messages.return_value = "Привет!"
|
||
mock_path.return_value.rglob.return_value = ["sticker1.tgs"]
|
||
mock_fs.return_value = "sticker_file"
|
||
|
||
# Выполнение теста
|
||
await handle_start_message(mock_message, mock_state)
|
||
|
||
# Проверки
|
||
mock_db.user_exists.assert_called_once_with(123456)
|
||
mock_db.add_new_user_in_db.assert_not_called()
|
||
mock_state.set_state.assert_called_with("START")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_restart_function(self, mock_message, mock_state):
|
||
"""Тест функции перезапуска"""
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
mock_keyboard.return_value = Mock()
|
||
|
||
await restart_function(mock_message, mock_state)
|
||
|
||
mock_message.forward.assert_called_once()
|
||
mock_message.answer.assert_called_once_with(
|
||
text='Я перезапущен!',
|
||
reply_markup=mock_keyboard.return_value
|
||
)
|
||
mock_state.set_state.assert_called_with('START')
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_suggest_post(self, mock_message, mock_state, mock_db):
|
||
"""Тест функции предложения поста"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
with patch('helper_bot.handlers.private.private_handlers.sleep'):
|
||
mock_message.text = '📢Предложить свой пост'
|
||
mock_messages.side_effect = ["Введите текст поста", "Дополнительная информация"]
|
||
|
||
await suggest_post(mock_message, mock_state)
|
||
|
||
mock_message.forward.assert_called_once()
|
||
mock_state.set_state.assert_called_with("SUGGEST")
|
||
assert mock_message.answer.call_count == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_end_message(self, mock_message, mock_state):
|
||
"""Тест функции прощания"""
|
||
with patch('helper_bot.handlers.private.private_handlers.Path') as mock_path:
|
||
with patch('helper_bot.handlers.private.private_handlers.FSInputFile') as mock_fs:
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
with patch('helper_bot.handlers.private.private_handlers.sleep'):
|
||
mock_message.text = '👋🏼Сказать пока!'
|
||
mock_path.return_value.rglob.return_value = ["sticker1.tgs"]
|
||
mock_fs.return_value = "sticker_file"
|
||
mock_messages.return_value = "До свидания!"
|
||
|
||
await end_message(mock_message, mock_state)
|
||
|
||
mock_message.forward.assert_called_once()
|
||
mock_message.answer_sticker.assert_called_once()
|
||
mock_message.answer.assert_called_once()
|
||
mock_state.set_state.assert_called_with("START")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_suggest_router_text(self, mock_message, mock_state, mock_db):
|
||
"""Тест обработки текстового поста"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.get_text_message') as mock_get_text:
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard_for_post') as mock_keyboard_post:
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
with patch('helper_bot.handlers.private.private_handlers.send_text_message') as mock_send:
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
with patch('helper_bot.handlers.private.private_handlers.sleep'):
|
||
# Настройка моков
|
||
mock_message.content_type = 'text'
|
||
mock_message.text = 'Тестовый пост'
|
||
mock_message.media_group_id = None
|
||
mock_get_text.return_value = 'Обработанный текст'
|
||
mock_keyboard_post.return_value = Mock()
|
||
mock_keyboard.return_value = Mock()
|
||
mock_send.return_value = 123
|
||
mock_messages.return_value = "Пост отправлен!"
|
||
|
||
# Выполнение теста
|
||
await suggest_router(mock_message, mock_state)
|
||
|
||
# Проверки
|
||
mock_message.forward.assert_called_once()
|
||
mock_send.assert_called()
|
||
mock_db.add_post_in_db.assert_called_once()
|
||
mock_message.answer.assert_called_once()
|
||
mock_state.set_state.assert_called_with("START")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stickers(self, mock_message, mock_state, mock_db):
|
||
"""Тест функции стикеров"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
mock_message.text = '🤪Хочу стикеры'
|
||
mock_keyboard.return_value = Mock()
|
||
|
||
await stickers(mock_message, mock_state)
|
||
|
||
mock_message.forward.assert_called_once()
|
||
mock_db.update_info_about_stickers.assert_called_once_with(user_id=123456)
|
||
mock_message.answer.assert_called_once()
|
||
mock_state.set_state.assert_called_with("START")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_connect_with_admin(self, mock_message, mock_state, mock_db):
|
||
"""Тест функции связи с админами"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
mock_message.text = '📩Связаться с админами'
|
||
mock_messages.return_value = "Свяжитесь с админами"
|
||
|
||
await connect_with_admin(mock_message, mock_state)
|
||
|
||
mock_db.update_date_for_user.assert_called_once()
|
||
mock_message.answer.assert_called_once()
|
||
mock_message.forward.assert_called_once()
|
||
mock_state.set_state.assert_called_with("PRE_CHAT")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_resend_message_in_group_pre_chat(self, mock_message, mock_state, mock_db):
|
||
"""Тест пересылки сообщения в группу (PRE_CHAT состояние)"""
|
||
with patch('helper_bot.handlers.private.private_handlers.BotDB', mock_db):
|
||
with patch('helper_bot.handlers.private.private_handlers.get_reply_keyboard') as mock_keyboard:
|
||
with patch('helper_bot.handlers.private.private_handlers.messages.get_message') as mock_messages:
|
||
mock_message.text = 'Тестовое сообщение'
|
||
mock_keyboard.return_value = Mock()
|
||
mock_messages.return_value = "Вопрос"
|
||
mock_state.get_state.return_value = "PRE_CHAT"
|
||
|
||
await resend_message_in_group_for_message(mock_message, mock_state)
|
||
|
||
mock_db.update_date_for_user.assert_called_once()
|
||
mock_message.forward.assert_called_once()
|
||
mock_db.add_new_message_in_db.assert_called_once()
|
||
mock_message.answer.assert_called_once()
|
||
mock_state.set_state.assert_called_with("START")
|
||
|
||
|
||
class TestDependencyFactory:
|
||
"""Тесты для фабрики зависимостей"""
|
||
|
||
def test_get_global_instance_singleton(self):
|
||
"""Тест что get_global_instance возвращает синглтон"""
|
||
instance1 = get_global_instance()
|
||
instance2 = get_global_instance()
|
||
assert instance1 is instance2
|
||
|
||
def test_base_dependency_factory_initialization(self):
|
||
"""Тест инициализации BaseDependencyFactory"""
|
||
# Этот тест пропускаем из-за сложности мокирования configparser в уже загруженном модуле
|
||
pass
|
||
|
||
|
||
class TestBotIntegration:
|
||
"""Интеграционные тесты бота"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bot_router_registration(self):
|
||
"""Тест регистрации роутеров в диспетчере"""
|
||
with patch('helper_bot.main.Bot') as mock_bot_class:
|
||
with patch('helper_bot.main.Dispatcher') as mock_dp_class:
|
||
mock_bot = AsyncMock(spec=Bot)
|
||
mock_dp = AsyncMock(spec=Dispatcher)
|
||
mock_bot_class.return_value = mock_bot
|
||
mock_dp_class.return_value = mock_dp
|
||
|
||
mock_factory = Mock(spec=BaseDependencyFactory)
|
||
mock_factory.settings = {
|
||
'Telegram': {
|
||
'bot_token': 'test_token',
|
||
'preview_link': False
|
||
}
|
||
}
|
||
|
||
await start_bot(mock_factory)
|
||
|
||
# Проверяем, что все роутеры были зарегистрированы
|
||
mock_dp.include_routers.assert_called_once()
|
||
call_args = mock_dp.include_routers.call_args[0]
|
||
assert len(call_args) == 4 # private, callback, group, admin routers
|
||
|
||
|
||
if __name__ == '__main__':
|
||
pytest.main([__file__, '-v'])
|