Тестирование и отладка

Проблема: Вы изменили код бота, чтобы добавить новую фичу. Через час пользователи пишут: "Перестала работать оплата!", "Регистрация сломалась!". Как этого избежать?

Реальность без тестов:

Баги обнаруживаются пользователями в продакшене

Страх менять код — "а вдруг что-то сломаю?"

Ручное тестирование занимает часы

Невозможно откатить — неясно, что работало раньше

Решение: Автоматические тесты ловят баги до продакшена. Unit-тесты проверяют отдельные функции, интеграционные — взаимодействие компонентов, E2E — полные сценарии пользователя.

Зачем это нужно: Тесты — это страховка. Они дают уверенность менять код, ускоряют разработку (не нужно вручную проверять всё каждый раз), документируют ожидаемое поведение.

#Типы тестов

Тип	Что тестирует	Скорость	Изоляция	Когда использовать
Unit	Отдельные функции, классы	⚡ Быстро (мс)	✅ Полная	Логика сервисов, фильтры, валидация
Integration	Взаимодействие компонентов	🐌 Медленнее (сек)	⚠️ Частичная	Handlers + БД, API + сервисы
E2E	Полный сценарий пользователя	🐢 Очень медленно (мин)	❌ Нет	Критичные пользовательские сценарии

💡 Правило: Пишите больше unit-тестов (70%), меньше интеграционных (20%), минимум E2E (10%).

#pytest — фреймворк для тестирования

#Установка

pip install pytest pytest-asyncio pytest-cov pytest-mock

Пакет	Зачем
`pytest`	Фреймворк для тестирования
`pytest-asyncio`	Поддержка async тестов
`pytest-cov`	Покрытие кода (coverage)
`pytest-mock`	Моки (unittest.mock)

#Конфигурация pytest

# pytest.ini
[pytest]
asyncio_mode = auto  # 💡 Зачем: Автоматическое определение async тестов
testpaths = tests
python_files = test_*.py
python_functions = test_*
addopts = -v --cov=bot --cov-report=term-missing

# tests/conftest.py
"""
Фикстуры для тестов.

💡 Зачем conftest.py:
- Общие фикстуры для всех тестов
- Автоматическая загрузка pytest
- Изоляция тестов
"""
import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from bot.dispatcher import create_dispatcher
from bot.config import settings


@pytest.fixture
def mock_bot():
    """
    Мокированный бот.
    
    💡 Зачем:
    - Не отправляет реальные сообщения
    - Можно проверить вызовы методов
    - Быстрее реального бота
    """
    bot = AsyncMock()
    bot.token = settings.bot_token.get_secret_value()
    bot.id = 123456789
    return bot


@pytest.fixture
def mock_redis():
    """Мокированный Redis."""
    redis = AsyncMock()
    redis.get = AsyncMock(return_value=None)
    redis.set = AsyncMock(return_value=True)
    redis.delete = AsyncMock(return_value=True)
    return redis


@pytest.fixture
async def db_session():
    """
    Тестовая сессия БД.
    
    💡 Зачем:
    - SQLite in-memory — быстро, не нужен PostgreSQL
    - Изоляция — каждый тест в чистой БД
    - Автоматическая очистка после теста
    """
    engine = create_async_engine(
        'sqlite+aiosqlite:///:memory:',
        echo=False
    )

    # Создаём таблицы
    from bot.database import Base
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    async with session_factory() as session:
        yield session

    await engine.dispose()


@pytest.fixture
def mock_user_repo(db_session):
    """Репозиторий с тестовой БД."""
    from bot.repositories.user import UserRepository
    return UserRepository(db_session)


@pytest.fixture
async def dispatcher(mock_bot, mock_redis):
    """Dispatcher для тестов."""
    dp = create_dispatcher(mock_bot, mock_redis)
    return dp

⚠️ Антипаттерн: Не используйте реальную БД и Redis в unit-тестах. Это медленно и создаёт зависимости между тестами.

#Unit-тесты

#Тестирование фильтров

# tests/test_filters.py
import pytest
from aiogram.types import Message, User
from aiogram import F

from bot.filters.is_admin import admin_filter
from bot.config import settings


@pytest.fixture
def admin_user():
    """Пользователь с правами админа."""
    return User(id=settings.admin_ids[0], is_bot=False, first_name='Admin')


@pytest.fixture
def regular_user():
    """Обычный пользователь."""
    return User(id=999999, is_bot=False, first_name='User')


@pytest.fixture
def make_message():
    """
    Фабрика сообщений.
    
    💡 Зачем: Упрощение создания тестовых данных.
    """
    def _make_message(user: User, text: str = '') -> Message:
        return Message(
            message_id=1,
            date=1234567890,
            chat={'id': 1, 'type': 'private'},
            from_user=user,
            text=text
        )
    return _make_message


async def test_admin_filter_passes_for_admin(make_message, admin_user):
    """
    Тест: фильтр пропускает админа.
    
    💡 Зачем: Проверка базовой функциональности.
    """
    message = make_message(admin_user)
    assert await admin_filter(message) is True


async def test_admin_filter_blocks_regular_user(make_message, regular_user):
    """
    Тест: фильтр блокирует обычного пользователя.
    
    💡 Зачем: Проверка безопасности.
    """
    message = make_message(regular_user)
    assert await admin_filter(message) is False


async def test_magic_filter_text_match():
    """Тест Magic Filter."""
    message = Message(
        message_id=1,
        date=1234567890,
        chat={'id': 1, 'type': 'private'},
        from_user=User(id=1, is_bot=False, first_name='Test'),
        text='Привет'
    )

    # Проверка фильтра
    assert F.text == 'Привет'
    assert F.text != 'Пока'
    assert F.text.startswith('Пр')
    assert F.text.endswith('ет')

💡 Совет: Называйте тесты по принципу test_<функция>_<сценарий>_<ожидаемый_результат>.

#Тестирование handlers

# tests/test_handlers.py
import pytest
from aiogram.types import Message, User, Chat
from aiogram import F
from unittest.mock import AsyncMock

from bot.handlers.start import cmd_start


@pytest.fixture
def mock_message():
    """
    Мок сообщения.
    
    💡 Зачем:
    - Изоляция от Telegram API
    - Проверка вызовов answer()
    """
    message = AsyncMock(spec=Message)
    message.message_id = 1
    message.date = 1234567890
    message.from_user = User(id=123, is_bot=False, first_name='Test')
    message.chat = Chat(id=123, type='private')
    message.text = '/start'
    message.answer = AsyncMock()
    return message


async def test_cmd_start(mock_message):
    """
    Тест обработчика /start.
    
    💡 Зачем: Проверка базовой логики хендлера.
    """
    await cmd_start(mock_message)

    # Проверяем, что answer был вызван
    mock_message.answer.assert_called_once()

    # Проверяем текст ответа
    call_args = mock_message.answer.call_args
    assert 'Привет' in call_args[0][0]


async def test_cmd_start_with_user_data(mock_message, mock_user_repo):
    """
    Тест /start с существующим пользователем.
    
    💡 Зачем: Проверка интеграции с БД.
    """
    # Создаём пользователя в тестовой БД
    await mock_user_repo.create(
        telegram_id=mock_message.from_user.id,
        username='testuser'
    )

    await cmd_start(mock_message, user_repo=mock_user_repo)

    mock_message.answer.assert_called_once()

⚠️ Антипаттерн: Не тестируйте внутренние детали реализации (сколько раз вызван метод). Тестируйте результат (ответ пользователю).

#Тестирование сервисов

# tests/test_services.py
import pytest
from unittest.mock import AsyncMock, patch

from bot.services.order import OrderService
from bot.models.order import Order, OrderStatus


@pytest.fixture
def mock_order_repo():
    """
    Мокированный репозиторий заказов.
    
    💡 Зачем: Изоляция от реальной БД.
    """
    repo = AsyncMock()
    repo.create = AsyncMock()
    repo.get_by_id = AsyncMock()
    repo.update = AsyncMock()
    return repo


@pytest.fixture
def mock_notification_service():
    """Мокированный сервис уведомлений."""
    service = AsyncMock()
    service.send_order_created = AsyncMock()
    service.send_order_cancelled = AsyncMock()
    return service


@pytest.fixture
def order_service(mock_order_repo, mock_notification_service):
    """Сервис с мокированными зависимостями."""
    return OrderService(
        order_repo=mock_order_repo,
        notification_service=mock_notification_service
    )


async def test_create_order(order_service, mock_order_repo):
    """
    Тест создания заказа.
    
    💡 Зачем: Проверка бизнес-логики.
    """
    # Настраиваем mock
    mock_order_repo.create.return_value = Order(
        id=1,
        user_id=123,
        total=1000,
        status=OrderStatus.PENDING
    )

    # Вызываем сервис
    order = await order_service.create_order(
        user_id=123,
        items=[{'id': 1, 'quantity': 2, 'price': 500}]
    )

    # Проверяем результат
    assert order.id == 1
    assert order.total == 1000
    mock_order_repo.create.assert_called_once()

    # Проверяем уведомление
    order_service.notification_service.send_order_created.assert_called_once()


async def test_cancel_order_success(order_service, mock_order_repo):
    """Тест успешной отмены заказа."""
    mock_order_repo.get_by_id.return_value = Order(
        id=1,
        user_id=123,
        status=OrderStatus.PENDING
    )

    result = await order_service.cancel_order(order_id=1, reason='User request')

    assert result is True
    mock_order_repo.update.assert_called_once()


async def test_cancel_order_invalid_status(order_service, mock_order_repo):
    """
    Тест отмены заказа в неверном статусе.
    
    💡 Зачем: Проверка обработки ошибок.
    """
    mock_order_repo.get_by_id.return_value = Order(
        id=1,
        user_id=123,
        status=OrderStatus.SHIPPED  # Уже отправлен
    )

    result = await order_service.cancel_order(order_id=1, reason='User request')

    assert result is False
    mock_order_repo.update.assert_not_called()

💡 Совет: Тестируйте сервисы с мокированными зависимостями — это быстро и изолированно.

#Тестирование callback factories

# tests/test_callback_data.py
import pytest

from bot.keyboards.inline import ProductCallback, PaginationCallback


def test_product_callback_pack():
    """
    Тест генерации callback_data.
    
    💡 Зачем: Проверка сериализации.
    """
    callback = ProductCallback(id=123, action='view')
    data = callback.pack()

    assert data == 'product:123:view'


def test_product_callback_unpack():
    """
    Тест парсинга callback_data.
    
    💡 Зачем: Проверка десериализации.
    """
    data = 'product:456:edit'
    callback = ProductCallback.unpack(data)

    assert callback.id == 456
    assert callback.action == 'edit'


def test_pagination_callback():
    """Тест пагинации."""
    callback = PaginationCallback(entity='products', page=3, per_page=10)
    data = callback.pack()

    unpacked = PaginationCallback.unpack(data)
    assert unpacked.page == 3
    assert unpacked.per_page == 10


def test_callback_with_optional_field():
    """
    Тест с необязательным полем.
    
    💡 Зачем: Проверка edge cases.
    """
    # С полем
    callback = ProductCallback(id=1, action='list', page=5)
    assert callback.pack().endswith(':5')

    # Без поля
    callback = ProductCallback(id=1, action='list')
    assert callback.pack().endswith(':')

#Интеграционные тесты

#Тестирование с тестовой БД

# tests/integration/test_user_repository.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from bot.repositories.user import UserRepository
from bot.models.user import User
from bot.database import Base


@pytest.fixture
async def session():
    """
    Создаёт тестовую БД в памяти.
    
    💡 Зачем:
    - Быстрее PostgreSQL
    - Изоляция тестов
    - Автоматическая очистка
    """
    engine = create_async_engine('sqlite+aiosqlite:///:memory:')

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    async with session_factory() as session:
        yield session

    await engine.dispose()


@pytest.fixture
def user_repo(session):
    return UserRepository(session)


async def test_get_or_create_new_user(user_repo):
    """
    Тест получения или создания нового пользователя.
    
    💡 Зачем: Проверка частого сценария.
    """
    user = await user_repo.get_or_create(
        telegram_id=12345,
        username='testuser'
    )

    assert user.telegram_id == 12345
    assert user.username == 'testuser'


async def test_get_or_create_existing_user(user_repo):
    """
    Тест получения существующего пользователя.
    
    💡 Зачем: Проверка, что данные не перезаписываются.
    """
    # Создаём пользователя
    created = await user_repo.create(
        telegram_id=12345,
        username='original'
    )

    # Получаем или создаём (должен получить существующего)
    user = await user_repo.get_or_create(
        telegram_id=12345,
        username='new_username'  # Это игнорируется
    )

    assert user.id == created.id
    assert user.username == 'original'


async def test_get_by_telegram_id(user_repo):
    """Тест поиска по Telegram ID."""
    await user_repo.create(telegram_id=999, username='findme')

    user = await user_repo.get_by_telegram_id(999)

    assert user is not None
    assert user.username == 'findme'

    # Несуществующий
    user = await user_repo.get_by_telegram_id(0)
    assert user is None

#Тестирование FSM

# tests/integration/test_fsm.py
import pytest
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
from aiogram.types import User, Chat


class Registration(StatesGroup):
    name = State()
    email = State()


@pytest.fixture
def fsm_context():
    """
    FSM контекст для тестов.
    
    💡 Зачем: MemoryStorage для быстрых тестов.
    """
    storage = MemoryStorage()
    user = User(id=123, is_bot=False, first_name='Test')
    chat = Chat(id=123, type='private')
    return FSMContext(storage=storage, chat=chat, user=user)


async def test_fsm_state_transitions(fsm_context):
    """
    Тест переходов между состояниями.
    
    💡 Зачем: Проверка FSM логики.
    """
    # Начальное состояние
    state = await fsm_context.get_state()
    assert state is None

    # Переход к name
    await fsm_context.set_state(Registration.name)
    state = await fsm_context.get_state()
    assert state == 'Registration:name'

    # Сохранение данных
    await fsm_context.update_data(name='Ivan')
    data = await fsm_context.get_data()
    assert data == {'name': 'Ivan'}

    # Переход к email
    await fsm_context.set_state(Registration.email)
    state = await fsm_context.get_state()
    assert state == 'Registration:email'

    # Данные сохраняются
    data = await fsm_context.get_data()
    assert data['name'] == 'Ivan'

    # Очистка
    await fsm_context.clear()
    state = await fsm_context.get_state()
    assert state is None

#Тестирование с реальным Redis (опционально)

# tests/integration/test_redis_storage.py
import pytest
import redis.asyncio as redis
from aiogram.fsm.storage.redis import RedisStorage
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext


class TestState(StatesGroup):
    step1 = State()


@pytest.fixture
async def redis_client():
    """
    Redis клиент для тестов.
    
    ⚠️ Важно: Отдельная БД Redis (db=15) для тестов.
    """
    client = redis.Redis(host='localhost', port=6379, db=15)
    yield client
    await client.flushdb()  # Очищаем после теста
    await client.close()


@pytest.fixture
def redis_storage(redis_client):
    return RedisStorage(redis=redis_client)


async def test_redis_storage(redis_storage):
    """Тест хранения в Redis."""
    user = User(id=123, is_bot=False)
    chat = Chat(id=123, type='private')

    state = FSMContext(storage=redis_storage, chat=chat, user=user)

    # Устанавливаем состояние
    await state.set_state(TestState.step1)
    await state.update_data(value='test')

    # Проверяем
    stored_state = await state.get_state()
    assert stored_state == 'TestState:step1'

    stored_data = await state.get_data()
    assert stored_data == {'value': 'test'}

⚠️ Антипаттерн: Не используйте Redis в unit-тестах. Только в интеграционных, когда тестируете именно хранение в Redis.

#Моки и фикстуры

#Мокирование внешних API

# tests/test_external_api.py
import pytest
from unittest.mock import AsyncMock, patch

from bot.services.api_client import APIClient


async def test_api_client_with_mock():
    """
    Тест API клиента с моком.
    
    💡 Зачем:
    - Не делает реальные HTTP-запросы
    - Быстро
    - Предсказуемый результат
    """
    mock_response = {'status': 'success', 'data': {'id': 1}}

    with patch('aiohttp.ClientSession.post') as mock_post:
        mock_post.return_value.__aenter__ = AsyncMock()
        mock_post.return_value.__aenter__.return_value.json = AsyncMock(return_value=mock_response)

        client = APIClient(api_key='test')
        result = await client.fetch_data('https://api.example.com/data')

        assert result == mock_response
        mock_post.assert_called_once()


async def test_api_client_error_handling():
    """
    Тест обработки ошибок API.
    
    💡 Зачем: Проверка устойчивости к сбоям.
    """
    from aiohttp import ClientError

    with patch('aiohttp.ClientSession.post') as mock_post:
        mock_post.return_value.__aenter__ = AsyncMock(side_effect=ClientError())

        client = APIClient(api_key='test')

        with pytest.raises(ClientError):
            await client.fetch_data('https://api.example.com/data')

#Мокирование бота

# tests/conftest.py
@pytest.fixture
def bot_mock():
    """
    Полный мок бота.
    
    💡 Зачем:
    - Тестирование без реального бота
    - Проверка вызовов методов
    """
    bot = AsyncMock()
    bot.token = '123:ABC'
    bot.id = 123

    # Мокируем методы
    bot.send_message = AsyncMock()
    bot.edit_message_text = AsyncMock()
    bot.answer_callback_query = AsyncMock()
    bot.send_photo = AsyncMock()

    return bot


# tests/test_notifications.py
async def test_send_notification(bot_mock, notification_service):
    """Тест отправки уведомления."""
    await notification_service.send_message(
        bot=bot_mock,
        user_id=123,
        text='Hello'
    )

    bot_mock.send_message.assert_called_once_with(
        chat_id=123,
        text='Hello'
    )

#Запуск тестов

# Все тесты
pytest

# С покрытием
pytest --cov=bot --cov-report=html
# 💡 Зачем: Показывает, какой код не покрыт тестами

# Только unit-тесты
pytest tests/unit/

# Только интеграционные
pytest tests/integration/

# Один тест
pytest tests/test_handlers.py::test_cmd_start -v

# С выводом логов
pytest -s --log-cli-level=INFO

# Параллельный запуск (ускоряет тесты)
pytest -n auto

# Тесты с флагом stop on first failure
pytest -x

💡 Совет: Настройте CI/CD для автоматического запуска тестов при каждом коммите.

#Отладка

#Логирование в тестах

import logging
import pytest

logger = logging.getLogger(__name__)


@pytest.fixture(autouse=True)
def enable_logging():
    """Включает логирование в тестах."""
    logging.basicConfig(level=logging.DEBUG)


async def test_with_logging():
    logger.debug('Debug message')
    logger.info('Info message')
    ...

#Отладка с pdb

async def test_debug():
    import pdb; pdb.set_trace()  # Точка останова

    result = await some_function()
    assert result == expected

#Отладка локального бота

# bot/debug_bot.py
import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.fsm.storage.memory import MemoryStorage

logging.basicConfig(level=logging.DEBUG)


async def main():
    """
    Отладочный бот.
    
    💡 Зачем:
    - Быстрый запуск без Docker
    - Debug-логирование
    - MemoryStorage для простоты
    """
    bot = Bot(token='YOUR_TEST_BOT_TOKEN')

    # MemoryStorage для отладки
    dp = Dispatcher(storage=MemoryStorage())

    # Регистрируем handlers
    from bot.handlers import start, profile
    dp.include_router(start.router)
    dp.include_router(profile.router)

    print('Bot started. Press Ctrl+C to stop.')
    await dp.start_polling(bot)


if __name__ == '__main__':
    asyncio.run(main())

⚠️ Важно: Не используйте debug_bot в продакшене. Только локально.

#Итоги

#✅ Чек-лист: покрыто ли ваше решение тестами?

Unit-тесты на фильтры
Unit-тесты на сервисы (с моками)
Unit-тесты на callback factories
Integration-тесты на handlers + БД
Integration-тесты на FSM сценарии
Тесты на обработку ошибок
Покрытие кода >80%
Тесты запускаются в CI/CD

#🔗 Связь с другими темами

Тема	Как связана
Архитектура	Модульная структура упрощает тестирование
Базы данных	SQLite in-memory для быстрых тестов
Мониторинг	Тесты в CI/CD как часть pipeline
Безопасность	Тесты на уязвимости

#📚 Что дальше?

→ CI/CD — автоматизация запуска тестов
→ Мониторинг — тесты производительности
→ Безопасность — security-тестирование

Тип

Что тестирует

Скорость

Изоляция

Когда использовать

Unit

Отдельные функции, классы

⚡ Быстро (мс)

✅ Полная

Логика сервисов, фильтры, валидация

Integration

Взаимодействие компонентов

🐌 Медленнее (сек)

⚠️ Частичная

Handlers + БД, API + сервисы

E2E

Полный сценарий пользователя

🐢 Очень медленно (мин)

❌ Нет

Критичные пользовательские сценарии

Пакет

Зачем

pytest

Фреймворк для тестирования

pytest-asyncio

Поддержка async тестов

pytest-cov

Покрытие кода (coverage)

pytest-mock

Моки (unittest.mock)

Тема

Как связана

Архитектура

Модульная структура упрощает тестирование

Базы данных

SQLite in-memory для быстрых тестов

Мониторинг

Тесты в CI/CD как часть pipeline

Безопасность

Тесты на уязвимости