refactor
This commit is contained in:
277
tests/test_new_features.py
Normal file
277
tests/test_new_features.py
Normal file
@@ -0,0 +1,277 @@
|
||||
"""
|
||||
Тест новой функциональности: клиентские счета и многопоточность
|
||||
"""
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Добавляем путь к проекту
|
||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from src.core.database import init_db, async_session_maker
|
||||
from src.core.services import UserService, LotteryService
|
||||
from src.utils.account_utils import generate_account_number, validate_account_number, mask_account_number
|
||||
from src.utils.task_manager import task_manager, TaskPriority
|
||||
from src.display.winner_display import format_winner_display, validate_display_type
|
||||
from src.utils.async_decorators import get_task_stats, format_task_stats
|
||||
import time
|
||||
|
||||
|
||||
async def test_account_functionality():
|
||||
"""Тест функций работы со счетами"""
|
||||
print("🧪 Тестирование функций работы со счетами")
|
||||
print("=" * 50)
|
||||
|
||||
async with async_session_maker() as session:
|
||||
# Создаём тестовых пользователей
|
||||
users_data = [
|
||||
{
|
||||
'telegram_id': 777000001,
|
||||
'username': 'client1',
|
||||
'first_name': 'Клиент',
|
||||
'last_name': 'Первый',
|
||||
'account': '12-34-56-78-90-12-34-56'
|
||||
},
|
||||
{
|
||||
'telegram_id': 777000002,
|
||||
'username': 'client2',
|
||||
'first_name': 'Клиент',
|
||||
'last_name': 'Второй',
|
||||
'account': '11-22-33-44-55-66-77-88'
|
||||
},
|
||||
{
|
||||
'telegram_id': 777000003,
|
||||
'username': 'client3',
|
||||
'first_name': 'Клиент',
|
||||
'last_name': 'Третий',
|
||||
'account': None # Без счёта
|
||||
}
|
||||
]
|
||||
|
||||
created_users = []
|
||||
|
||||
for user_data in users_data:
|
||||
# Создаём пользователя
|
||||
user = await UserService.get_or_create_user(
|
||||
session,
|
||||
telegram_id=user_data['telegram_id'],
|
||||
username=user_data['username'],
|
||||
first_name=user_data['first_name'],
|
||||
last_name=user_data['last_name']
|
||||
)
|
||||
|
||||
# Устанавливаем номер счёта, если есть
|
||||
if user_data['account']:
|
||||
success = await UserService.set_account_number(
|
||||
session, user.telegram_id, user_data['account']
|
||||
)
|
||||
if success:
|
||||
print(f"✅ Создан пользователь {user.first_name} со счётом {user_data['account']}")
|
||||
else:
|
||||
print(f"❌ Ошибка установки счёта для {user.first_name}")
|
||||
else:
|
||||
print(f"✅ Создан пользователь {user.first_name} без счёта")
|
||||
|
||||
created_users.append(user)
|
||||
|
||||
# Тесты валидации
|
||||
print("\n📋 Тестирование валидации номеров счетов:")
|
||||
|
||||
test_numbers = [
|
||||
("12-34-56-78-90-12-34-56", True), # Корректный
|
||||
("12345678901234567890123456", False), # Без дефисов
|
||||
("12-34-56-78-90-12-34", False), # Короткий
|
||||
("12-34-56-78-90-12-34-56-78", False), # Длинный
|
||||
("ab-cd-ef-gh-ij-kl-mn-op", False), # Буквы
|
||||
("", False) # Пустой
|
||||
]
|
||||
|
||||
for number, should_be_valid in test_numbers:
|
||||
is_valid = validate_account_number(number)
|
||||
status = "✅" if is_valid == should_be_valid else "❌"
|
||||
print(f"{status} {number} -> {'Корректный' if is_valid else 'Некорректный'}")
|
||||
|
||||
# Тест маскирования
|
||||
print("\n🎭 Тестирование маскирования номеров:")
|
||||
test_account = "12-34-56-78-90-12-34-56"
|
||||
for digits in [4, 6, 8]:
|
||||
masked = mask_account_number(test_account, show_last_digits=digits)
|
||||
print(f"Показать последние {digits} цифр: {masked}")
|
||||
|
||||
# Тест поиска по счёту
|
||||
print("\n🔍 Тестирование поиска по номеру счёта:")
|
||||
|
||||
async with async_session_maker() as session:
|
||||
# Полный номер
|
||||
user = await UserService.get_user_by_account(session, "12-34-56-78-90-12-34-56")
|
||||
if user:
|
||||
print(f"✅ Найден пользователь по полному номеру: {user.first_name}")
|
||||
|
||||
# Поиск по части номера
|
||||
users = await UserService.search_by_account(session, "12-34")
|
||||
print(f"🔍 Найдено пользователей по шаблону '12-34': {len(users)}")
|
||||
|
||||
for user in users:
|
||||
if user.account_number:
|
||||
masked = mask_account_number(user.account_number, show_last_digits=6)
|
||||
print(f" • {user.first_name}: {masked}")
|
||||
|
||||
|
||||
async def test_display_types():
|
||||
"""Тест различных типов отображения победителей"""
|
||||
print("\n🎨 Тестирование типов отображения победителей")
|
||||
print("=" * 50)
|
||||
|
||||
async with async_session_maker() as session:
|
||||
# Создаём розыгрыш
|
||||
lottery = await LotteryService.create_lottery(
|
||||
session,
|
||||
title="Тест отображения победителей",
|
||||
description="Розыгрыш для тестирования различных типов отображения",
|
||||
prizes=["Первый приз", "Второй приз"],
|
||||
creator_id=1
|
||||
)
|
||||
|
||||
# Получаем пользователя со счётом
|
||||
user = await UserService.get_user_by_account(session, "12-34-56-78-90-12-34-56")
|
||||
|
||||
if user and lottery:
|
||||
print(f"👤 Тестируем отображение для пользователя: {user.first_name}")
|
||||
print(f"💳 Номер счёта: {user.account_number}")
|
||||
|
||||
# Тестируем разные типы отображения
|
||||
display_types = ['username', 'chat_id', 'account_number']
|
||||
|
||||
for display_type in display_types:
|
||||
# Устанавливаем тип отображения
|
||||
success = await LotteryService.set_winner_display_type(
|
||||
session, lottery.id, display_type
|
||||
)
|
||||
|
||||
if success:
|
||||
# Получаем обновлённый розыгрыш
|
||||
updated_lottery = await LotteryService.get_lottery(session, lottery.id)
|
||||
|
||||
# Тестируем отображение
|
||||
public_display = format_winner_display(user, updated_lottery, show_sensitive_data=False)
|
||||
admin_display = format_winner_display(user, updated_lottery, show_sensitive_data=True)
|
||||
|
||||
print(f"\n📺 Тип отображения: {display_type}")
|
||||
print(f" 👥 Публичное: {public_display}")
|
||||
print(f" 🔧 Админское: {admin_display}")
|
||||
else:
|
||||
print(f"❌ Ошибка установки типа отображения {display_type}")
|
||||
else:
|
||||
print("❌ Не удалось создать тестовые данные")
|
||||
|
||||
|
||||
async def test_multithreading():
|
||||
"""Тест многопоточности"""
|
||||
print("\n⚡ Тестирование системы многопоточности")
|
||||
print("=" * 50)
|
||||
|
||||
# Запускаем менеджер задач
|
||||
await task_manager.start()
|
||||
|
||||
# Функция-заглушка для тестирования
|
||||
async def test_task(task_name: str, duration: float, user_id_arg: int):
|
||||
print(f"🔄 Запуск задачи {task_name} для пользователя {user_id_arg}")
|
||||
await asyncio.sleep(duration)
|
||||
print(f"✅ Завершена задача {task_name} для пользователя {user_id_arg}")
|
||||
return f"Результат {task_name}"
|
||||
|
||||
# Добавляем задачи разных приоритетов
|
||||
tasks = [
|
||||
("critical_task", 2.0, 100, TaskPriority.CRITICAL),
|
||||
("high_task_1", 1.5, 200, TaskPriority.HIGH),
|
||||
("normal_task_1", 1.0, 300, TaskPriority.NORMAL),
|
||||
("normal_task_2", 1.2, 300, TaskPriority.NORMAL),
|
||||
("high_task_2", 0.8, 200, TaskPriority.HIGH),
|
||||
("low_task", 0.5, 400, TaskPriority.LOW)
|
||||
]
|
||||
|
||||
print(f"📤 Добавляем {len(tasks)} задач...")
|
||||
|
||||
for task_name, duration, user_id, priority in tasks:
|
||||
try:
|
||||
task_id = await task_manager.add_task(
|
||||
task_id=f"{task_name}_{int(time.time())}",
|
||||
user_id=user_id,
|
||||
func=test_task,
|
||||
priority=priority,
|
||||
timeout=10.0,
|
||||
task_name=task_name,
|
||||
duration=duration,
|
||||
user_id_arg=user_id
|
||||
)
|
||||
print(f" ➕ Добавлена задача {task_name} (приоритет: {priority.name})")
|
||||
except ValueError as e:
|
||||
print(f" ❌ Ошибка добавления задачи {task_name}: {e}")
|
||||
|
||||
# Показываем статистику
|
||||
print("\n📊 Начальная статистика:")
|
||||
stats_text = await format_task_stats()
|
||||
print(stats_text)
|
||||
|
||||
# Ждём выполнения задач
|
||||
print("\n⏳ Ожидание выполнения задач...")
|
||||
await asyncio.sleep(5.0)
|
||||
|
||||
# Показываем финальную статистику
|
||||
print("\n📈 Финальная статистика:")
|
||||
final_stats = await format_task_stats()
|
||||
print(final_stats)
|
||||
|
||||
# Тестируем лимиты пользователя
|
||||
print("\n🚧 Тестирование лимитов пользователя:")
|
||||
|
||||
# Пробуем добавить много задач одному пользователю
|
||||
user_id = 999
|
||||
added_tasks = 0
|
||||
|
||||
for i in range(8): # Больше чем лимит (5)
|
||||
try:
|
||||
await task_manager.add_task(
|
||||
task_id=f"limit_test_{i}_{int(time.time())}",
|
||||
user_id=user_id,
|
||||
func=test_task,
|
||||
priority=TaskPriority.NORMAL,
|
||||
task_name=f"limit_test_{i}",
|
||||
duration=0.1,
|
||||
user_id_arg=user_id
|
||||
)
|
||||
added_tasks += 1
|
||||
except ValueError as e:
|
||||
print(f" 🛑 Лимит достигнут после {added_tasks} задач: {e}")
|
||||
break
|
||||
|
||||
# Ждём немного и останавливаем менеджер
|
||||
await asyncio.sleep(2.0)
|
||||
await task_manager.stop()
|
||||
|
||||
|
||||
async def main():
|
||||
"""Главная функция теста"""
|
||||
print("🚀 Запуск тестирования новой функциональности")
|
||||
print("=" * 60)
|
||||
|
||||
# Инициализируем базу данных
|
||||
await init_db()
|
||||
|
||||
try:
|
||||
# Запускаем тесты
|
||||
await test_account_functionality()
|
||||
await test_display_types()
|
||||
await test_multithreading()
|
||||
|
||||
print("\n🎉 Все тесты завершены успешно!")
|
||||
print("=" * 60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Ошибка во время тестирования: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user