pytest, моки, интеграционные тесты, дебаггинг, локальное тестирование
Проблема: Вы изменили код бота, чтобы добавить новую фичу. Через час пользователи пишут: "Перестала работать оплата!", "Регистрация сломалась!". Как этого избежать?
Реальность без тестов:
- Баги обнаруживаются пользователями в продакшене
- Страх менять код — "а вдруг что-то сломаю?"
- Ручное тестирование занимает часы
- Невозможно откатить — неясно, что работало раньше
Решение: Автоматические тесты ловят баги до продакшена. Unit-тесты проверяют отдельные функции, интеграционные — взаимодействие компонентов, E2E — полные сценарии пользователя.
Зачем это нужно: Тесты — это страховка. Они дают уверенность менять код, ускоряют разработку (не нужно вручную проверять всё каждый раз), документируют ожидаемое поведение.
| Тип | Что тестирует | Скорость | Изоляция | Когда использовать |
|---|---|---|---|---|
| Unit | Отдельные функции, классы | ⚡ Быстро (мс) | ✅ Полная | Логика сервисов, фильтры, валидация |
| Integration | Взаимодействие компонентов | 🐌 Медленнее (сек) | ⚠️ Частичная | Handlers + БД, API + сервисы |
| E2E | Полный сценарий пользователя | 🐢 Очень медленно (мин) | ❌ Нет | Критичные пользовательские сценарии |
💡 Правило: Пишите больше unit-тестов (70%), меньше интеграционных (20%), минимум E2E (10%).
pip install pytest pytest-asyncio pytest-cov pytest-mock| Пакет | Зачем |
|---|---|
pytest | Фреймворк для тестирования |
pytest-asyncio | Поддержка async тестов |
pytest-cov | Покрытие кода (coverage) |
pytest-mock | Моки (unittest.mock) |
# 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-тестах. Это медленно и создаёт зависимости между тестами.
# 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_<функция>_<сценарий>_<ожидаемый_результат>.
# 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()💡 Совет: Тестируйте сервисы с мокированными зависимостями — это быстро и изолированно.
# 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# 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# 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.
# 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')
...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 в продакшене. Только локально.
| Тема | Как связана |
|---|---|
| Архитектура | Модульная структура упрощает тестирование |
| Базы данных | SQLite in-memory для быстрых тестов |
| Мониторинг | Тесты в CI/CD как часть pipeline |
| Безопасность | Тесты на уязвимости |
→ CI/CD — автоматизация запуска тестов
→ Мониторинг — тесты производительности
→ Безопасность — security-тестирование
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.