444 lines
19 KiB
Python
444 lines
19 KiB
Python
import time
|
||
from datetime import datetime, timezone
|
||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||
|
||
import pytest
|
||
from database.models import AudioListenRecord, AudioMessage, AudioModerate
|
||
from database.repositories.audio_repository import AudioRepository
|
||
|
||
|
||
class TestAudioRepository:
|
||
"""Тесты для AudioRepository"""
|
||
|
||
@pytest.fixture
|
||
def mock_db_connection(self):
|
||
"""Мок для DatabaseConnection"""
|
||
mock_connection = Mock()
|
||
mock_connection._execute_query = AsyncMock()
|
||
mock_connection._execute_query_with_result = AsyncMock()
|
||
mock_connection.logger = Mock()
|
||
return mock_connection
|
||
|
||
@pytest.fixture
|
||
def audio_repository(self, mock_db_connection):
|
||
"""Экземпляр AudioRepository для тестов"""
|
||
# Патчим наследование от DatabaseConnection
|
||
with patch.object(AudioRepository, "__init__", return_value=None):
|
||
repo = AudioRepository()
|
||
repo._execute_query = mock_db_connection._execute_query
|
||
repo._execute_query_with_result = (
|
||
mock_db_connection._execute_query_with_result
|
||
)
|
||
repo.logger = mock_db_connection.logger
|
||
return repo
|
||
|
||
@pytest.fixture
|
||
def sample_audio_message(self):
|
||
"""Тестовое аудио сообщение"""
|
||
return AudioMessage(
|
||
file_name="test_audio_123.ogg",
|
||
author_id=12345,
|
||
date_added="2025-01-15 14:30:00",
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
@pytest.fixture
|
||
def sample_datetime(self):
|
||
"""Тестовая дата"""
|
||
return datetime(2025, 1, 15, 14, 30, 0)
|
||
|
||
@pytest.fixture
|
||
def sample_timestamp(self):
|
||
"""Тестовый UNIX timestamp"""
|
||
return int(time.mktime(datetime(2025, 1, 15, 14, 30, 0).timetuple()))
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_enable_foreign_keys(self, audio_repository):
|
||
"""Тест включения внешних ключей"""
|
||
await audio_repository.enable_foreign_keys()
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"PRAGMA foreign_keys = ON;"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables(self, audio_repository):
|
||
"""Тест создания таблиц"""
|
||
await audio_repository.create_tables()
|
||
|
||
# Проверяем, что все три таблицы созданы
|
||
assert audio_repository._execute_query.call_count == 3
|
||
|
||
# Проверяем вызовы для каждой таблицы
|
||
calls = audio_repository._execute_query.call_args_list
|
||
assert any("audio_message_reference" in str(call) for call in calls)
|
||
assert any("user_audio_listens" in str(call) for call in calls)
|
||
assert any("audio_moderate" in str(call) for call in calls)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_with_string_date(
|
||
self, audio_repository, sample_audio_message
|
||
):
|
||
"""Тест добавления аудио записи со строковой датой"""
|
||
await audio_repository.add_audio_record(sample_audio_message)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
audio_repository._execute_query.assert_called_once()
|
||
call_args = audio_repository._execute_query.call_args
|
||
assert call_args[0][0] == """
|
||
INSERT INTO audio_message_reference (file_name, author_id, date_added)
|
||
VALUES (?, ?, ?)
|
||
"""
|
||
# Проверяем, что date_added преобразован в timestamp
|
||
assert call_args[0][1][0] == "test_audio_123.ogg"
|
||
assert call_args[0][1][1] == 12345
|
||
assert isinstance(call_args[0][1][2], int) # timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_with_datetime_date(self, audio_repository):
|
||
"""Тест добавления аудио записи с datetime датой"""
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio_456.ogg",
|
||
author_id=67890,
|
||
date_added=datetime(2025, 1, 20, 10, 15, 0),
|
||
file_id="test_file_id_2",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем, что date_added преобразован в timestamp
|
||
call_args = audio_repository._execute_query.call_args
|
||
assert isinstance(call_args[0][1][2], int) # timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_with_timestamp_date(self, audio_repository):
|
||
"""Тест добавления аудио записи с timestamp датой"""
|
||
timestamp = int(time.time())
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio_789.ogg",
|
||
author_id=11111,
|
||
date_added=timestamp,
|
||
file_id="test_file_id_3",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем, что date_added остался timestamp
|
||
call_args = audio_repository._execute_query.call_args
|
||
assert call_args[0][1][2] == timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_with_string_date(self, audio_repository):
|
||
"""Тест упрощенного добавления аудио записи со строковой датой"""
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, "2025-01-15 14:30:00"
|
||
)
|
||
|
||
# Проверяем, что метод вызван
|
||
audio_repository._execute_query.assert_called_once()
|
||
call_args = audio_repository._execute_query.call_args
|
||
assert call_args[0][1][0] == "test_audio.ogg" # file_name
|
||
assert call_args[0][1][1] == 12345 # user_id
|
||
assert isinstance(call_args[0][1][2], int) # timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_with_datetime_date(
|
||
self, audio_repository, sample_datetime
|
||
):
|
||
"""Тест упрощенного добавления аудио записи с datetime датой"""
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, sample_datetime
|
||
)
|
||
|
||
# Проверяем, что date_added преобразован в timestamp
|
||
call_args = audio_repository._execute_query.call_args
|
||
assert isinstance(call_args[0][1][2], int) # timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_last_date_audio(self, audio_repository):
|
||
"""Тест получения даты последнего аудио"""
|
||
expected_timestamp = 1642248600 # 2022-01-17 10:30:00
|
||
audio_repository._execute_query_with_result.return_value = [
|
||
(expected_timestamp,)
|
||
]
|
||
|
||
result = await audio_repository.get_last_date_audio()
|
||
|
||
assert result == expected_timestamp
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"SELECT date_added FROM audio_message_reference ORDER BY date_added DESC LIMIT 1"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_last_date_audio_no_records(self, audio_repository):
|
||
"""Тест получения даты последнего аудио когда записей нет"""
|
||
audio_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await audio_repository.get_last_date_audio()
|
||
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_audio_records_count(self, audio_repository):
|
||
"""Тест получения количества аудио записей пользователя"""
|
||
audio_repository._execute_query_with_result.return_value = [(5,)]
|
||
|
||
result = await audio_repository.get_user_audio_records_count(12345)
|
||
|
||
assert result == 5
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"SELECT COUNT(*) FROM audio_message_reference WHERE author_id = ?", (12345,)
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_path_for_audio_record(self, audio_repository):
|
||
"""Тест получения пути к аудио записи пользователя"""
|
||
audio_repository._execute_query_with_result.return_value = [("test_audio.ogg",)]
|
||
|
||
result = await audio_repository.get_path_for_audio_record(12345)
|
||
|
||
assert result == "test_audio.ogg"
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"""
|
||
SELECT file_name FROM audio_message_reference
|
||
WHERE author_id = ? ORDER BY date_added DESC LIMIT 1
|
||
""",
|
||
(12345,),
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_path_for_audio_record_no_records(self, audio_repository):
|
||
"""Тест получения пути к аудио записи когда записей нет"""
|
||
audio_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await audio_repository.get_path_for_audio_record(12345)
|
||
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_listen_audio(self, audio_repository):
|
||
"""Тест проверки непрослушанных аудио"""
|
||
# Мокаем результаты запросов
|
||
audio_repository._execute_query_with_result.side_effect = [
|
||
[("audio1.ogg",), ("audio2.ogg",)], # прослушанные
|
||
[("audio1.ogg",), ("audio2.ogg",), ("audio3.ogg",)], # все аудио
|
||
]
|
||
|
||
result = await audio_repository.check_listen_audio(12345)
|
||
|
||
# Должно вернуться только непрослушанные (audio3.ogg)
|
||
assert result == ["audio3.ogg"]
|
||
assert audio_repository._execute_query_with_result.call_count == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_mark_listened_audio(self, audio_repository):
|
||
"""Тест отметки аудио как прослушанного"""
|
||
await audio_repository.mark_listened_audio("test_audio.ogg", 12345)
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"INSERT OR IGNORE INTO user_audio_listens (file_name, user_id) VALUES (?, ?)",
|
||
("test_audio.ogg", 12345),
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_id_by_file_name(self, audio_repository):
|
||
"""Тест получения user_id по имени файла"""
|
||
audio_repository._execute_query_with_result.return_value = [(12345,)]
|
||
|
||
result = await audio_repository.get_user_id_by_file_name("test_audio.ogg")
|
||
|
||
assert result == 12345
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"SELECT author_id FROM audio_message_reference WHERE file_name = ?",
|
||
("test_audio.ogg",),
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_id_by_file_name_not_found(self, audio_repository):
|
||
"""Тест получения user_id по имени файла когда файл не найден"""
|
||
audio_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await audio_repository.get_user_id_by_file_name("nonexistent.ogg")
|
||
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name(self, audio_repository):
|
||
"""Тест получения даты по имени файла (UTC, без зависимости от локали)."""
|
||
timestamp = 1642404600 # 2022-01-17 10:30:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"SELECT date_added FROM audio_message_reference WHERE file_name = ?",
|
||
("test_audio.ogg",),
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_not_found(self, audio_repository):
|
||
"""Тест получения даты по имени файла когда файл не найден"""
|
||
audio_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await audio_repository.get_date_by_file_name("nonexistent.ogg")
|
||
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_refresh_listen_audio(self, audio_repository):
|
||
"""Тест очистки записей прослушивания пользователя"""
|
||
await audio_repository.refresh_listen_audio(12345)
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"DELETE FROM user_audio_listens WHERE user_id = ?", (12345,)
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_listen_count_for_user(self, audio_repository):
|
||
"""Тест удаления данных о прослушанных аудио пользователя"""
|
||
await audio_repository.delete_listen_count_for_user(12345)
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"DELETE FROM user_audio_listens WHERE user_id = ?", (12345,)
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_user_id_and_message_id_for_voice_bot_success(
|
||
self, audio_repository
|
||
):
|
||
"""Тест успешной установки связи для voice bot"""
|
||
result = await audio_repository.set_user_id_and_message_id_for_voice_bot(
|
||
123, 456
|
||
)
|
||
|
||
assert result is True
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"INSERT OR IGNORE INTO audio_moderate (user_id, message_id) VALUES (?, ?)",
|
||
(456, 123),
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_user_id_and_message_id_for_voice_bot_exception(
|
||
self, audio_repository
|
||
):
|
||
"""Тест установки связи для voice bot при ошибке"""
|
||
audio_repository._execute_query.side_effect = Exception("Database error")
|
||
|
||
result = await audio_repository.set_user_id_and_message_id_for_voice_bot(
|
||
123, 456
|
||
)
|
||
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_id_by_message_id_for_voice_bot(self, audio_repository):
|
||
"""Тест получения user_id по message_id для voice bot"""
|
||
audio_repository._execute_query_with_result.return_value = [(456,)]
|
||
|
||
result = await audio_repository.get_user_id_by_message_id_for_voice_bot(123)
|
||
|
||
assert result == 456
|
||
audio_repository._execute_query_with_result.assert_called_once_with(
|
||
"SELECT user_id FROM audio_moderate WHERE message_id = ?", (123,)
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_id_by_message_id_for_voice_bot_not_found(
|
||
self, audio_repository
|
||
):
|
||
"""Тест получения user_id по message_id когда связь не найдена"""
|
||
audio_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await audio_repository.get_user_id_by_message_id_for_voice_bot(123)
|
||
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_audio_moderate_record(self, audio_repository):
|
||
"""Тест удаления записи из таблицы audio_moderate"""
|
||
message_id = 12345
|
||
|
||
await audio_repository.delete_audio_moderate_record(message_id)
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"DELETE FROM audio_moderate WHERE message_id = ?", (message_id,)
|
||
)
|
||
audio_repository.logger.info.assert_called_once_with(
|
||
f"Удалена запись из audio_moderate для message_id {message_id}"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_logging(
|
||
self, audio_repository, sample_audio_message
|
||
):
|
||
"""Тест логирования при добавлении аудио записи"""
|
||
await audio_repository.add_audio_record(sample_audio_message)
|
||
|
||
# Проверяем, что лог записан
|
||
audio_repository.logger.info.assert_called_once()
|
||
log_message = audio_repository.logger.info.call_args[0][0]
|
||
assert "Аудио добавлено" in log_message
|
||
assert "test_audio_123.ogg" in log_message
|
||
assert "12345" in log_message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_logging(self, audio_repository):
|
||
"""Тест логирования при упрощенном добавлении аудио записи"""
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, "2025-01-15 14:30:00"
|
||
)
|
||
|
||
# Проверяем, что лог записан
|
||
audio_repository.logger.info.assert_called_once()
|
||
log_message = audio_repository.logger.info.call_args[0][0]
|
||
assert "Аудио добавлено" in log_message
|
||
assert "test_audio.ogg" in log_message
|
||
assert "12345" in log_message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_logging(self, audio_repository):
|
||
"""Тест логирования при получении даты по имени файла (UTC)."""
|
||
timestamp = 1642404600 # 2022-01-17 10:30:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(timestamp,)]
|
||
|
||
await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
audio_repository.logger.info.assert_called_once()
|
||
log_message = audio_repository.logger.info.call_args[0][0]
|
||
assert "Получена дата" in log_message
|
||
assert expected in log_message
|
||
assert "test_audio.ogg" in log_message
|
||
|
||
|
||
class TestAudioRepositoryIntegration:
|
||
"""Интеграционные тесты для AudioRepository"""
|
||
|
||
@pytest.fixture
|
||
def real_audio_repository(self):
|
||
"""Реальный экземпляр AudioRepository для интеграционных тестов"""
|
||
# Здесь можно создать реальное подключение к тестовой БД
|
||
# Но для простоты используем мок
|
||
return Mock()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_audio_workflow(self, real_audio_repository):
|
||
"""Тест полного рабочего процесса с аудио"""
|
||
# Этот тест можно расширить для реальной БД
|
||
assert True # Placeholder для будущих интеграционных тестов
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_foreign_keys_enabled(self, real_audio_repository):
|
||
"""Тест что внешние ключи включены"""
|
||
# Этот тест можно расширить для реальной БД
|
||
assert True # Placeholder для будущих интеграционных тестов
|