Приведи пример реализации сложного автотеста

Пример реализации сложного автотеста: Тестирование многошаговой операции с внешним API, параллельными процессами и состоянием системы

В качестве примера сложного автотеста рассмотрим сквозное тестирование процесса оформления заказа в интернет-магазине, который включает: взаимодействие с внешним платежным API, обработку параллельных событий (например, уведомлений), проверку состояния базы данных и генерацию отчетов. Этот тест объединяет несколько слоев системы.

Архитектура теста и ключевые сложности

Сложности, которые необходимо учесть:

1. Взаимодействие с внешним сервисом (платежный шлюз) — требует мокирования/стабинга для независимости тестов.
2. Параллельные процессы — уведомления отправляются асинхронно, их получение нужно валидировать.
3. Состояние данных — проверка корректности записи заказа в БД и обновления库存 (склада).
4. Генерация сложного отчета — PDF-генерация с динамическими данными.
5. Восстановление состояния системы — чистые данные для каждого теста.

Реализация теста на Python с использованием pytest и современных библиотек

import pytest
import requests
from unittest.mock import patch, MagicMock
from contextlib import contextmanager
from datetime import datetime
import json
import asyncio

# Тестируемый модуль (пример структуры)
from order_service import OrderProcessor, PaymentGatewayClient, InventoryService, NotificationService, ReportGenerator

class TestComplexOrderProcessing:
    """
    Сложный автотест для полного цикла оформления заказа.
    Использует моки для внешних сервисов, проверяет асинхронные операции
    и состояние БД.
    """
    
    @pytest.fixture(autouse=True)
    def setup_test_data(self, db_session):
        """Фикстура для подготовки и очистки тестовых данных."""
        # Создание тестового пользователя и товара
        user = db_session.create_user(id=1, email="test@example.com")
        product = db_session.create_product(id=100, name="Тестовый товар", stock=10, price=1500)
        yield
        # Автоматическая очистка после теста
        db_session.rollback()

    @pytest.fixture
    def mock_payment_gateway(self):
        """Мок для платежного шлюза с имитацией успешного и неуспешного ответов."""
        with patch('order_service.PaymentGatewayClient._send_request') as mock:
            # Конфигурация мока для успешного платежа
            successful_response = {
                'transaction_id': 'txn_789',
                'status': 'approved',
                'amount': 1500,
                'timestamp': datetime.now().isoformat()
            }
            mock.return_value = MagicMock(status_code=200, json=lambda: successful_response)
            yield mock

    @contextmanager
    def mock_notification_service(self):
        """Контекстный менеджер для мокирования асинхронной службы уведомлений."""
        with patch('order_service.NotificationService.send') as mock_send:
            with patch('order_service.NotificationService.listen') as mock_listen:
                # Очередь имитируемых полученных уведомлений
                notification_queue = asyncio.Queue()
                mock_send.return_value = {'notification_id': 'notif_123'}
                mock_listen.return_value = notification_queue
                yield notification_queue

    def test_complete_order_flow_with_successful_payment(self, db_session, mock_payment_gateway):
        """Основной тест: успешный поток оформления заказа."""
        
        # 1. Инициализация сервисов
        processor = OrderProcessor()
        inventory = InventoryService(db_session)
        
        # 2. Создание заказа
        order_data = {
            'user_id': 1,
            'product_id': 100,
            'quantity': 2
        }
        order = processor.create_order(order_data)
        
        # Проверка промежуточного состояния: заказ создан, но не оплачен
        assert order['status'] == 'pending'
        assert order['total'] == 3000  # 2 * 1500
        
        # 3. Проверка резервирования товара на складе
        initial_stock = inventory.get_stock(100)
        processor.reserve_items(order['id'])
        updated_stock = inventory.get_stock(100)
        assert updated_stock == initial_stock - 2  #库存 уменьшился
        
        # 4. Обработка платежа через мок
        payment_result = processor.process_payment(order['id'], 'card_token_xyz')
        
        # Валидация ответа платежного шлюза
        assert payment_result['transaction_id'] == 'txn_789'
        assert payment_result['status'] == 'approved'
        
        # 5. Проверка обновления статуса заказа в БД
        db_order = db_session.get_order(order['id'])
        assert db_order.status == 'paid'
        assert db_order.payment_transaction_id == 'txn_789'
        
        # 6. Тестирование асинхронных уведомлений
        with self.mock_notification_service() as notification_queue:
            # Имитация отправки уведомления
            processor.send_order_notifications(order['id'])
            
            # Имитация получения уведомления в "очереди"
            simulated_notification = {
                'type': 'order_confirmation',
                'order_id': order['id'],
                'user_id': 1
            }
            notification_queue.put_nowait(simulated_notification)
            
            # Проверка, что система корректно обрабатывает уведомление
            received = processor.check_notification_receipt(order['id'])
            assert received['notification_type'] == 'order_confirmation'
        
        # 7. Генерация и валидация сложного отчета (PDF)
        report = ReportGenerator.generate_order_report(order['id'])
        
        # Проверка структуры и данных в отчете
        assert 'order_id' in report['metadata']
        assert report['metadata']['order_id'] == order['id']
        assert report['total_amount'] == 3000
        assert report['items'][0]['product_id'] == 100
        
        # 8. Итоговые интеграционные проверки
        final_order_state = processor.get_full_order_state(order['id'])
        expected_state = {
            'status': 'completed',
            'payment': 'approved',
            'inventory_reserved': True,
            'notifications_sent': True,
            'report_generated': True
        }
        for key, value in expected_state.items():
            assert final_order_state[key] == value

    @pytest.mark.parametrize('failure_scenario', [
        'payment_failed',
        'insufficient_stock',
        'notification_timeout'
    ])
    def test_order_flow_with_failures(self, failure_scenario, db_session):
        """Параметризованный тест для различных сценариев сбоев."""
        
        # Динамическая конфигурация моков для разных сбоев
        if failure_scenario == 'payment_failed':
            with patch('order_service.PaymentGatewayClient._send_request') as mock:
                mock.return_value = MagicMock(status_code=400, json=lambda: {'error': 'Payment rejected'})
                processor = OrderProcessor()
                order = processor.create_order({'user_id': 1, 'product_id': 100, 'quantity': 1})
                payment_result = processor.process_payment(order['id'], 'card_token_xyz')
                assert payment_result['status'] == 'failed'
                assert order['status'] == 'cancelled'  # Автоматическая отмена
            
        elif failure_scenario == 'insufficient_stock':
            inventory = InventoryService(db_session)
            # Сначала резервируем весь stock
            inventory.reserve_stock(100, 10)  # Все 10 единиц
            processor = OrderProcessor()
            order = processor.create_order({'user_id': 1, 'product_id': 100, 'quantity': 1})
            reservation_result = processor.reserve_items(order['id'])
            assert reservation_result['success'] == False
            assert 'out_of_stock' in reservation_result['error']
            
        elif failure_scenario == 'notification_timeout':
            with patch('order_service.NotificationService.send', side_effect=TimeoutError):
                processor = OrderProcessor()
                order = processor.create_order({'user_id': 1, 'product_id': 100, 'quantity': 1})
                with pytest.raises(TimeoutError):
                    processor.send_order_notifications(order['id'])
                # Проверка, что заказ остается в состоянии "paid" несмотря на проблему уведомлений
                assert order['status'] == 'paid'

if __name__ == '__main__':
    pytest.main([__file__, '-v', '--tb=short'])

Ключевые элементы сложного автотеста в примере

1. Мокирование внешних зависимостей (mock_payment_gateway, mock_notification_service) — обеспечивает изоляцию теста от реальных сервисов, позволяя имитировать различные ответы (успешные, ошибки, таймауты).

2. Параметризованное тестирование (@pytest.mark.parametrize) — позволяет одним тестовым методом покрыть несколько сценариев сбоев, уменьшая дублирование кода.

3. Управление состоянием БД через фикстуры (setup_test_data) — гарантирует чистоту тестовых данных и автоматическую очистку после каждого теста, что критично для повторяемости.

4. Асинхронные операции — тест имитирует очередь уведомлений и проверяет их обработку, что отражает реальную архитектуру микросервисов.

5. Многоуровневая валидация — тест проверяет не только конечный результат, но промежуточные состояния:

   • Статус заказа после создания
   • Обновление库存 после резервирования
   • Данные в платежном ответе
   • Корректность записи в БД
   • Структуру сложного PDF-отчета
   • Итоговое интеграционное состояние

6. Контекстные менеджеры и фикстуры — обеспечивают управление ресурсами и моками в разных контекстах выполнения.

Рекомендации для реализации сложных автотестов

• Принцип модульности — разбивайте сложный тест на логические этапы (создание, платеж, уведомления, отчет).
• Использование стабинга — когда мокирование недостаточно (например, для тестирования сетевых форматов), используйте инструменты типа WireMock для стабинга внешних API.
• Параллельное выполнение — для тестов с длительными операциями рассмотрите pytest-xdist для параллельного запуска независимых сценариев.
• Детальная logging и отчетность — добавьте в тест логирование ключевых шагов и создание allure-отчетов для визуализации сложных flow.
• Тестирование восстановления после сбоев — как показано в параметризованной части, всегда включайте сценарии с ошибками и проверяйте, как система восстанавливается или откатывает операции.

Этот пример демонстрирует, что сложный автотест — не просто длинный скрипт, а структурированная, поддерживаемая и надежная проверка бизнес-процесса, которая учитывает множество аспектов системы и обеспечивает высокое качество покрытия.