309 lines
12 KiB
Python
309 lines
12 KiB
Python
import asyncio
|
||
from datetime import datetime, timedelta, timezone
|
||
from unittest.mock import AsyncMock, Mock, patch
|
||
|
||
import pytest
|
||
|
||
from helper_bot.utils.auto_unban_scheduler import (
|
||
AutoUnbanScheduler,
|
||
get_auto_unban_scheduler,
|
||
)
|
||
|
||
|
||
class TestAutoUnbanScheduler:
|
||
"""Тесты для класса AutoUnbanScheduler"""
|
||
|
||
@pytest.fixture
|
||
def scheduler(self):
|
||
"""Создает экземпляр планировщика для тестов"""
|
||
return AutoUnbanScheduler()
|
||
|
||
@pytest.fixture
|
||
def mock_bot_db(self):
|
||
"""Создает мок базы данных"""
|
||
mock_db = Mock()
|
||
mock_db.get_users_for_unblock_today = AsyncMock(
|
||
return_value={123: "test_user1", 456: "test_user2"}
|
||
)
|
||
mock_db.delete_user_blacklist = AsyncMock(return_value=True)
|
||
return mock_db
|
||
|
||
@pytest.fixture
|
||
def mock_bdf(self):
|
||
"""Создает мок фабрики зависимостей"""
|
||
mock_factory = Mock()
|
||
mock_factory.settings = {
|
||
"Telegram": {
|
||
"group_for_logs": "-1001234567890",
|
||
"important_logs": "-1001234567891",
|
||
}
|
||
}
|
||
return mock_factory
|
||
|
||
@pytest.fixture
|
||
def mock_bot(self):
|
||
"""Создает мок бота"""
|
||
mock_bot = Mock()
|
||
mock_bot.send_message = AsyncMock()
|
||
return mock_bot
|
||
|
||
def test_scheduler_initialization(self, scheduler):
|
||
"""Тест инициализации планировщика"""
|
||
assert scheduler.bot_db is not None
|
||
assert scheduler.scheduler is not None
|
||
assert scheduler.bot is None
|
||
|
||
def test_set_bot(self, scheduler, mock_bot):
|
||
"""Тест установки бота"""
|
||
scheduler.set_bot(mock_bot)
|
||
assert scheduler.bot == mock_bot
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_auto_unban_users_success(
|
||
self, mock_get_instance, scheduler, mock_bot_db, mock_bdf, mock_bot
|
||
):
|
||
"""Тест успешного выполнения автоматического разбана"""
|
||
# Настройка моков
|
||
mock_get_instance.return_value = mock_bdf
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
# Выполнение теста
|
||
await scheduler.auto_unban_users()
|
||
|
||
# Проверки
|
||
mock_bot_db.get_users_for_unblock_today.assert_called_once()
|
||
assert mock_bot_db.delete_user_blacklist.call_count == 2
|
||
mock_bot.send_message.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_auto_unban_users_no_users(
|
||
self, mock_get_instance, scheduler, mock_bot_db, mock_bdf, mock_bot
|
||
):
|
||
"""Тест разбана когда нет пользователей для разблокировки"""
|
||
# Настройка моков
|
||
mock_get_instance.return_value = mock_bdf
|
||
mock_bot_db.get_users_for_unblock_today = AsyncMock(return_value={})
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
# Выполнение теста
|
||
await scheduler.auto_unban_users()
|
||
|
||
# Проверки
|
||
mock_bot_db.get_users_for_unblock_today.assert_called_once()
|
||
mock_bot_db.delete_user_blacklist.assert_not_called()
|
||
mock_bot.send_message.assert_not_called()
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_auto_unban_users_partial_failure(
|
||
self, mock_get_instance, scheduler, mock_bot_db, mock_bdf, mock_bot
|
||
):
|
||
"""Тест разбана с частичными ошибками"""
|
||
# Настройка моков
|
||
mock_get_instance.return_value = mock_bdf
|
||
mock_bot_db.get_users_for_unblock_today = AsyncMock(
|
||
return_value={123: "test_user1", 456: "test_user2"}
|
||
)
|
||
# Первый вызов успешен, второй - ошибка
|
||
mock_bot_db.delete_user_blacklist = AsyncMock(side_effect=[True, False])
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
# Выполнение теста
|
||
await scheduler.auto_unban_users()
|
||
|
||
# Проверки
|
||
assert mock_bot_db.delete_user_blacklist.call_count == 2
|
||
mock_bot.send_message.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_auto_unban_users_exception(
|
||
self, mock_get_instance, scheduler, mock_bot_db, mock_bdf, mock_bot
|
||
):
|
||
"""Тест разбана с исключением"""
|
||
# Настройка моков
|
||
mock_get_instance.return_value = mock_bdf
|
||
mock_bot_db.get_users_for_unblock_today = AsyncMock(
|
||
side_effect=Exception("Database error")
|
||
)
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
# Выполнение теста
|
||
await scheduler.auto_unban_users()
|
||
|
||
# Проверки
|
||
mock_bot.send_message.assert_called_once()
|
||
# Проверяем, что сообщение об ошибке было отправлено
|
||
call_args = mock_bot.send_message.call_args
|
||
assert "Ошибка автоматического разбана" in call_args[1]["text"]
|
||
|
||
def test_generate_report(self, scheduler):
|
||
"""Тест генерации отчета"""
|
||
users = {123: "test_user1", 456: "test_user2"}
|
||
failed_users = ["456 (test_user2)"]
|
||
|
||
report = scheduler._generate_report(1, 1, failed_users, users)
|
||
|
||
assert "Отчет об автоматическом разбане" in report
|
||
assert "Успешно разблокировано: 1" in report
|
||
assert "Ошибок: 1" in report
|
||
assert "ID: 123" in report
|
||
assert "456 (test_user2)" in report
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_send_report(self, mock_get_instance, scheduler, mock_bdf, mock_bot):
|
||
"""Тест отправки отчета"""
|
||
mock_get_instance.return_value = mock_bdf
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
report = "Test report"
|
||
await scheduler._send_report(report)
|
||
|
||
# Проверяем, что send_message был вызван
|
||
mock_bot.send_message.assert_called_once()
|
||
|
||
# Проверяем аргументы вызова
|
||
call_args = mock_bot.send_message.call_args
|
||
assert call_args[1]["text"] == report
|
||
assert call_args[1]["parse_mode"] == "HTML"
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_send_error_report(
|
||
self, mock_get_instance, scheduler, mock_bdf, mock_bot
|
||
):
|
||
"""Тест отправки отчета об ошибке"""
|
||
mock_get_instance.return_value = mock_bdf
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
error_msg = "Test error"
|
||
await scheduler._send_error_report(error_msg)
|
||
|
||
# Проверяем, что send_message был вызван
|
||
mock_bot.send_message.assert_called_once()
|
||
|
||
# Проверяем аргументы вызова
|
||
call_args = mock_bot.send_message.call_args
|
||
assert "Ошибка автоматического разбана" in call_args[1]["text"]
|
||
assert error_msg in call_args[1]["text"]
|
||
assert call_args[1]["parse_mode"] == "HTML"
|
||
|
||
def test_start_scheduler(self, scheduler):
|
||
"""Тест запуска планировщика"""
|
||
with (
|
||
patch.object(scheduler.scheduler, "add_job") as mock_add_job,
|
||
patch.object(scheduler.scheduler, "start") as mock_start,
|
||
):
|
||
|
||
scheduler.start_scheduler()
|
||
|
||
mock_add_job.assert_called_once()
|
||
mock_start.assert_called_once()
|
||
|
||
def test_stop_scheduler(self, scheduler):
|
||
"""Тест остановки планировщика"""
|
||
# Сначала запускаем планировщик
|
||
scheduler.start_scheduler()
|
||
|
||
# Проверяем, что планировщик запущен
|
||
assert scheduler.scheduler.running
|
||
|
||
# Теперь останавливаем (должно пройти без ошибок)
|
||
scheduler.stop_scheduler()
|
||
|
||
# Проверяем, что метод выполнился без исключений
|
||
# APScheduler может не сразу остановиться, но это нормально
|
||
|
||
@pytest.mark.asyncio
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_run_manual_unban(
|
||
self, mock_get_instance, scheduler, mock_bot_db, mock_bdf, mock_bot
|
||
):
|
||
"""Тест ручного запуска разбана"""
|
||
mock_get_instance.return_value = mock_bdf
|
||
mock_bot_db.get_users_for_unblock_today.return_value = {}
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
await scheduler.run_manual_unban()
|
||
|
||
mock_bot_db.get_users_for_unblock_today.assert_called_once()
|
||
|
||
|
||
class TestGetAutoUnbanScheduler:
|
||
"""Тесты для функции get_auto_unban_scheduler"""
|
||
|
||
def test_get_auto_unban_scheduler(self):
|
||
"""Тест получения глобального экземпляра планировщика"""
|
||
scheduler = get_auto_unban_scheduler()
|
||
assert isinstance(scheduler, AutoUnbanScheduler)
|
||
|
||
# Проверяем, что возвращается один и тот же экземпляр
|
||
scheduler2 = get_auto_unban_scheduler()
|
||
assert scheduler is scheduler2
|
||
|
||
|
||
class TestDateHandling:
|
||
"""Тесты для обработки дат"""
|
||
|
||
def test_moscow_timezone(self):
|
||
"""Тест работы с московским временем"""
|
||
scheduler = AutoUnbanScheduler()
|
||
|
||
# Проверяем, что дата формируется в правильном формате
|
||
moscow_tz = timezone(timedelta(hours=3))
|
||
today = datetime.now(moscow_tz).strftime("%Y-%m-%d")
|
||
|
||
assert len(today) == 10 # YYYY-MM-DD
|
||
assert today.count("-") == 2
|
||
assert today[:4].isdigit() # Год
|
||
assert today[5:7].isdigit() # Месяц
|
||
assert today[8:10].isdigit() # День
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestAsyncOperations:
|
||
"""Тесты асинхронных операций"""
|
||
|
||
@patch("helper_bot.utils.auto_unban_scheduler.get_global_instance")
|
||
async def test_async_auto_unban_flow(self, mock_get_instance):
|
||
"""Тест полного асинхронного потока разбана"""
|
||
# Создаем моки
|
||
mock_bdf = Mock()
|
||
mock_bdf.settings = {
|
||
"Telegram": {
|
||
"group_for_logs": "-1001234567890",
|
||
"important_logs": "-1001234567891",
|
||
}
|
||
}
|
||
mock_get_instance.return_value = mock_bdf
|
||
|
||
mock_bot_db = Mock()
|
||
mock_bot_db.get_users_for_unblock_today = AsyncMock(
|
||
return_value={123: "test_user"}
|
||
)
|
||
mock_bot_db.delete_user_blacklist = AsyncMock(return_value=True)
|
||
|
||
mock_bot = Mock()
|
||
mock_bot.send_message = AsyncMock()
|
||
|
||
# Создаем планировщик
|
||
scheduler = AutoUnbanScheduler()
|
||
scheduler.bot_db = mock_bot_db
|
||
scheduler.set_bot(mock_bot)
|
||
|
||
# Выполняем разбан
|
||
await scheduler.auto_unban_users()
|
||
|
||
# Проверяем результаты
|
||
mock_bot_db.get_users_for_unblock_today.assert_called_once()
|
||
mock_bot_db.delete_user_blacklist.assert_called_once_with(123)
|
||
mock_bot.send_message.assert_called_once()
|