Enhance bot functionality and refactor database interactions
- Added `ca-certificates` installation to Dockerfile for improved network security. - Updated health check command in Dockerfile to include better timeout handling. - Refactored `run_helper.py` to implement proper signal handling and logging during shutdown. - Transitioned database operations to an asynchronous model in `async_db.py`, improving performance and responsiveness. - Updated database schema to support new foreign key relationships and optimized indexing for better query performance. - Enhanced various bot handlers to utilize async database methods, improving overall efficiency and user experience. - Removed obsolete database and fix scripts to streamline the project structure.
This commit is contained in:
295
tests/test_admin_repository.py
Normal file
295
tests/test_admin_repository.py
Normal file
@@ -0,0 +1,295 @@
|
||||
import pytest
|
||||
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
||||
from datetime import datetime
|
||||
import time
|
||||
|
||||
from database.repositories.admin_repository import AdminRepository
|
||||
from database.models import Admin
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user