Add auto unban functionality and update related tests and dependencies
This commit is contained in:
251
tests/test_auto_unban_integration.py
Normal file
251
tests/test_auto_unban_integration.py
Normal file
@@ -0,0 +1,251 @@
|
||||
import pytest
|
||||
import sqlite3
|
||||
import os
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from unittest.mock import Mock, patch, AsyncMock
|
||||
|
||||
from helper_bot.utils.auto_unban_scheduler import AutoUnbanScheduler
|
||||
|
||||
|
||||
class TestAutoUnbanIntegration:
|
||||
"""Интеграционные тесты для автоматического разбана"""
|
||||
|
||||
@pytest.fixture
|
||||
def test_db_path(self):
|
||||
"""Путь к тестовой базе данных"""
|
||||
return 'database/test_auto_unban.db'
|
||||
|
||||
@pytest.fixture
|
||||
def setup_test_db(self, test_db_path):
|
||||
"""Создает тестовую базу данных с таблицей blacklist"""
|
||||
# Удаляем старую тестовую базу если она существует
|
||||
if os.path.exists(test_db_path):
|
||||
os.remove(test_db_path)
|
||||
|
||||
# Создаем новую базу данных
|
||||
conn = sqlite3.connect(test_db_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Создаем таблицу blacklist
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS blacklist (
|
||||
user_id INTEGER PRIMARY KEY,
|
||||
user_name TEXT,
|
||||
message_for_user TEXT,
|
||||
date_to_unban TEXT
|
||||
)
|
||||
''')
|
||||
|
||||
# Добавляем тестовые данные
|
||||
today = datetime.now(timezone(timedelta(hours=3))).strftime("%Y-%m-%d")
|
||||
tomorrow = (datetime.now(timezone(timedelta(hours=3))) + timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
|
||||
test_data = [
|
||||
(123, "test_user1", "Test ban 1", today), # Разблокируется сегодня
|
||||
(456, "test_user2", "Test ban 2", today), # Разблокируется сегодня
|
||||
(789, "test_user3", "Test ban 3", tomorrow), # Разблокируется завтра
|
||||
(999, "test_user4", "Test ban 4", None), # Навсегда заблокирован
|
||||
]
|
||||
|
||||
cursor.executemany(
|
||||
"INSERT INTO blacklist (user_id, user_name, message_for_user, date_to_unban) VALUES (?, ?, ?, ?)",
|
||||
test_data
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
yield test_db_path
|
||||
|
||||
# Очистка после тестов
|
||||
if os.path.exists(test_db_path):
|
||||
os.remove(test_db_path)
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bdf(self, test_db_path):
|
||||
"""Создает мок фабрики зависимостей с тестовой базой"""
|
||||
mock_factory = Mock()
|
||||
mock_factory.settings = {
|
||||
'Telegram': {
|
||||
'group_for_logs': '-1001234567890',
|
||||
'important_logs': '-1001234567891'
|
||||
}
|
||||
}
|
||||
|
||||
# Создаем реальный экземпляр базы данных с тестовым файлом
|
||||
from database.db import BotDB
|
||||
import os
|
||||
current_dir = os.getcwd()
|
||||
mock_factory.database = BotDB(current_dir, test_db_path)
|
||||
|
||||
return mock_factory
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bot(self):
|
||||
"""Создает мок бота"""
|
||||
mock_bot = Mock()
|
||||
mock_bot.send_message = AsyncMock()
|
||||
return mock_bot
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('helper_bot.utils.auto_unban_scheduler.get_global_instance')
|
||||
async def test_auto_unban_with_real_db(self, mock_get_instance, setup_test_db, mock_bdf, mock_bot):
|
||||
"""Тест автоматического разбана с реальной базой данных"""
|
||||
# Настройка моков
|
||||
mock_get_instance.return_value = mock_bdf
|
||||
|
||||
# Создаем планировщик
|
||||
scheduler = AutoUnbanScheduler()
|
||||
scheduler.bot_db = mock_bdf.database
|
||||
scheduler.set_bot(mock_bot)
|
||||
|
||||
# Проверяем начальное состояние базы
|
||||
conn = sqlite3.connect(setup_test_db)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM blacklist")
|
||||
initial_count = cursor.fetchone()[0]
|
||||
assert initial_count == 4
|
||||
|
||||
# Выполняем автоматический разбан
|
||||
await scheduler.auto_unban_users()
|
||||
|
||||
# Проверяем, что пользователи с сегодняшней датой разблокированы
|
||||
cursor.execute("SELECT COUNT(*) FROM blacklist WHERE date_to_unban = ?",
|
||||
(datetime.now(timezone(timedelta(hours=3))).strftime("%Y-%m-%d"),))
|
||||
today_count = cursor.fetchone()[0]
|
||||
assert today_count == 0
|
||||
|
||||
# Проверяем, что пользователи с завтрашней датой остались
|
||||
tomorrow = (datetime.now(timezone(timedelta(hours=3))) + timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
cursor.execute("SELECT COUNT(*) FROM blacklist WHERE date_to_unban = ?", (tomorrow,))
|
||||
tomorrow_count = cursor.fetchone()[0]
|
||||
assert tomorrow_count == 1
|
||||
|
||||
# Проверяем, что навсегда заблокированные пользователи остались
|
||||
cursor.execute("SELECT COUNT(*) FROM blacklist WHERE date_to_unban IS NULL")
|
||||
permanent_count = cursor.fetchone()[0]
|
||||
assert permanent_count == 1
|
||||
|
||||
# Проверяем общее количество записей
|
||||
cursor.execute("SELECT COUNT(*) FROM blacklist")
|
||||
final_count = cursor.fetchone()[0]
|
||||
assert final_count == 2 # Остались только завтрашние и навсегда заблокированные
|
||||
|
||||
conn.close()
|
||||
|
||||
# Проверяем, что отчет был отправлен
|
||||
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_no_users_today(self, mock_get_instance, setup_test_db, mock_bdf, mock_bot):
|
||||
"""Тест разбана когда нет пользователей для разблокировки сегодня"""
|
||||
# Настройка моков
|
||||
mock_get_instance.return_value = mock_bdf
|
||||
|
||||
# Удаляем пользователей с сегодняшней датой
|
||||
conn = sqlite3.connect(setup_test_db)
|
||||
cursor = conn.cursor()
|
||||
today = datetime.now(timezone(timedelta(hours=3))).strftime("%Y-%m-%d")
|
||||
cursor.execute("DELETE FROM blacklist WHERE date_to_unban = ?", (today,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Создаем планировщик
|
||||
scheduler = AutoUnbanScheduler()
|
||||
scheduler.bot_db = mock_bdf.database
|
||||
scheduler.set_bot(mock_bot)
|
||||
|
||||
# Выполняем автоматический разбан
|
||||
await scheduler.auto_unban_users()
|
||||
|
||||
# Проверяем, что отчет не был отправлен (нет пользователей для разблокировки)
|
||||
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_database_error(self, mock_get_instance, setup_test_db, mock_bdf, mock_bot):
|
||||
"""Тест обработки ошибок базы данных"""
|
||||
# Настройка моков
|
||||
mock_get_instance.return_value = mock_bdf
|
||||
|
||||
# Создаем планировщик
|
||||
scheduler = AutoUnbanScheduler()
|
||||
scheduler.bot_db = mock_bdf.database
|
||||
scheduler.set_bot(mock_bot)
|
||||
|
||||
# Удаляем таблицу чтобы вызвать ошибку
|
||||
conn = sqlite3.connect(setup_test_db)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DROP TABLE blacklist")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Выполняем автоматический разбан
|
||||
await scheduler.auto_unban_users()
|
||||
|
||||
# Проверяем, что отчет об ошибке был отправлен
|
||||
mock_bot.send_message.assert_called_once()
|
||||
call_args = mock_bot.send_message.call_args
|
||||
assert call_args[1]['chat_id'] == '-1001234567891' # important_logs
|
||||
assert "Ошибка автоматического разбана" in call_args[1]['text']
|
||||
|
||||
def test_date_format_consistency(self, setup_test_db, mock_bdf):
|
||||
"""Тест консистентности формата дат"""
|
||||
scheduler = AutoUnbanScheduler()
|
||||
scheduler.bot_db = mock_bdf.database
|
||||
|
||||
# Проверяем, что дата в базе соответствует ожидаемому формату
|
||||
conn = sqlite3.connect(setup_test_db)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT date_to_unban FROM blacklist WHERE date_to_unban IS NOT NULL LIMIT 1")
|
||||
result = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
if result and result[0]:
|
||||
date_str = result[0]
|
||||
# Проверяем формат YYYY-MM-DD
|
||||
assert len(date_str) == 10
|
||||
assert date_str.count('-') == 2
|
||||
assert date_str[:4].isdigit() # Год
|
||||
assert date_str[5:7].isdigit() # Месяц
|
||||
assert date_str[8:10].isdigit() # День
|
||||
|
||||
|
||||
class TestSchedulerLifecycle:
|
||||
"""Тесты жизненного цикла планировщика"""
|
||||
|
||||
def test_scheduler_start_stop(self):
|
||||
"""Тест запуска и остановки планировщика"""
|
||||
scheduler = AutoUnbanScheduler()
|
||||
|
||||
# Запускаем планировщик
|
||||
scheduler.start_scheduler()
|
||||
assert scheduler.scheduler.running
|
||||
|
||||
# Останавливаем планировщик (должно пройти без ошибок)
|
||||
scheduler.stop_scheduler()
|
||||
# APScheduler может не сразу остановиться, но это нормально
|
||||
|
||||
def test_scheduler_job_creation(self):
|
||||
"""Тест создания задачи в планировщике"""
|
||||
scheduler = AutoUnbanScheduler()
|
||||
|
||||
with patch.object(scheduler.scheduler, 'add_job') as mock_add_job:
|
||||
scheduler.start_scheduler()
|
||||
|
||||
# Проверяем, что задача была создана с правильными параметрами
|
||||
mock_add_job.assert_called_once()
|
||||
call_args = mock_add_job.call_args
|
||||
|
||||
# Проверяем функцию
|
||||
assert call_args[0][0] == scheduler.auto_unban_users
|
||||
|
||||
# Проверяем триггер (должен быть CronTrigger)
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
assert isinstance(call_args[0][1], CronTrigger)
|
||||
|
||||
# Проверяем ID и имя задачи
|
||||
assert call_args[1]['id'] == 'auto_unban_users'
|
||||
assert call_args[1]['name'] == 'Автоматический разбан пользователей'
|
||||
assert call_args[1]['replace_existing'] is True
|
||||
288
tests/test_auto_unban_scheduler.py
Normal file
288
tests/test_auto_unban_scheduler.py
Normal file
@@ -0,0 +1,288 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from unittest.mock import Mock, patch, AsyncMock
|
||||
|
||||
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.return_value = {
|
||||
123: "test_user1",
|
||||
456: "test_user2"
|
||||
}
|
||||
mock_db.delete_user_blacklist.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.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.return_value = {
|
||||
123: "test_user1",
|
||||
456: "test_user2"
|
||||
}
|
||||
# Первый вызов успешен, второй - ошибка
|
||||
mock_bot_db.delete_user_blacklist.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.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 "test_user1" 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.return_value = {123: "test_user"}
|
||||
mock_bot_db.delete_user_blacklist.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()
|
||||
@@ -512,30 +512,7 @@ def test_update_info_about_stickers_error(bot):
|
||||
bot.update_info_about_stickers(12345)
|
||||
|
||||
|
||||
def test_get_users_blacklist_empty(bot):
|
||||
"""Проверяет, что функция возвращает пустой словарь, если в черном списке нет пользователей."""
|
||||
conn = sqlite3.connect('database/test.db')
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM blacklist")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
blacklist = bot.get_users_blacklist()
|
||||
assert blacklist == {}
|
||||
|
||||
|
||||
def test_get_users_blacklist_non_empty(bot):
|
||||
"""Проверяет, что функция возвращает словарь с пользователями из черного списка."""
|
||||
blacklist = bot.get_users_blacklist()
|
||||
assert blacklist == {12345: "@iban", 14278: "@boris"}
|
||||
|
||||
|
||||
def test_get_users_blacklist_error(bot):
|
||||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||||
__drop_table('blacklist')
|
||||
|
||||
with pytest.raises(sqlite3.Error):
|
||||
bot.get_users_blacklist()
|
||||
|
||||
|
||||
def test_get_blacklist_users_by_id_found(bot, setup_db):
|
||||
|
||||
Reference in New Issue
Block a user