Andrey
fc0517c011
- Added a new `/status` endpoint in `server_prometheus.py` to provide process status information, including uptime and resource usage metrics. - Implemented a PID manager in `run_helper.py` to track the bot's process, improving monitoring capabilities. - Introduced a method to delete audio moderation records in `audio_repository.py`, enhancing database management. - Updated voice message handling in callback handlers to ensure proper deletion of audio moderation records. - Improved error handling and logging in various services, ensuring better tracking of media processing and file downloads. - Refactored media handling functions to streamline operations and improve code readability. - Enhanced metrics tracking for file downloads and media processing, providing better insights into bot performance.
267 lines
12 KiB
Python
267 lines
12 KiB
Python
import pytest
|
||
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
||
from datetime import datetime
|
||
import time
|
||
|
||
from helper_bot.handlers.voice.services import AudioFileService
|
||
from helper_bot.handlers.voice.exceptions import FileOperationError, DatabaseError
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_bot_db():
|
||
"""Мок для базы данных"""
|
||
mock_db = Mock()
|
||
mock_db.get_user_audio_records_count = AsyncMock(return_value=0)
|
||
mock_db.get_path_for_audio_record = AsyncMock(return_value=None)
|
||
mock_db.add_audio_record_simple = AsyncMock()
|
||
return mock_db
|
||
|
||
@pytest.fixture
|
||
def audio_service(mock_bot_db):
|
||
"""Экземпляр AudioFileService для тестов"""
|
||
return AudioFileService(mock_bot_db)
|
||
|
||
@pytest.fixture
|
||
def sample_datetime():
|
||
"""Тестовая дата"""
|
||
return datetime(2025, 1, 15, 14, 30, 0)
|
||
|
||
@pytest.fixture
|
||
def mock_bot():
|
||
"""Мок для бота"""
|
||
bot = Mock()
|
||
bot.get_file = AsyncMock()
|
||
bot.download_file = AsyncMock()
|
||
return bot
|
||
|
||
@pytest.fixture
|
||
def mock_message():
|
||
"""Мок для сообщения"""
|
||
message = Mock()
|
||
message.voice = Mock()
|
||
message.voice.file_id = "test_file_id"
|
||
return message
|
||
|
||
@pytest.fixture
|
||
def mock_file_info():
|
||
"""Мок для информации о файле"""
|
||
file_info = Mock()
|
||
file_info.file_path = "voice/test_file_id.ogg"
|
||
return file_info
|
||
|
||
|
||
class TestGenerateFileName:
|
||
"""Тесты для метода generate_file_name"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_file_name_first_record(self, audio_service, mock_bot_db):
|
||
"""Тест генерации имени файла для первой записи пользователя"""
|
||
mock_bot_db.get_user_audio_records_count.return_value = 0
|
||
|
||
result = await audio_service.generate_file_name(12345)
|
||
|
||
assert result == "message_from_12345_number_1"
|
||
mock_bot_db.get_user_audio_records_count.assert_called_once_with(user_id=12345)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_file_name_existing_records(self, audio_service, mock_bot_db):
|
||
"""Тест генерации имени файла для существующих записей"""
|
||
mock_bot_db.get_user_audio_records_count.return_value = 3
|
||
mock_bot_db.get_path_for_audio_record.return_value = "message_from_12345_number_3"
|
||
|
||
result = await audio_service.generate_file_name(12345)
|
||
|
||
assert result == "message_from_12345_number_4"
|
||
mock_bot_db.get_user_audio_records_count.assert_called_once_with(user_id=12345)
|
||
mock_bot_db.get_path_for_audio_record.assert_called_once_with(user_id=12345)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_file_name_no_last_record(self, audio_service, mock_bot_db):
|
||
"""Тест генерации имени файла когда нет последней записи"""
|
||
mock_bot_db.get_user_audio_records_count.return_value = 2
|
||
mock_bot_db.get_path_for_audio_record.return_value = None
|
||
|
||
result = await audio_service.generate_file_name(12345)
|
||
|
||
assert result == "message_from_12345_number_3"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_file_name_invalid_last_record_format(self, audio_service, mock_bot_db):
|
||
"""Тест генерации имени файла с некорректным форматом последней записи"""
|
||
mock_bot_db.get_user_audio_records_count.return_value = 2
|
||
mock_bot_db.get_path_for_audio_record.return_value = "invalid_format"
|
||
|
||
result = await audio_service.generate_file_name(12345)
|
||
|
||
assert result == "message_from_12345_number_3"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_file_name_exception_handling(self, audio_service, mock_bot_db):
|
||
"""Тест обработки исключений при генерации имени файла"""
|
||
mock_bot_db.get_user_audio_records_count.side_effect = Exception("Database error")
|
||
|
||
with pytest.raises(FileOperationError) as exc_info:
|
||
await audio_service.generate_file_name(12345)
|
||
|
||
assert "Не удалось сгенерировать имя файла" in str(exc_info.value)
|
||
|
||
|
||
class TestSaveAudioFile:
|
||
"""Тесты для метода save_audio_file"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_audio_file_success(self, audio_service, mock_bot_db, sample_datetime):
|
||
"""Тест успешного сохранения аудио файла"""
|
||
file_name = "test_audio.ogg"
|
||
user_id = 12345
|
||
file_id = "test_file_id"
|
||
|
||
await audio_service.save_audio_file(file_name, user_id, sample_datetime, file_id)
|
||
|
||
mock_bot_db.add_audio_record_simple.assert_called_once_with(file_name, user_id, sample_datetime)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_audio_file_with_string_date(self, audio_service, mock_bot_db):
|
||
"""Тест сохранения аудио файла со строковой датой"""
|
||
file_name = "test_audio.ogg"
|
||
user_id = 12345
|
||
date_string = "2025-01-15 14:30:00"
|
||
file_id = "test_file_id"
|
||
|
||
await audio_service.save_audio_file(file_name, user_id, date_string, file_id)
|
||
|
||
mock_bot_db.add_audio_record_simple.assert_called_once_with(file_name, user_id, date_string)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_save_audio_file_exception_handling(self, audio_service, mock_bot_db, sample_datetime):
|
||
"""Тест обработки исключений при сохранении аудио файла"""
|
||
mock_bot_db.add_audio_record_simple.side_effect = Exception("Database error")
|
||
|
||
with pytest.raises(DatabaseError) as exc_info:
|
||
await audio_service.save_audio_file("test.ogg", 12345, sample_datetime, "file_id")
|
||
|
||
assert "Не удалось сохранить аудио файл в БД" in str(exc_info.value)
|
||
|
||
|
||
class TestDownloadAndSaveAudio:
|
||
"""Тесты для метода download_and_save_audio"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_and_save_audio_success(self, audio_service, mock_bot, mock_message, mock_file_info):
|
||
"""Тест успешного скачивания и сохранения аудио"""
|
||
mock_bot.get_file.return_value = mock_file_info
|
||
|
||
# Мокаем скачанный файл
|
||
mock_downloaded_file = Mock()
|
||
mock_downloaded_file.tell.return_value = 0
|
||
mock_downloaded_file.seek = Mock()
|
||
mock_downloaded_file.read.return_value = b"audio_data"
|
||
mock_bot.download_file.return_value = mock_downloaded_file
|
||
|
||
with patch('builtins.open', mock_open()) as mock_file:
|
||
with patch('os.makedirs'):
|
||
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
|
||
|
||
mock_bot.get_file.assert_called_once_with(file_id="test_file_id")
|
||
mock_bot.download_file.assert_called_once_with(file_path="voice/test_file_id.ogg")
|
||
mock_file.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_and_save_audio_no_message(self, audio_service, mock_bot):
|
||
"""Тест скачивания когда сообщение отсутствует"""
|
||
with pytest.raises(FileOperationError) as exc_info:
|
||
await audio_service.download_and_save_audio(mock_bot, None, "test_audio")
|
||
|
||
assert "Сообщение или голосовое сообщение не найдено" in str(exc_info.value)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_and_save_audio_no_voice(self, audio_service, mock_bot):
|
||
"""Тест скачивания когда у сообщения нет voice атрибута"""
|
||
message = Mock()
|
||
message.voice = None
|
||
|
||
with pytest.raises(FileOperationError) as exc_info:
|
||
await audio_service.download_and_save_audio(mock_bot, message, "test_audio")
|
||
|
||
assert "Сообщение или голосовое сообщение не найдено" in str(exc_info.value)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_and_save_audio_download_failed(self, audio_service, mock_bot, mock_message, mock_file_info):
|
||
"""Тест скачивания когда загрузка не удалась"""
|
||
mock_bot.get_file.return_value = mock_file_info
|
||
mock_bot.download_file.return_value = None
|
||
|
||
with pytest.raises(FileOperationError) as exc_info:
|
||
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
|
||
|
||
assert "Не удалось скачать файл" in str(exc_info.value)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_and_save_audio_exception_handling(self, audio_service, mock_bot, mock_message):
|
||
"""Тест обработки исключений при скачивании"""
|
||
mock_bot.get_file.side_effect = Exception("Network error")
|
||
|
||
with pytest.raises(FileOperationError) as exc_info:
|
||
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
|
||
|
||
assert "Не удалось скачать и сохранить аудио" in str(exc_info.value)
|
||
|
||
|
||
def mock_open():
|
||
"""Мок для функции open"""
|
||
from unittest.mock import mock_open as _mock_open
|
||
return _mock_open()
|
||
|
||
|
||
class TestAudioFileServiceIntegration:
|
||
"""Интеграционные тесты для AudioFileService"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_audio_processing_workflow(self, mock_bot_db):
|
||
"""Тест полного рабочего процесса обработки аудио"""
|
||
service = AudioFileService(mock_bot_db)
|
||
|
||
# Настраиваем моки
|
||
mock_bot_db.get_user_audio_records_count.return_value = 1
|
||
mock_bot_db.get_path_for_audio_record.return_value = "message_from_12345_number_1"
|
||
mock_bot_db.add_audio_record_simple = AsyncMock()
|
||
|
||
# Тестируем генерацию имени файла
|
||
file_name = await service.generate_file_name(12345)
|
||
assert file_name == "message_from_12345_number_2"
|
||
|
||
# Тестируем сохранение в БД
|
||
test_date = datetime.now()
|
||
await service.save_audio_file(file_name, 12345, test_date, "test_file_id")
|
||
|
||
# Проверяем вызовы
|
||
mock_bot_db.get_user_audio_records_count.assert_called_once_with(user_id=12345)
|
||
mock_bot_db.get_path_for_audio_record.assert_called_once_with(user_id=12345)
|
||
mock_bot_db.add_audio_record_simple.assert_called_once_with(file_name, 12345, test_date)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_file_name_generation_sequence(self, mock_bot_db):
|
||
"""Тест последовательности генерации имен файлов"""
|
||
service = AudioFileService(mock_bot_db)
|
||
|
||
# Первая запись
|
||
mock_bot_db.get_user_audio_records_count.return_value = 0
|
||
file_name_1 = await service.generate_file_name(12345)
|
||
assert file_name_1 == "message_from_12345_number_1"
|
||
|
||
# Вторая запись
|
||
mock_bot_db.get_user_audio_records_count.return_value = 1
|
||
mock_bot_db.get_path_for_audio_record.return_value = "message_from_12345_number_1"
|
||
file_name_2 = await service.generate_file_name(12345)
|
||
assert file_name_2 == "message_from_12345_number_2"
|
||
|
||
# Третья запись
|
||
mock_bot_db.get_user_audio_records_count.return_value = 2
|
||
mock_bot_db.get_path_for_audio_record.return_value = "message_from_12345_number_2"
|
||
file_name_3 = await service.generate_file_name(12345)
|
||
assert file_name_3 == "message_from_12345_number_3"
|
||
|
||
|
||
if __name__ == '__main__':
|
||
pytest.main([__file__])
|