AI-система автоматической генерации integration-тестов

AI-автогенерация интеграционных тестов

Интеграционные тесты проверяют взаимодействие компонентов: сервис A вызывает сервис B, результат пишется в базу, событие летит в очередь. Написать такие тесты сложнее, чем unit — нужно понять граф зависимостей, настроить тестовые окружения, расставить fixtures. AI-генератор анализирует архитектуру сервисов, схему базы данных и OpenAPI-спецификации, создавая тесты для критических интеграционных путей.

Анализ зависимостей и генерация тестов

from langchain_openai import ChatOpenAI
import json
from pathlib import Path
import yaml

class IntegrationTestGenerator:
    INTEGRATION_PROMPT = """Создай интеграционный тест для взаимодействия компонентов.

Компоненты и их контракты:
{components}

Схема базы данных:
{db_schema}

Интеграционный сценарий:
{scenario}

Требования:
- pytest + SQLAlchemy для работы с БД (используй транзакции с rollback)
- httpx.AsyncClient для HTTP-вызовов
- pytest-asyncio для async тестов
- Тестовая база данных через pytest fixture (не боевая!)
- Проверяй не только HTTP-статус, но и состояние в БД после операции
- Используй factory_boy или pytest-factoryboy для тестовых данных

Верни полный код теста с fixtures."""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

    def generate_db_integration_tests(
        self,
        model_code: str,
        repository_code: str,
        db_schema: str
    ) -> str:
        prompt = f"""Создай pytest интеграционные тесты для Repository и Model.

SQLAlchemy Model:
```python
{model_code}

Repository:

{repository_code}

DDL схема:

{db_schema}

Создай тесты:

1. CRUD операции (create, read, update, delete)
2. Фильтрация и сортировка
3. Транзакции (успешный commit, rollback при ошибке)
4. Уникальные ограничения (попытка вставить дубликат)
5. Foreign key constraints
6. Пагинация (если есть в Repository)

Fixtures:

• db_session: SQLAlchemy session с rollback после каждого теста
• test_data factories через factory_boy

Верни код тестов.""" return self.llm.invoke(prompt).content

def generate_service_integration_tests(
    self,
    service_a_spec: dict,
    service_b_spec: dict,
    interaction_patterns: list[str]
) -> str:
    """Генерирует тесты для взаимодействия между сервисами"""
    prompt = f"""Создай pytest интеграционные тесты для взаимодействия сервисов.

Сервис A (клиент):

• Base URL: {service_a_spec['base_url']}
• Вызывает: {json.dumps(service_a_spec['calls'], ensure_ascii=False)}

Сервис B (сервер):

• Endpoints: {json.dumps(service_b_spec.get('endpoints', []), ensure_ascii=False)}

Паттерны взаимодействия: {chr(10).join(f"- {p}" for p in interaction_patterns)}

Используй:

• pytest + respx для мокирования HTTP-ответов сервиса B
• Тесты на retry-логику (что происходит при таймауте сервиса B)
• Тесты на circuit breaker (если есть)
• Тесты на корректную обработку 4xx/5xx от сервиса B

Верни код тестов с fixtures.""" return self.llm.invoke(prompt).content

### Генерация fixtures и тестовых данных

```python
class TestDataGenerator:
    FACTORY_PROMPT = """Создай factory_boy factories для моделей.

SQLAlchemy модели:
{models_code}

Создай:
1. Factory для каждой модели
2. SubFactory для связанных объектов
3. Trait для специфических состояний (напр. expired_user, admin_user)
4. Batch-создание через factory.build_batch

Верни код factories."""

    async def generate_factories(self, models_code: str) -> str:
        result = await self.llm.ainvoke(
            self.FACTORY_PROMPT.format(models_code=models_code)
        )
        return result.content

    async def generate_fixtures_from_schema(self, schema: dict) -> str:
        """Создаёт pytest fixtures из схемы БД"""
        prompt = f"""Создай pytest fixtures для тестовой базы данных.

Схема: {json.dumps(schema, ensure_ascii=False, indent=2)}

Fixtures нужны:
- engine: SQLAlchemy engine к test DB (PostgreSQL через pytest-postgresql)
- db_session: сессия с rollback после каждого теста
- Fixtures для каждой таблицы: минимальный валидный объект
- seeded_db: база с начальными данными для e2e тестов

Используй scope='function' для db_session, scope='session' для engine.
Верни Python код."""
        return (await self.llm.ainvoke(prompt)).content

Тесты для очередей сообщений

    QUEUE_INTEGRATION_PROMPT = """Создай тест для интеграции с очередью сообщений.

Producer-код:
{producer_code}

Consumer-код:
{consumer_code}

Используй:
- pytest + testcontainers-python (RabbitMQ/Kafka в Docker)
- Проверяй: сообщение отправлено → обработано → результат в БД
- Тест на dead letter queue (сообщение не обработано → DLQ)
- Тест на идемпотентность (повторная обработка того же сообщения)
- Тайм-аут ожидания обработки: asyncio.wait_for с timeout=10

Верни полный код теста."""

Кейс: e-commerce платформа, 15 микросервисов. Проблема: интеграционное тестирование делалось вручную перед каждым релизом (2 QA × 3 дня). Сгенерировали 180 интеграционных тестов для критических путей: заказ → оплата → уведомление → обновление склада. Из 180 тестов при первом запуске упало 23 — нашли реальные баги в обработке currency conversion, дублировании событий в очереди, неверном статусе при partial payment.

Сроки: интеграционные тесты для БД + сервисов: 3–5 недель; с очередями и полным окружением через testcontainers: 5–7 недель.