Тестирование ботов: unit и integration тесты
В этой статье мы рассмотрим различные подходы к тестированию ботов, включая unit-тесты, интеграционные тесты, мокирование внешних API и автоматизацию тестирования.Содержание
- Основы тестирования ботов
- Unit-тесты
- Интеграционные тесты
- Мокирование внешних сервисов
- Тестирование производительности
- Автоматизация тестирования
- CI/CD для ботов
Основы тестирования ботов
Структура тестов
# Структура тестов для ботов
import pytest
import asyncio
from unittest.mock import Mock, patch, AsyncMock
from typing import Dict, Any
class TestBotStructure:
"""Базовая структура тестов для ботов"""
@pytest.fixture
def bot_instance(self):
"""Фикстура для создания экземпляра бота"""
from my_bot import TelegramBot
return TelegramBot(token="test_token")
@pytest.fixture
def mock_update(self):
"""Фикстура для создания мока Update"""
update = Mock()
update.message = Mock()
update.message.from_user = Mock()
update.message.from_user.id = 12345
update.message.text = "/start"
update.message.reply_text = AsyncMock()
return update
@pytest.fixture
def mock_context(self):
"""Фикстура для создания мока Context"""
context = Mock()
context.bot = Mock()
context.bot.send_message = AsyncMock()
return context
# Пример базового теста
def test_bot_initialization(bot_instance):
"""Тест инициализации бота"""
assert bot_instance is not None
assert bot_instance.token == "test_token"
@pytest.mark.asyncio
async def test_start_command(bot_instance, mock_update, mock_context):
"""Тест команды /start"""
await bot_instance.start_command(mock_update, mock_context)
# Проверка, что был вызван reply_text
mock_update.message.reply_text.assert_called_once()
# Проверка содержимого ответа
call_args = mock_update.message.reply_text.call_args[0]
assert "привет" in call_args[0].lower()Конфигурация pytest
# pytest.ini
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
-v
--tb=short
--strict-markers
--disable-warnings
markers =
unit: Unit tests
integration: Integration tests
slow: Slow tests
api: API tests
database: Database tests
# conftest.py
import pytest
import asyncio
from unittest.mock import Mock
import os
@pytest.fixture(scope="session")
def event_loop():
"""Создание event loop для асинхронных тестов"""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
def test_config():
"""Конфигурация для тестов"""
return {
'bot_token': 'test_token',
'database_url': 'sqlite:///test.db',
'api_base_url': 'https://api.test.com',
'debug': True
}
@pytest.fixture
def mock_telegram_api():
"""Мок Telegram API"""
with patch('telegram.Bot') as mock_bot:
mock_instance = Mock()
mock_instance.get_me = AsyncMock(return_value={'id': 12345, 'username': 'test_bot'})
mock_instance.send_message = AsyncMock()
mock_instance.get_updates = AsyncMock(return_value=[])
mock_bot.return_value = mock_instance
yield mock_instance
@pytest.fixture
def mock_database():
"""Мок базы данных"""
with patch('sqlalchemy.create_engine') as mock_engine:
mock_instance = Mock()
mock_engine.return_value = mock_instance
yield mock_instanceUnit-тесты
Тестирование обработчиков команд
# tests/test_commands.py
import pytest
from unittest.mock import Mock, AsyncMock, patch
from my_bot.commands import CommandHandler
class TestCommandHandler:
"""Тесты обработчика команд"""
@pytest.fixture
def command_handler(self):
"""Фикстура для обработчика команд"""
return CommandHandler()
@pytest.fixture
def mock_update(self):
"""Мок Update объекта"""
update = Mock()
update.message = Mock()
update.message.from_user = Mock()
update.message.from_user.id = 12345
update.message.from_user.username = "testuser"
update.message.text = "/help"
update.message.reply_text = AsyncMock()
return update
@pytest.fixture
def mock_context(self):
"""Мок Context объекта"""
context = Mock()
context.bot = Mock()
context.bot.send_message = AsyncMock()
return context
@pytest.mark.asyncio
async def test_help_command(self, command_handler, mock_update, mock_context):
"""Тест команды /help"""
await command_handler.help_command(mock_update, mock_context)
# Проверка вызова reply_text
mock_update.message.reply_text.assert_called_once()
# Проверка содержимого ответа
call_args = mock_update.message.reply_text.call_args[0]
response_text = call_args[0]
assert "помощь" in response_text.lower()
assert "команды" in response_text.lower()
@pytest.mark.asyncio
async def test_start_command(self, command_handler, mock_update, mock_context):
"""Тест команды /start"""
await command_handler.start_command(mock_update, mock_context)
mock_update.message.reply_text.assert_called_once()
call_args = mock_update.message.reply_text.call_args[0]
response_text = call_args[0]
assert "добро пожаловать" in response_text.lower()
@pytest.mark.asyncio
async def test_unknown_command(self, command_handler, mock_update, mock_context):
"""Тест неизвестной команды"""
mock_update.message.text = "/unknown"
await command_handler.handle_command(mock_update, mock_context)
mock_update.message.reply_text.assert_called_once()
call_args = mock_update.message.reply_text.call_args[0]
response_text = call_args[0]
assert "неизвестная команда" in response_text.lower()Тестирование бизнес-логики
# tests/test_business_logic.py
import pytest
from unittest.mock import Mock, patch
from my_bot.services import UserService, PaymentService
class TestUserService:
"""Тесты сервиса пользователей"""
@pytest.fixture
def user_service(self):
"""Фикстура для сервиса пользователей"""
return UserService()
@pytest.fixture
def mock_database(self):
"""Мок базы данных"""
with patch('my_bot.services.get_database') as mock_db:
mock_instance = Mock()
mock_db.return_value = mock_instance
yield mock_instance
def test_create_user(self, user_service, mock_database):
"""Тест создания пользователя"""
user_data = {
'telegram_id': 12345,
'username': 'testuser',
'first_name': 'Test',
'last_name': 'User'
}
# Настройка мока
mock_database.execute.return_value.rowcount = 1
result = user_service.create_user(user_data)
assert result is True
mock_database.execute.assert_called_once()
def test_get_user_by_telegram_id(self, user_service, mock_database):
"""Тест получения пользователя по Telegram ID"""
telegram_id = 12345
expected_user = {
'id': 1,
'telegram_id': telegram_id,
'username': 'testuser'
}
# Настройка мока
mock_result = Mock()
mock_result.fetchone.return_value = expected_user
mock_database.execute.return_value = mock_result
result = user_service.get_user_by_telegram_id(telegram_id)
assert result == expected_user
mock_database.execute.assert_called_once()
def test_update_user_subscription(self, user_service, mock_database):
"""Тест обновления подписки пользователя"""
user_id = 1
subscription_type = "premium"
mock_database.execute.return_value.rowcount = 1
result = user_service.update_subscription(user_id, subscription_type)
assert result is True
mock_database.execute.assert_called_once()
class TestPaymentService:
"""Тесты сервиса платежей"""
@pytest.fixture
def payment_service(self):
"""Фикстура для сервиса платежей"""
return PaymentService()
@pytest.fixture
def mock_payment_gateway(self):
"""Мок платежного шлюза"""
with patch('my_bot.services.PaymentGateway') as mock_gateway:
mock_instance = Mock()
mock_gateway.return_value = mock_instance
yield mock_instance
def test_process_payment_success(self, payment_service, mock_payment_gateway):
"""Тест успешной обработки платежа"""
payment_data = {
'amount': 1000,
'currency': 'RUB',
'user_id': 1
}
# Настройка мока для успешного платежа
mock_payment_gateway.process_payment.return_value = {
'status': 'success',
'transaction_id': 'txn_123',
'amount': 1000
}
result = payment_service.process_payment(payment_data)
assert result['status'] == 'success'
assert 'transaction_id' in result
mock_payment_gateway.process_payment.assert_called_once_with(payment_data)
def test_process_payment_failure(self, payment_service, mock_payment_gateway):
"""Тест неудачной обработки платежа"""
payment_data = {
'amount': 1000,
'currency': 'RUB',
'user_id': 1
}
# Настройка мока для неудачного платежа
mock_payment_gateway.process_payment.return_value = {
'status': 'failed',
'error': 'Insufficient funds'
}
result = payment_service.process_payment(payment_data)
assert result['status'] == 'failed'
assert 'error' in resultИнтеграционные тесты
Тестирование с реальными API
# tests/test_integration.py
import pytest
import asyncio
from unittest.mock import patch
import aiohttp
from my_bot.integrations import ExternalAPIClient
class TestExternalAPIIntegration:
"""Интеграционные тесты с внешними API"""
@pytest.fixture
def api_client(self):
"""Фикстура для API клиента"""
return ExternalAPIClient(base_url="https://api.test.com")
@pytest.mark.asyncio
async def test_api_connection(self, api_client):
"""Тест подключения к API"""
async with aiohttp.ClientSession() as session:
try:
response = await session.get("https://httpbin.org/get")
assert response.status == 200
except Exception as e:
pytest.skip(f"API недоступен: {e}")
@pytest.mark.asyncio
async def test_api_request_with_mock(self, api_client):
"""Тест API запроса с моком"""
with patch('aiohttp.ClientSession.get') as mock_get:
# Настройка мока
mock_response = Mock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={'status': 'ok'})
mock_get.return_value.__aenter__.return_value = mock_response
result = await api_client.make_request('GET', '/test')
assert result['status'] == 'ok'
mock_get.assert_called_once()
@pytest.mark.asyncio
async def test_api_error_handling(self, api_client):
"""Тест обработки ошибок API"""
with patch('aiohttp.ClientSession.get') as mock_get:
# Настройка мока для ошибки
mock_response = Mock()
mock_response.status = 500
mock_response.text = AsyncMock(return_value='Internal Server Error')
mock_get.return_value.__aenter__.return_value = mock_response
with pytest.raises(Exception):
await api_client.make_request('GET', '/test')
class TestDatabaseIntegration:
"""Интеграционные тесты с базой данных"""
@pytest.fixture
def test_database(self):
"""Фикстура для тестовой базы данных"""
import sqlite3
import tempfile
import os
# Создание временной базы данных
db_fd, db_path = tempfile.mkstemp()
conn = sqlite3.connect(db_path)
# Создание тестовых таблиц
conn.execute('''
CREATE TABLE users (
id INTEGER PRIMARY KEY,
telegram_id INTEGER UNIQUE,
username TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
yield db_path
# Очистка
os.close(db_fd)
os.unlink(db_path)
def test_database_operations(self, test_database):
"""Тест операций с базой данных"""
import sqlite3
conn = sqlite3.connect(test_database)
# Тест вставки
conn.execute(
'INSERT INTO users (telegram_id, username) VALUES (?, ?)',
(12345, 'testuser')
)
conn.commit()
# Тест выборки
cursor = conn.execute(
'SELECT * FROM users WHERE telegram_id = ?',
(12345,)
)
user = cursor.fetchone()
assert user is not None
assert user[1] == 12345 # telegram_id
assert user[2] == 'testuser' # username
conn.close()Мокирование внешних сервисов
Мокирование Telegram API
# tests/test_telegram_mocks.py
import pytest
from unittest.mock import Mock, AsyncMock, patch
from telegram import Update, Message, User, Chat
class TestTelegramMocks:
"""Тесты с моками Telegram API"""
@pytest.fixture
def mock_user(self):
"""Мок пользователя Telegram"""
user = Mock(spec=User)
user.id = 12345
user.username = "testuser"
user.first_name = "Test"
user.last_name = "User"
user.is_bot = False
return user
@pytest.fixture
def mock_chat(self):
"""Мок чата Telegram"""
chat = Mock(spec=Chat)
chat.id = 12345
chat.type = "private"
chat.title = None
return chat
@pytest.fixture
def mock_message(self, mock_user, mock_chat):
"""Мок сообщения Telegram"""
message = Mock(spec=Message)
message.message_id = 1
message.from_user = mock_user
message.chat = mock_chat
message.text = "/start"
message.reply_text = AsyncMock()
message.reply_photo = AsyncMock()
message.reply_document = AsyncMock()
return message
@pytest.fixture
def mock_update(self, mock_message):
"""Мок Update объекта"""
update = Mock(spec=Update)
update.update_id = 1
update.message = mock_message
update.callback_query = None
return update
@pytest.fixture
def mock_context(self):
"""Мок Context объекта"""
context = Mock()
context.bot = Mock()
context.bot.send_message = AsyncMock()
context.bot.send_photo = AsyncMock()
context.bot.send_document = AsyncMock()
context.bot.get_file = AsyncMock()
return context
@pytest.mark.asyncio
async def test_start_command_with_mocks(self, mock_update, mock_context):
"""Тест команды /start с моками"""
from my_bot.commands import start_command
await start_command(mock_update, mock_context)
# Проверка вызова reply_text
mock_update.message.reply_text.assert_called_once()
# Проверка содержимого ответа
call_args = mock_update.message.reply_text.call_args[0]
response_text = call_args[0]
assert "добро пожаловать" in response_text.lower()
@pytest.mark.asyncio
async def test_photo_command_with_mocks(self, mock_update, mock_context):
"""Тест команды отправки фото с моками"""
from my_bot.commands import photo_command
# Настройка мока для фото
mock_file = Mock()
mock_file.file_path = "https://example.com/photo.jpg"
mock_context.bot.get_file.return_value = mock_file
await photo_command(mock_update, mock_context)
# Проверка вызова reply_photo
mock_update.message.reply_photo.assert_called_once()
# Проверка аргументов
call_args = mock_update.message.reply_photo.call_args[0]
assert call_args[0] == "https://example.com/photo.jpg"Мокирование внешних API
# tests/test_external_api_mocks.py
import pytest
from unittest.mock import Mock, AsyncMock, patch
import aiohttp
from my_bot.services import WeatherService, NewsService
class TestExternalAPIMocks:
"""Тесты с моками внешних API"""
@pytest.fixture
def mock_http_session(self):
"""Мок HTTP сессии"""
with patch('aiohttp.ClientSession') as mock_session:
mock_instance = Mock()
mock_session.return_value.__aenter__.return_value = mock_instance
yield mock_instance
@pytest.mark.asyncio
async def test_weather_service_success(self, mock_http_session):
"""Тест успешного получения погоды"""
weather_service = WeatherService()
# Настройка мока для успешного ответа
mock_response = Mock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={
'main': {'temp': 20, 'humidity': 60},
'weather': [{'description': 'clear sky'}]
})
mock_http_session.get.return_value.__aenter__.return_value = mock_response
result = await weather_service.get_weather("Moscow")
assert result['temperature'] == 20
assert result['humidity'] == 60
assert result['description'] == 'clear sky'
@pytest.mark.asyncio
async def test_weather_service_error(self, mock_http_session):
"""Тест ошибки получения погоды"""
weather_service = WeatherService()
# Настройка мока для ошибки
mock_response = Mock()
mock_response.status = 404
mock_response.text = AsyncMock(return_value='City not found')
mock_http_session.get.return_value.__aenter__.return_value = mock_response
with pytest.raises(Exception):
await weather_service.get_weather("UnknownCity")
@pytest.mark.asyncio
async def test_news_service_with_mock(self, mock_http_session):
"""Тест сервиса новостей с моком"""
news_service = NewsService()
# Настройка мока для новостей
mock_response = Mock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={
'articles': [
{
'title': 'Test News',
'description': 'Test description',
'url': 'https://example.com/news1'
},
{
'title': 'Another News',
'description': 'Another description',
'url': 'https://example.com/news2'
}
]
})
mock_http_session.get.return_value.__aenter__.return_value = mock_response
result = await news_service.get_latest_news()
assert len(result) == 2
assert result[0]['title'] == 'Test News'
assert result[1]['title'] == 'Another News'Тестирование производительности
Нагрузочное тестирование
# tests/test_performance.py
import pytest
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from my_bot.services import DatabaseService, APIService
class TestPerformance:
"""Тесты производительности"""
@pytest.fixture
def database_service(self):
"""Фикстура для сервиса базы данных"""
return DatabaseService()
@pytest.fixture
def api_service(self):
"""Фикстура для сервиса API"""
return APIService()
def test_database_query_performance(self, database_service):
"""Тест производительности запросов к БД"""
start_time = time.time()
# Выполнение множественных запросов
for i in range(100):
database_service.get_user_by_id(i)
end_time = time.time()
duration = end_time - start_time
# Проверка, что запросы выполняются достаточно быстро
assert duration < 1.0 # Менее 1 секунды для 100 запросов
print(f"100 запросов к БД выполнены за {duration:.3f} секунд")
@pytest.mark.asyncio
async def test_concurrent_api_requests(self, api_service):
"""Тест конкурентных API запросов"""
start_time = time.time()
# Создание задач для конкурентного выполнения
tasks = []
for i in range(50):
task = api_service.make_request(f"/test/{i}")
tasks.append(task)
# Выполнение всех задач конкурентно
results = await asyncio.gather(*tasks)
end_time = time.time()
duration = end_time - start_time
# Проверка результатов
assert len(results) == 50
assert duration < 5.0 # Менее 5 секунд для 50 запросов
print(f"50 конкурентных API запросов выполнены за {duration:.3f} секунд")
def test_memory_usage(self, database_service):
"""Тест использования памяти"""
import psutil
import gc
# Измерение памяти до операции
process = psutil.Process()
memory_before = process.memory_info().rss / 1024 / 1024 # MB
# Выполнение операции, которая может потреблять много памяти
large_data = []
for i in range(10000):
large_data.append(database_service.get_user_by_id(i))
# Измерение памяти после операции
memory_after = process.memory_info().rss / 1024 / 1024 # MB
# Очистка памяти
del large_data
gc.collect()
memory_after_cleanup = process.memory_info().rss / 1024 / 1024 # MB
print(f"Память до операции: {memory_before:.2f} MB")
print(f"Память после операции: {memory_after:.2f} MB")
print(f"Память после очистки: {memory_after_cleanup:.2f} MB")
# Проверка, что память была освобождена
assert memory_after_cleanup - memory_before < 100 # Увеличение менее чем на 100 MB
@pytest.mark.asyncio
async def test_response_time_consistency(self, api_service):
"""Тест консистентности времени ответа"""
response_times = []
# Выполнение множественных запросов
for i in range(20):
start_time = time.time()
await api_service.make_request("/test")
end_time = time.time()
response_times.append(end_time - start_time)
# Анализ времени ответа
avg_response_time = sum(response_times) / len(response_times)
max_response_time = max(response_times)
min_response_time = min(response_times)
print(f"Среднее время ответа: {avg_response_time:.3f} секунд")
print(f"Максимальное время ответа: {max_response_time:.3f} секунд")
print(f"Минимальное время ответа: {min_response_time:.3f} секунд")
# Проверка консистентности
assert max_response_time - min_response_time < 1.0 # Разброс менее 1 секунды
assert avg_response_time < 2.0 # Среднее время менее 2 секундАвтоматизация тестирования
Настройка GitHub Actions
# .github/workflows/tests.yml
name: Tests
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.8, 3.9, 3.10, 3.11]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run unit tests
run: |
pytest tests/unit/ -v --cov=my_bot --cov-report=xml
- name: Run integration tests
run: |
pytest tests/integration/ -v
env:
BOT_TOKEN: ${{ secrets.TEST_BOT_TOKEN }}
DATABASE_URL: sqlite:///test.db
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
flags: unittests
name: codecov-umbrella
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: 3.11
- name: Install linting tools
run: |
pip install flake8 black isort mypy
- name: Run black
run: black --check .
- name: Run isort
run: isort --check-only .
- name: Run flake8
run: flake8 .
- name: Run mypy
run: mypy my_bot/Настройка pre-commit
# .pre-commit-config.yaml
repos:
- repo: https://github.com/psf/black
rev: 22.3.0
hooks:
- id: black
language_version: python3
- repo: https://github.com/pycqa/isort
rev: 5.10.1
hooks:
- id: isort
- repo: https://github.com/pycqa/flake8
rev: 4.0.1
hooks:
- id: flake8
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v0.950
hooks:
- id: mypy
additional_dependencies: [types-requests]
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.2.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-filesCI/CD для ботов
Автоматическое развертывание
# .github/workflows/deploy.yml
name: Deploy
on:
push:
branches: [ main ]
tags: [ 'v*' ]
jobs:
deploy:
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: 3.11
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Run tests
run: |
pytest tests/ -v
env:
BOT_TOKEN: ${{ secrets.TEST_BOT_TOKEN }}
- name: Build Docker image
run: |
docker build -t my-bot:${{ github.sha }} .
- name: Deploy to production
run: |
# Здесь должна быть логика развертывания
echo "Deploying to production..."
env:
PRODUCTION_TOKEN: ${{ secrets.PRODUCTION_BOT_TOKEN }}
SERVER_HOST: ${{ secrets.SERVER_HOST }}
SERVER_USER: ${{ secrets.SERVER_USER }}Мониторинг тестов
# tests/test_monitoring.py
import pytest
from unittest.mock import patch
from my_bot.monitoring import HealthChecker
class TestMonitoring:
"""Тесты мониторинга"""
@pytest.fixture
def health_checker(self):
"""Фикстура для проверки здоровья"""
return HealthChecker()
def test_database_health_check(self, health_checker):
"""Тест проверки здоровья БД"""
with patch('my_bot.monitoring.get_database') as mock_db:
mock_db.return_value.execute.return_value.fetchone.return_value = (1,)
result = health_checker.check_database()
assert result is True
def test_api_health_check(self, health_checker):
"""Тест проверки здоровья API"""
with patch('requests.get') as mock_get:
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
result = health_checker.check_external_api()
assert result is True
def test_health_check_failure(self, health_checker):
"""Тест неудачной проверки здоровья"""
with patch('my_bot.monitoring.get_database') as mock_db:
mock_db.side_effect = Exception("Database connection failed")
result = health_checker.check_database()
assert result is FalseЗаключение
В этой статье мы рассмотрели комплексный подход к тестированию ботов:- ✅ Структура и организация тестов
- ✅ Unit-тесты для команд и бизнес-логики
- ✅ Интеграционные тесты с реальными сервисами
- ✅ Мокирование внешних API и сервисов
- ✅ Тестирование производительности
- ✅ Автоматизация тестирования с CI/CD
- ✅ Мониторинг и проверка здоровья
Полезные ссылки
798 просмотров
49 лайков
0 комментариев
Комментарии (0)
Пока нет комментариев. Будьте первым!