290 lines
13 KiB
Python
290 lines
13 KiB
Python
import time
|
||
from datetime import datetime
|
||
from unittest.mock import AsyncMock, Mock, patch
|
||
|
||
import pytest
|
||
|
||
from database.models import BlacklistHistoryRecord
|
||
from database.repositories.blacklist_history_repository import (
|
||
BlacklistHistoryRepository,
|
||
)
|
||
|
||
|
||
class TestBlacklistHistoryRepository:
|
||
"""Тесты для BlacklistHistoryRepository"""
|
||
|
||
@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 blacklist_history_repository(self, mock_db_connection):
|
||
"""Экземпляр BlacklistHistoryRepository для тестов"""
|
||
# Патчим наследование от DatabaseConnection
|
||
with patch.object(BlacklistHistoryRepository, "__init__", return_value=None):
|
||
repo = BlacklistHistoryRepository()
|
||
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_history_record(self):
|
||
"""Тестовая запись истории бана"""
|
||
current_time = int(time.time())
|
||
return BlacklistHistoryRecord(
|
||
user_id=12345,
|
||
message_for_user="Нарушение правил",
|
||
date_ban=current_time,
|
||
date_unban=None,
|
||
ban_author=999,
|
||
created_at=current_time,
|
||
updated_at=current_time,
|
||
)
|
||
|
||
@pytest.fixture
|
||
def sample_history_record_with_unban(self):
|
||
"""Тестовая запись истории бана с датой разбана"""
|
||
current_time = int(time.time())
|
||
return BlacklistHistoryRecord(
|
||
user_id=12345,
|
||
message_for_user="Нарушение правил",
|
||
date_ban=current_time - 86400, # Бан был вчера
|
||
date_unban=current_time, # Разбан сегодня
|
||
ban_author=999,
|
||
created_at=current_time - 86400,
|
||
updated_at=current_time,
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables(self, blacklist_history_repository):
|
||
"""Тест создания таблицы истории банов/разбанов"""
|
||
await blacklist_history_repository.create_tables()
|
||
|
||
# Проверяем, что метод вызван (4 раза: таблица + 3 индекса)
|
||
assert blacklist_history_repository._execute_query.call_count == 4
|
||
calls = blacklist_history_repository._execute_query.call_args_list
|
||
|
||
# Проверяем, что создается таблица с правильной структурой
|
||
create_table_call = calls[0]
|
||
assert "CREATE TABLE IF NOT EXISTS blacklist_history" in create_table_call[0][0]
|
||
assert (
|
||
"id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT" in create_table_call[0][0]
|
||
)
|
||
assert "user_id INTEGER NOT NULL" in create_table_call[0][0]
|
||
assert "message_for_user TEXT" in create_table_call[0][0]
|
||
assert "date_ban INTEGER NOT NULL" in create_table_call[0][0]
|
||
assert "date_unban INTEGER" in create_table_call[0][0]
|
||
assert "ban_author INTEGER" in create_table_call[0][0]
|
||
assert (
|
||
"created_at INTEGER DEFAULT (strftime('%s', 'now'))"
|
||
in create_table_call[0][0]
|
||
)
|
||
assert (
|
||
"updated_at INTEGER DEFAULT (strftime('%s', 'now'))"
|
||
in create_table_call[0][0]
|
||
)
|
||
assert (
|
||
"FOREIGN KEY (user_id) REFERENCES our_users(user_id) ON DELETE CASCADE"
|
||
in create_table_call[0][0]
|
||
)
|
||
assert (
|
||
"FOREIGN KEY (ban_author) REFERENCES our_users(user_id) ON DELETE SET NULL"
|
||
in create_table_call[0][0]
|
||
)
|
||
|
||
# Проверяем создание индексов
|
||
index_calls = calls[1:4]
|
||
index_names = [call[0][0] for call in index_calls]
|
||
assert any("idx_blacklist_history_user_id" in idx for idx in index_names)
|
||
assert any("idx_blacklist_history_date_ban" in idx for idx in index_names)
|
||
assert any("idx_blacklist_history_date_unban" in idx for idx in index_names)
|
||
|
||
# Проверяем логирование
|
||
blacklist_history_repository.logger.info.assert_called_once_with(
|
||
"Таблица истории банов/разбанов создана"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_record_on_ban(
|
||
self, blacklist_history_repository, sample_history_record
|
||
):
|
||
"""Тест добавления записи о бане в историю"""
|
||
await blacklist_history_repository.add_record_on_ban(sample_history_record)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
blacklist_history_repository._execute_query.assert_called_once()
|
||
call_args = blacklist_history_repository._execute_query.call_args
|
||
|
||
# Проверяем SQL запрос
|
||
sql_query = call_args[0][0].replace("\n", " ").replace(" ", " ").strip()
|
||
assert "INSERT INTO blacklist_history" in sql_query
|
||
assert (
|
||
"user_id, message_for_user, date_ban, date_unban, ban_author, created_at, updated_at"
|
||
in sql_query
|
||
)
|
||
|
||
# Проверяем параметры
|
||
params = call_args[0][1]
|
||
assert params[0] == 12345 # user_id
|
||
assert params[1] == "Нарушение правил" # message_for_user
|
||
assert params[2] == sample_history_record.date_ban # date_ban
|
||
assert params[3] is None # date_unban
|
||
assert params[4] == 999 # ban_author
|
||
assert params[5] == sample_history_record.created_at # created_at
|
||
assert params[6] == sample_history_record.updated_at # updated_at
|
||
|
||
# Проверяем логирование
|
||
blacklist_history_repository.logger.info.assert_called_once()
|
||
log_call = blacklist_history_repository.logger.info.call_args[0][0]
|
||
assert "Запись о бане добавлена в историю" in log_call
|
||
assert "user_id=12345" in log_call
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_record_on_ban_with_defaults(self, blacklist_history_repository):
|
||
"""Тест добавления записи о бане с дефолтными значениями created_at и updated_at"""
|
||
record = BlacklistHistoryRecord(
|
||
user_id=12345,
|
||
message_for_user="Тест",
|
||
date_ban=int(time.time()),
|
||
date_unban=None,
|
||
ban_author=None,
|
||
created_at=None, # Будет установлено автоматически
|
||
updated_at=None, # Будет установлено автоматически
|
||
)
|
||
|
||
await blacklist_history_repository.add_record_on_ban(record)
|
||
|
||
# Проверяем, что метод вызван
|
||
blacklist_history_repository._execute_query.assert_called_once()
|
||
call_args = blacklist_history_repository._execute_query.call_args
|
||
|
||
# Проверяем, что created_at и updated_at установлены (не None)
|
||
params = call_args[0][1]
|
||
assert params[5] is not None # created_at
|
||
assert params[6] is not None # updated_at
|
||
assert isinstance(params[5], int)
|
||
assert isinstance(params[6], int)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_unban_date_success(self, blacklist_history_repository):
|
||
"""Тест успешного обновления даты разбана"""
|
||
user_id = 12345
|
||
date_unban = int(time.time())
|
||
|
||
# Мокируем результат проверки - находим открытую запись
|
||
blacklist_history_repository._execute_query_with_result.return_value = [
|
||
(100,)
|
||
] # id записи
|
||
|
||
result = await blacklist_history_repository.set_unban_date(user_id, date_unban)
|
||
|
||
# Проверяем, что сначала проверяется наличие записи
|
||
assert blacklist_history_repository._execute_query_with_result.call_count == 1
|
||
check_call = blacklist_history_repository._execute_query_with_result.call_args
|
||
assert "SELECT id FROM blacklist_history" in check_call[0][0]
|
||
assert check_call[0][1] == (user_id,)
|
||
|
||
# Проверяем, что затем обновляется запись
|
||
assert blacklist_history_repository._execute_query.call_count == 1
|
||
update_call = blacklist_history_repository._execute_query.call_args
|
||
update_query = (
|
||
update_call[0][0].replace("\n", " ").replace(" ", " ").strip()
|
||
)
|
||
assert "UPDATE blacklist_history" in update_query
|
||
assert "SET date_unban = ?" in update_query
|
||
assert "updated_at = ?" in update_query
|
||
|
||
# Проверяем параметры обновления
|
||
update_params = update_call[0][1]
|
||
assert update_params[0] == date_unban
|
||
assert update_params[1] is not None # updated_at (текущее время)
|
||
assert isinstance(update_params[1], int)
|
||
assert update_params[2] == 100 # id записи
|
||
|
||
# Проверяем результат
|
||
assert result is True
|
||
|
||
# Проверяем логирование
|
||
blacklist_history_repository.logger.info.assert_called_once()
|
||
log_call = blacklist_history_repository.logger.info.call_args[0][0]
|
||
assert "Дата разбана обновлена в истории" in log_call
|
||
assert f"user_id={user_id}" in log_call
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_unban_date_no_open_record(self, blacklist_history_repository):
|
||
"""Тест обновления даты разбана когда нет открытой записи"""
|
||
user_id = 12345
|
||
date_unban = int(time.time())
|
||
|
||
# Мокируем результат проверки - нет открытых записей
|
||
blacklist_history_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await blacklist_history_repository.set_unban_date(user_id, date_unban)
|
||
|
||
# Проверяем, что проверка была выполнена
|
||
assert blacklist_history_repository._execute_query_with_result.call_count == 1
|
||
|
||
# Проверяем, что UPDATE не был вызван (нет записей для обновления)
|
||
blacklist_history_repository._execute_query.assert_not_called()
|
||
|
||
# Проверяем результат
|
||
assert result is False
|
||
|
||
# Проверяем логирование предупреждения
|
||
blacklist_history_repository.logger.warning.assert_called_once()
|
||
log_call = blacklist_history_repository.logger.warning.call_args[0][0]
|
||
assert "Не найдена открытая запись в истории для обновления" in log_call
|
||
assert f"user_id={user_id}" in log_call
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_unban_date_exception(self, blacklist_history_repository):
|
||
"""Тест обработки исключения при обновлении даты разбана"""
|
||
user_id = 12345
|
||
date_unban = int(time.time())
|
||
|
||
# Мокируем исключение при проверке
|
||
blacklist_history_repository._execute_query_with_result.side_effect = Exception(
|
||
"Database error"
|
||
)
|
||
|
||
result = await blacklist_history_repository.set_unban_date(user_id, date_unban)
|
||
|
||
# Проверяем, что метод вернул False при ошибке
|
||
assert result is False
|
||
|
||
# Проверяем логирование ошибки
|
||
blacklist_history_repository.logger.error.assert_called_once()
|
||
log_call = blacklist_history_repository.logger.error.call_args[0][0]
|
||
assert "Ошибка обновления даты разбана в истории" in log_call
|
||
assert f"user_id={user_id}" in log_call
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_set_unban_date_update_exception(self, blacklist_history_repository):
|
||
"""Тест обработки исключения при обновлении записи"""
|
||
user_id = 12345
|
||
date_unban = int(time.time())
|
||
|
||
# Мокируем успешную проверку, но ошибку при обновлении
|
||
blacklist_history_repository._execute_query_with_result.return_value = [(100,)]
|
||
blacklist_history_repository._execute_query.side_effect = Exception(
|
||
"Update error"
|
||
)
|
||
|
||
result = await blacklist_history_repository.set_unban_date(user_id, date_unban)
|
||
|
||
# Проверяем, что метод вернул False при ошибке
|
||
assert result is False
|
||
|
||
# Проверяем логирование ошибки
|
||
blacklist_history_repository.logger.error.assert_called_once()
|
||
log_call = blacklist_history_repository.logger.error.call_args[0][0]
|
||
assert "Ошибка обновления даты разбана в истории" in log_call
|