Andrey
3d6b4353f9
All checks were successful
CI pipeline / Test & Code Quality (push) Successful in 34s
451 lines
18 KiB
Python
451 lines
18 KiB
Python
import time
|
||
from datetime import datetime, timezone
|
||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||
|
||
import pytest
|
||
|
||
from database.repositories.audio_repository import AudioRepository
|
||
|
||
|
||
class TestAudioRepositoryNewSchema:
|
||
"""Тесты для 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 для тестов"""
|
||
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.mark.asyncio
|
||
async def test_create_tables_new_schema(self, audio_repository):
|
||
"""Тест создания таблиц с новой схемой БД"""
|
||
await audio_repository.create_tables()
|
||
|
||
# Проверяем, что все три таблицы созданы
|
||
assert audio_repository._execute_query.call_count == 3
|
||
|
||
# Получаем все вызовы
|
||
calls = audio_repository._execute_query.call_args_list
|
||
|
||
# Проверяем таблицу audio_message_reference
|
||
audio_table_call = next(
|
||
call for call in calls if "audio_message_reference" in str(call)
|
||
)
|
||
assert "id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT" in str(audio_table_call)
|
||
assert "file_name TEXT NOT NULL UNIQUE" in str(audio_table_call)
|
||
assert "author_id INTEGER NOT NULL" in str(audio_table_call)
|
||
assert "date_added INTEGER NOT NULL" in str(audio_table_call)
|
||
assert (
|
||
"FOREIGN KEY (author_id) REFERENCES our_users (user_id) ON DELETE CASCADE"
|
||
in str(audio_table_call)
|
||
)
|
||
|
||
# Проверяем таблицу user_audio_listens
|
||
listens_table_call = next(
|
||
call for call in calls if "user_audio_listens" in str(call)
|
||
)
|
||
assert "file_name TEXT NOT NULL" in str(listens_table_call)
|
||
assert "user_id INTEGER NOT NULL" in str(listens_table_call)
|
||
assert "PRIMARY KEY (file_name, user_id)" in str(listens_table_call)
|
||
assert (
|
||
"FOREIGN KEY (user_id) REFERENCES our_users (user_id) ON DELETE CASCADE"
|
||
in str(listens_table_call)
|
||
)
|
||
|
||
# Проверяем таблицу audio_moderate
|
||
moderate_table_call = next(
|
||
call for call in calls if "audio_moderate" in str(call)
|
||
)
|
||
assert "user_id INTEGER NOT NULL" in str(moderate_table_call)
|
||
assert "message_id INTEGER" in str(moderate_table_call)
|
||
assert "PRIMARY KEY (user_id, message_id)" in str(moderate_table_call)
|
||
assert (
|
||
"FOREIGN KEY (user_id) REFERENCES our_users (user_id) ON DELETE CASCADE"
|
||
in str(moderate_table_call)
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_string_date_conversion(self, audio_repository):
|
||
"""Тест преобразования строковой даты в UNIX timestamp"""
|
||
from database.models import AudioMessage
|
||
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added="2025-01-15 14:30:00",
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем, что метод вызван
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
|
||
# Проверяем параметры
|
||
assert params[0] == "test_audio.ogg"
|
||
assert params[1] == 12345
|
||
assert isinstance(params[2], int) # timestamp
|
||
|
||
# Проверяем, что timestamp соответствует дате
|
||
expected_timestamp = int(datetime(2025, 1, 15, 14, 30, 0).timestamp())
|
||
assert params[2] == expected_timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_datetime_conversion(self, audio_repository):
|
||
"""Тест преобразования datetime в UNIX timestamp"""
|
||
from database.models import AudioMessage
|
||
|
||
test_datetime = datetime(2025, 1, 20, 10, 15, 30)
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added=test_datetime,
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем параметры
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
|
||
expected_timestamp = int(test_datetime.timestamp())
|
||
assert params[2] == expected_timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_timestamp_no_conversion(self, audio_repository):
|
||
"""Тест что timestamp остается timestamp без преобразования"""
|
||
from database.models import AudioMessage
|
||
|
||
test_timestamp = int(time.time())
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added=test_timestamp,
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем параметры
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
|
||
assert params[2] == test_timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_string_date(self, audio_repository):
|
||
"""Тест упрощенного добавления со строковой датой"""
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, "2025-01-15 14:30:00"
|
||
)
|
||
|
||
# Проверяем параметры
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
|
||
assert params[0] == "test_audio.ogg"
|
||
assert params[1] == 12345
|
||
assert isinstance(params[2], int) # timestamp
|
||
|
||
# Проверяем timestamp
|
||
expected_timestamp = int(datetime(2025, 1, 15, 14, 30, 0).timestamp())
|
||
assert params[2] == expected_timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_datetime(self, audio_repository):
|
||
"""Тест упрощенного добавления с datetime"""
|
||
test_datetime = datetime(2025, 1, 25, 16, 45, 0)
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, test_datetime
|
||
)
|
||
|
||
# Проверяем параметры
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
|
||
expected_timestamp = int(test_datetime.timestamp())
|
||
assert params[2] == expected_timestamp
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_timestamp_conversion(self, audio_repository):
|
||
"""Тест преобразования UNIX timestamp в читаемую дату (UTC)."""
|
||
test_timestamp = 1642248600 # 2022-01-15 12:10:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(test_timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(test_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
assert isinstance(result, str)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_different_timestamp(self, audio_repository):
|
||
"""Тест преобразования другого timestamp в читаемую дату (UTC)."""
|
||
test_timestamp = 1705312800 # 2024-01-16 12:00:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(test_timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(test_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_midnight(self, audio_repository):
|
||
"""Тест преобразования timestamp для полуночи (UTC)."""
|
||
test_timestamp = 1705190400 # 2024-01-14 00:00:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(test_timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(test_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_year_end(self, audio_repository):
|
||
"""Тест преобразования timestamp для конца года (UTC)."""
|
||
test_timestamp = 1704067200 # 2023-12-31 00:00:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(test_timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(test_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_foreign_keys_enabled_called(self, audio_repository):
|
||
"""Тест что метод enable_foreign_keys вызывается"""
|
||
await audio_repository.enable_foreign_keys()
|
||
|
||
audio_repository._execute_query.assert_called_once_with(
|
||
"PRAGMA foreign_keys = ON;"
|
||
)
|
||
audio_repository.logger.info.assert_not_called() # Этот метод не логирует
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables_logging(self, audio_repository):
|
||
"""Тест логирования при создании таблиц"""
|
||
await audio_repository.create_tables()
|
||
|
||
# Проверяем, что лог записан
|
||
audio_repository.logger.info.assert_called_once_with(
|
||
"Таблицы для аудио созданы"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_logging_format(self, audio_repository):
|
||
"""Тест формата лога при добавлении аудио записи"""
|
||
from database.models import AudioMessage
|
||
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added="2025-01-15 14:30:00",
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем формат лога
|
||
log_call = audio_repository.logger.info.call_args
|
||
log_message = log_call[0][0]
|
||
|
||
assert "Аудио добавлено:" in log_message
|
||
assert "file_name=test_audio.ogg" in log_message
|
||
assert "author_id=12345" in log_message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_logging_format(self, audio_repository):
|
||
"""Тест формата лога при упрощенном добавлении"""
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, "2025-01-15 14:30:00"
|
||
)
|
||
|
||
# Проверяем формат лога
|
||
log_call = audio_repository.logger.info.call_args
|
||
log_message = log_call[0][0]
|
||
|
||
assert "Аудио добавлено:" in log_message
|
||
assert "file_name=test_audio.ogg" in log_message
|
||
assert "user_id=12345" in log_message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_logging_format(self, audio_repository):
|
||
"""Тест формата лога при получении даты (UTC)."""
|
||
test_timestamp = 1642248600 # 2022-01-15 12:10:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(test_timestamp,)]
|
||
|
||
await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(test_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
log_call = audio_repository.logger.info.call_args
|
||
log_message = log_call[0][0]
|
||
|
||
assert "Получена дата" in log_message
|
||
assert expected in log_message
|
||
assert "test_audio.ogg" in log_message
|
||
|
||
|
||
class TestAudioRepositoryEdgeCases:
|
||
"""Тесты граничных случаев для AudioRepository"""
|
||
|
||
@pytest.fixture
|
||
def audio_repository(self):
|
||
"""Экземпляр AudioRepository для тестов"""
|
||
with patch.object(AudioRepository, "__init__", return_value=None):
|
||
repo = AudioRepository()
|
||
repo._execute_query = AsyncMock()
|
||
repo._execute_query_with_result = AsyncMock()
|
||
repo.logger = Mock()
|
||
return repo
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_empty_string_date(self, audio_repository):
|
||
"""Тест добавления с пустой строковой датой"""
|
||
from database.models import AudioMessage
|
||
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added="",
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
# Должно вызвать ValueError при парсинге пустой строки
|
||
with pytest.raises(ValueError):
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_invalid_string_date(self, audio_repository):
|
||
"""Тест добавления с некорректной строковой датой"""
|
||
from database.models import AudioMessage
|
||
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added="invalid_date",
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
# Должно вызвать ValueError при парсинге некорректной даты
|
||
with pytest.raises(ValueError):
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_none_date(self, audio_repository):
|
||
"""Тест добавления с None датой"""
|
||
from database.models import AudioMessage
|
||
|
||
audio_msg = AudioMessage(
|
||
file_name="test_audio.ogg",
|
||
author_id=12345,
|
||
date_added=None,
|
||
file_id="test_file_id",
|
||
listen_count=0,
|
||
)
|
||
|
||
# Метод обрабатывает None как timestamp без преобразования
|
||
await audio_repository.add_audio_record(audio_msg)
|
||
|
||
# Проверяем, что метод был вызван с None
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
assert params[2] is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_empty_string_date(self, audio_repository):
|
||
"""Тест упрощенного добавления с пустой строковой датой"""
|
||
# Должно вызвать ValueError при парсинге пустой строки
|
||
with pytest.raises(ValueError):
|
||
await audio_repository.add_audio_record_simple("test_audio.ogg", 12345, "")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_invalid_string_date(self, audio_repository):
|
||
"""Тест упрощенного добавления с некорректной строковой датой"""
|
||
# Должно вызвать ValueError при парсинге некорректной даты
|
||
with pytest.raises(ValueError):
|
||
await audio_repository.add_audio_record_simple(
|
||
"test_audio.ogg", 12345, "invalid_date"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_audio_record_simple_none_date(self, audio_repository):
|
||
"""Тест упрощенного добавления с None датой"""
|
||
# Метод обрабатывает None как timestamp без преобразования
|
||
await audio_repository.add_audio_record_simple("test_audio.ogg", 12345, None)
|
||
|
||
# Проверяем, что метод был вызван с None
|
||
call_args = audio_repository._execute_query.call_args
|
||
params = call_args[0][1]
|
||
assert params[2] is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_zero_timestamp(self, audio_repository):
|
||
"""Тест получения даты для timestamp = 0 (1970-01-01 UTC)."""
|
||
audio_repository._execute_query_with_result.return_value = [(0,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(0, tz=timezone.utc).strftime("%d.%m.%Y %H:%M")
|
||
assert result == expected
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_negative_timestamp(self, audio_repository):
|
||
"""Тест получения даты для отрицательного timestamp (UTC)."""
|
||
ts = -3600 # 1969-12-31 23:00:00 UTC
|
||
audio_repository._execute_query_with_result.return_value = [(ts,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(ts, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_date_by_file_name_future_timestamp(self, audio_repository):
|
||
"""Тест получения даты для будущего timestamp (UTC, без зависимости от локали)."""
|
||
future_timestamp = int(
|
||
datetime(2030, 12, 31, 23, 59, 59, tzinfo=timezone.utc).timestamp()
|
||
)
|
||
audio_repository._execute_query_with_result.return_value = [(future_timestamp,)]
|
||
|
||
result = await audio_repository.get_date_by_file_name("test_audio.ogg")
|
||
|
||
expected = datetime.fromtimestamp(future_timestamp, tz=timezone.utc).strftime(
|
||
"%d.%m.%Y %H:%M"
|
||
)
|
||
assert result == expected
|