306 lines
14 KiB
Python
306 lines
14 KiB
Python
import time
|
||
from datetime import datetime
|
||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||
|
||
import pytest
|
||
from database.models import Admin
|
||
from database.repositories.admin_repository import AdminRepository
|
||
|
||
|
||
class TestAdminRepository:
|
||
"""Тесты для AdminRepository"""
|
||
|
||
@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 admin_repository(self, mock_db_connection):
|
||
"""Экземпляр AdminRepository для тестов"""
|
||
# Патчим наследование от DatabaseConnection
|
||
with patch.object(AdminRepository, "__init__", return_value=None):
|
||
repo = AdminRepository()
|
||
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_admin(self):
|
||
"""Тестовый администратор"""
|
||
return Admin(user_id=12345, role="admin")
|
||
|
||
@pytest.fixture
|
||
def sample_admin_with_created_at(self):
|
||
"""Тестовый администратор с датой создания"""
|
||
return Admin(
|
||
user_id=12345, role="admin", created_at="1705312200" # UNIX timestamp
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables(self, admin_repository):
|
||
"""Тест создания таблицы администраторов"""
|
||
await admin_repository.create_tables()
|
||
|
||
# Проверяем, что включены внешние ключи
|
||
admin_repository._execute_query.assert_called()
|
||
calls = admin_repository._execute_query.call_args_list
|
||
|
||
# Первый вызов должен быть для включения внешних ключей
|
||
assert calls[0][0][0] == "PRAGMA foreign_keys = ON"
|
||
|
||
# Второй вызов должен быть для создания таблицы
|
||
create_table_call = calls[1]
|
||
assert "CREATE TABLE IF NOT EXISTS admins" in create_table_call[0][0]
|
||
assert "user_id INTEGER NOT NULL PRIMARY KEY" in create_table_call[0][0]
|
||
assert "role TEXT DEFAULT 'admin'" in create_table_call[0][0]
|
||
assert (
|
||
"created_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]
|
||
)
|
||
|
||
# Проверяем логирование
|
||
admin_repository.logger.info.assert_called_once_with(
|
||
"Таблица администраторов создана"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_admin(self, admin_repository, sample_admin):
|
||
"""Тест добавления администратора"""
|
||
await admin_repository.add_admin(sample_admin)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
admin_repository._execute_query.assert_called_once()
|
||
call_args = admin_repository._execute_query.call_args
|
||
|
||
assert call_args[0][0] == "INSERT INTO admins (user_id, role) VALUES (?, ?)"
|
||
assert call_args[0][1] == (12345, "admin")
|
||
|
||
# Проверяем логирование
|
||
admin_repository.logger.info.assert_called_once_with(
|
||
"Администратор добавлен: user_id=12345, role=admin"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_admin_with_custom_role(self, admin_repository):
|
||
"""Тест добавления администратора с кастомной ролью"""
|
||
admin = Admin(user_id=67890, role="super_admin")
|
||
await admin_repository.add_admin(admin)
|
||
|
||
call_args = admin_repository._execute_query.call_args
|
||
assert call_args[0][1] == (67890, "super_admin")
|
||
|
||
admin_repository.logger.info.assert_called_once_with(
|
||
"Администратор добавлен: user_id=67890, role=super_admin"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_admin(self, admin_repository):
|
||
"""Тест удаления администратора"""
|
||
user_id = 12345
|
||
await admin_repository.remove_admin(user_id)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
admin_repository._execute_query.assert_called_once()
|
||
call_args = admin_repository._execute_query.call_args
|
||
|
||
assert call_args[0][0] == "DELETE FROM admins WHERE user_id = ?"
|
||
assert call_args[0][1] == (user_id,)
|
||
|
||
# Проверяем логирование
|
||
admin_repository.logger.info.assert_called_once_with(
|
||
"Администратор удален: user_id=12345"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_is_admin_true(self, admin_repository):
|
||
"""Тест проверки администратора - пользователь является администратором"""
|
||
user_id = 12345
|
||
# Мокаем результат запроса - пользователь найден
|
||
admin_repository._execute_query_with_result.return_value = [(1,)]
|
||
|
||
result = await admin_repository.is_admin(user_id)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
admin_repository._execute_query_with_result.assert_called_once()
|
||
call_args = admin_repository._execute_query_with_result.call_args
|
||
|
||
assert call_args[0][0] == "SELECT 1 FROM admins WHERE user_id = ?"
|
||
assert call_args[0][1] == (user_id,)
|
||
|
||
# Проверяем результат
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_is_admin_false(self, admin_repository):
|
||
"""Тест проверки администратора - пользователь не является администратором"""
|
||
user_id = 12345
|
||
# Мокаем результат запроса - пользователь не найден
|
||
admin_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await admin_repository.is_admin(user_id)
|
||
|
||
# Проверяем результат
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_admin_found(self, admin_repository):
|
||
"""Тест получения информации об администраторе - администратор найден"""
|
||
user_id = 12345
|
||
# Мокаем результат запроса
|
||
admin_repository._execute_query_with_result.return_value = [
|
||
(12345, "admin", "1705312200")
|
||
]
|
||
|
||
result = await admin_repository.get_admin(user_id)
|
||
|
||
# Проверяем, что метод вызван с правильными параметрами
|
||
admin_repository._execute_query_with_result.assert_called_once()
|
||
call_args = admin_repository._execute_query_with_result.call_args
|
||
|
||
assert (
|
||
call_args[0][0]
|
||
== "SELECT user_id, role, created_at FROM admins WHERE user_id = ?"
|
||
)
|
||
assert call_args[0][1] == (user_id,)
|
||
|
||
# Проверяем результат
|
||
assert result is not None
|
||
assert result.user_id == 12345
|
||
assert result.role == "admin"
|
||
assert result.created_at == "1705312200"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_admin_not_found(self, admin_repository):
|
||
"""Тест получения информации об администраторе - администратор не найден"""
|
||
user_id = 12345
|
||
# Мокаем результат запроса - администратор не найден
|
||
admin_repository._execute_query_with_result.return_value = []
|
||
|
||
result = await admin_repository.get_admin(user_id)
|
||
|
||
# Проверяем результат
|
||
assert result is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_admin_without_created_at(self, admin_repository):
|
||
"""Тест получения информации об администраторе без даты создания"""
|
||
user_id = 12345
|
||
# Мокаем результат запроса без created_at
|
||
admin_repository._execute_query_with_result.return_value = [(12345, "admin")]
|
||
|
||
result = await admin_repository.get_admin(user_id)
|
||
|
||
# Проверяем результат
|
||
assert result is not None
|
||
assert result.user_id == 12345
|
||
assert result.role == "admin"
|
||
assert result.created_at is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_admin_error_handling(self, admin_repository, sample_admin):
|
||
"""Тест обработки ошибок при добавлении администратора"""
|
||
# Мокаем ошибку при выполнении запроса
|
||
admin_repository._execute_query.side_effect = Exception("Database error")
|
||
|
||
with pytest.raises(Exception, match="Database error"):
|
||
await admin_repository.add_admin(sample_admin)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_admin_error_handling(self, admin_repository):
|
||
"""Тест обработки ошибок при удалении администратора"""
|
||
# Мокаем ошибку при выполнении запроса
|
||
admin_repository._execute_query.side_effect = Exception("Database error")
|
||
|
||
with pytest.raises(Exception, match="Database error"):
|
||
await admin_repository.remove_admin(12345)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_is_admin_error_handling(self, admin_repository):
|
||
"""Тест обработки ошибок при проверке администратора"""
|
||
# Мокаем ошибку при выполнении запроса
|
||
admin_repository._execute_query_with_result.side_effect = Exception(
|
||
"Database error"
|
||
)
|
||
|
||
with pytest.raises(Exception, match="Database error"):
|
||
await admin_repository.is_admin(12345)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_admin_error_handling(self, admin_repository):
|
||
"""Тест обработки ошибок при получении информации об администраторе"""
|
||
# Мокаем ошибку при выполнении запроса
|
||
admin_repository._execute_query_with_result.side_effect = Exception(
|
||
"Database error"
|
||
)
|
||
|
||
with pytest.raises(Exception, match="Database error"):
|
||
await admin_repository.get_admin(12345)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables_error_handling(self, admin_repository):
|
||
"""Тест обработки ошибок при создании таблиц"""
|
||
# Мокаем ошибку при выполнении первого запроса
|
||
admin_repository._execute_query.side_effect = Exception("Database error")
|
||
|
||
with pytest.raises(Exception, match="Database error"):
|
||
await admin_repository.create_tables()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_model_compatibility(self, admin_repository):
|
||
"""Тест совместимости с моделью Admin"""
|
||
user_id = 12345
|
||
role = "moderator"
|
||
|
||
# Создаем администратора с помощью модели
|
||
admin = Admin(user_id=user_id, role=role)
|
||
|
||
# Проверяем, что можем передать его в репозиторий
|
||
await admin_repository.add_admin(admin)
|
||
|
||
# Проверяем, что вызов был с правильными параметрами
|
||
call_args = admin_repository._execute_query.call_args
|
||
assert call_args[0][1] == (user_id, role)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_multiple_admin_operations(self, admin_repository):
|
||
"""Тест множественных операций с администраторами"""
|
||
# Добавляем первого администратора
|
||
admin1 = Admin(user_id=111, role="admin")
|
||
await admin_repository.add_admin(admin1)
|
||
|
||
# Добавляем второго администратора
|
||
admin2 = Admin(user_id=222, role="moderator")
|
||
await admin_repository.add_admin(admin2)
|
||
|
||
# Проверяем, что оба добавлены
|
||
assert admin_repository._execute_query.call_count == 2
|
||
|
||
# Проверяем, что первый администратор существует
|
||
admin_repository._execute_query_with_result.return_value = [(1,)]
|
||
result1 = await admin_repository.is_admin(111)
|
||
assert result1 is True
|
||
|
||
# Проверяем, что второй администратор существует
|
||
result2 = await admin_repository.is_admin(222)
|
||
assert result2 is True
|
||
|
||
# Удаляем первого администратора
|
||
await admin_repository.remove_admin(111)
|
||
|
||
# Проверяем, что он больше не существует
|
||
admin_repository._execute_query_with_result.return_value = []
|
||
result3 = await admin_repository.is_admin(111)
|
||
assert result3 is False
|