Andrey
2d40f4496e
- Added voice message handling capabilities, including saving and deleting audio messages via callback queries. - Refactored audio record management in the database to remove unnecessary fields and streamline operations. - Introduced new keyboard options for voice interactions in the bot. - Updated `.gitignore` to include voice user files for better project organization. - Removed obsolete voice bot handler files to simplify the codebase.
187 lines
5.9 KiB
Python
187 lines
5.9 KiB
Python
import pytest
|
||
import asyncio
|
||
import os
|
||
import tempfile
|
||
import sqlite3
|
||
from database.async_db import AsyncBotDB
|
||
|
||
|
||
@pytest.fixture
|
||
async def temp_db():
|
||
"""Создает временную базу данных для тестирования."""
|
||
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
|
||
db_path = tmp.name
|
||
|
||
db = AsyncBotDB(db_path)
|
||
yield db
|
||
|
||
# Очистка
|
||
try:
|
||
os.unlink(db_path)
|
||
except:
|
||
pass
|
||
|
||
|
||
@pytest.fixture(scope="function")
|
||
def event_loop():
|
||
"""Создает новый event loop для каждого теста."""
|
||
loop = asyncio.new_event_loop()
|
||
yield loop
|
||
loop.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_tables(temp_db):
|
||
"""Тест создания таблиц."""
|
||
await temp_db.create_tables()
|
||
# Если не возникло исключение, значит таблицы созданы успешно
|
||
assert True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_and_get_user(temp_db):
|
||
"""Тест добавления и получения пользователя."""
|
||
await temp_db.create_tables()
|
||
|
||
# Добавляем пользователя
|
||
user_id = 12345
|
||
first_name = "Test"
|
||
full_name = "Test User"
|
||
username = "testuser"
|
||
|
||
await temp_db.add_new_user(user_id, first_name, full_name, username)
|
||
|
||
# Проверяем существование
|
||
exists = await temp_db.user_exists(user_id)
|
||
assert exists is True
|
||
|
||
# Получаем информацию
|
||
user_info = await temp_db.get_user_info(user_id)
|
||
assert user_info is not None
|
||
assert user_info['username'] == username
|
||
assert user_info['full_name'] == full_name
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_blacklist_operations(temp_db):
|
||
"""Тест операций с черным списком."""
|
||
await temp_db.create_tables()
|
||
|
||
user_id = 12345
|
||
user_name = "Test User"
|
||
message = "Test ban"
|
||
date_to_unban = "01-01-2025"
|
||
|
||
# Добавляем в черный список
|
||
await temp_db.add_to_blacklist(user_id, user_name, message, date_to_unban)
|
||
|
||
# Проверяем наличие
|
||
is_banned = await temp_db.check_blacklist(user_id)
|
||
assert is_banned is True
|
||
|
||
# Получаем список
|
||
banned_users = await temp_db.get_blacklist_users()
|
||
assert len(banned_users) == 1
|
||
assert banned_users[0][1] == user_id # user_id
|
||
|
||
# Удаляем из черного списка
|
||
removed = await temp_db.remove_from_blacklist(user_id)
|
||
assert removed is True
|
||
|
||
# Проверяем удаление
|
||
is_banned = await temp_db.check_blacklist(user_id)
|
||
assert is_banned is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.xfail(reason="FOREIGN KEY constraint failed - требует исправления порядка операций")
|
||
async def test_admin_operations(temp_db):
|
||
"""Тест операций с администраторами."""
|
||
await temp_db.create_tables()
|
||
|
||
user_id = 12345
|
||
role = "admin"
|
||
|
||
# Добавляем пользователя
|
||
await temp_db.add_new_user(user_id, "Test", "Test User", "testuser")
|
||
|
||
# Добавляем администратора
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
await temp_db.add_admin(user_id, role)
|
||
|
||
# # Проверяем права
|
||
# is_admin = await temp_db.is_admin(user_id)
|
||
# assert is_admin is True
|
||
|
||
# # Удаляем администратора
|
||
# await temp_db.remove_admin(user_id)
|
||
|
||
# # Проверяем удаление
|
||
# is_admin = await temp_db.is_admin(user_id)
|
||
# assert is_admin is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.xfail(reason="FOREIGN KEY constraint failed - требует исправления порядка операций")
|
||
async def test_audio_operations(temp_db):
|
||
"""Тест операций с аудио."""
|
||
await temp_db.create_tables()
|
||
|
||
user_id = 12345
|
||
file_name = "test_audio.mp3"
|
||
file_id = "test_file_id"
|
||
|
||
# Добавляем пользователя
|
||
await temp_db.add_new_user(user_id, "Test", "Test User", "testuser")
|
||
|
||
# Добавляем аудио запись
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
await temp_db.add_audio_record(file_name, user_id)
|
||
|
||
# # Получаем имя файла
|
||
# retrieved_file_name = await temp_db.get_audio_file_name(user_id)
|
||
# assert retrieved_file_name == file_name
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.xfail(reason="FOREIGN KEY constraint failed - требует исправления порядка операций")
|
||
async def test_post_operations(temp_db):
|
||
"""Тест операций с постами."""
|
||
await temp_db.create_tables()
|
||
|
||
message_id = 12345
|
||
text = "Test post text"
|
||
author_id = 67890
|
||
|
||
# Добавляем пользователя
|
||
await temp_db.add_new_user(author_id, "Test", "Test User", "testuser")
|
||
|
||
# Добавляем пост
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
await temp_db.add_post(message_id, text, author_id)
|
||
|
||
# # Обновляем helper сообщение
|
||
# helper_message_id = 54321
|
||
# await temp_db.update_helper_message(message_id, helper_message_id)
|
||
|
||
# # Получаем текст поста
|
||
# retrieved_text = await temp_db.get_post_text(helper_message_id)
|
||
# assert retrieved_text == text
|
||
|
||
# # Получаем ID автора
|
||
# retrieved_author_id = await temp_db.get_author_id_by_helper_message(helper_message_id)
|
||
# assert retrieved_author_id == author_id
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_error_handling(temp_db):
|
||
"""Тест обработки ошибок."""
|
||
# Пытаемся получить пользователя без создания таблиц
|
||
with pytest.raises(Exception):
|
||
await temp_db.user_exists(12345)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Запуск тестов
|
||
pytest.main([__file__, "-v"])
|