integration-test

name	integration-test
description	Generate integration tests for service-to-service communication. Use when: (1) Testing API endpoints with real database, (2) Testing service interactions, (3) Docker-based test environments, (4) End-to-end workflow validation. Creates pytest fixtures with testcontainers, async database setup, and cleanup. NOT for unit tests (use mock-provider) or load testing.

Integration Test Generator

Generate integration tests for microservice communication.

Quick Reference

/integration-test api              # API endpoint tests
/integration-test service          # Service layer tests
/integration-test workflow         # Multi-service workflow tests
/integration-test <name>           # Custom integration tests

Generated Structure

tests/
├── integration/
│   ├── conftest.py           # Shared fixtures
│   ├── test_api.py           # API tests
│   ├── test_conversations.py # Domain tests
│   └── test_workflows.py     # E2E workflows
└── docker-compose.test.yml   # Test infrastructure

Implementation

1. Test Configuration

# tests/integration/conftest.py
import pytest
import asyncio
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

from main import app
from core.database import get_session
from core.redis import get_redis

# Event loop fixture
@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# PostgreSQL container
@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16") as postgres:
        yield postgres

# Redis container
@pytest.fixture(scope="session")
def redis_container():
    with RedisContainer("redis:7") as redis:
        yield redis

# Async engine
@pytest.fixture(scope="session")
async def engine(postgres_container):
    url = postgres_container.get_connection_url().replace(
        "postgresql://", "postgresql+asyncpg://"
    )
    engine = create_async_engine(url, echo=False)

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

    yield engine

    await engine.dispose()

# Database session
@pytest.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
    async_session = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        yield session
        await session.rollback()

# Redis client
@pytest.fixture
async def redis_client(redis_container):
    import redis.asyncio as redis
    client = redis.from_url(f"redis://{redis_container.get_container_host_ip()}:{redis_container.get_exposed_port(6379)}")
    yield client
    await client.flushall()
    await client.aclose()

# Test client
@pytest.fixture
async def client(db_session, redis_client) -> AsyncGenerator[AsyncClient, None]:
    app.dependency_overrides[get_session] = lambda: db_session
    app.dependency_overrides[get_redis] = lambda: redis_client

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

    app.dependency_overrides.clear()

2. API Tests

# tests/integration/test_api.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
class TestConversationsAPI:
    async def test_create_conversation(self, client: AsyncClient):
        response = await client.post(
            "/api/v1/conversations",
            json={"user_id": "user123", "title": "Test Chat"},
        )

        assert response.status_code == 201
        data = response.json()
        assert data["user_id"] == "user123"
        assert data["title"] == "Test Chat"
        assert "id" in data

    async def test_get_conversation(self, client: AsyncClient):
        # Create first
        create_resp = await client.post(
            "/api/v1/conversations",
            json={"user_id": "user123"},
        )
        conv_id = create_resp.json()["id"]

        # Then get
        response = await client.get(f"/api/v1/conversations/{conv_id}")

        assert response.status_code == 200
        assert response.json()["id"] == conv_id

    async def test_list_conversations(self, client: AsyncClient):
        # Create multiple
        for i in range(3):
            await client.post(
                "/api/v1/conversations",
                json={"user_id": "user123", "title": f"Chat {i}"},
            )

        response = await client.get(
            "/api/v1/conversations",
            params={"user_id": "user123"},
        )

        assert response.status_code == 200
        data = response.json()
        assert len(data["items"]) == 3

    async def test_add_message(self, client: AsyncClient):
        # Create conversation
        conv_resp = await client.post(
            "/api/v1/conversations",
            json={"user_id": "user123"},
        )
        conv_id = conv_resp.json()["id"]

        # Add message
        response = await client.post(
            f"/api/v1/conversations/{conv_id}/messages",
            json={"role": "user", "content": "Hello!"},
        )

        assert response.status_code == 201
        assert response.json()["content"] == "Hello!"

3. Workflow Tests

# tests/integration/test_workflows.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
class TestChatWorkflow:
    """Test complete chat workflow."""

    async def test_complete_conversation_flow(self, client: AsyncClient):
        # 1. Create conversation
        conv_resp = await client.post(
            "/api/v1/conversations",
            json={"user_id": "user123", "title": "Test Conversation"},
        )
        assert conv_resp.status_code == 201
        conv_id = conv_resp.json()["id"]

        # 2. Add user message
        msg1_resp = await client.post(
            f"/api/v1/conversations/{conv_id}/messages",
            json={"role": "user", "content": "What is 2+2?"},
        )
        assert msg1_resp.status_code == 201

        # 3. Add assistant response
        msg2_resp = await client.post(
            f"/api/v1/conversations/{conv_id}/messages",
            json={"role": "assistant", "content": "2+2 equals 4."},
        )
        assert msg2_resp.status_code == 201

        # 4. Get conversation with messages
        conv_resp = await client.get(
            f"/api/v1/conversations/{conv_id}",
            params={"include_messages": True},
        )
        assert conv_resp.status_code == 200
        data = conv_resp.json()
        assert len(data["messages"]) == 2

        # 5. Archive conversation
        archive_resp = await client.patch(
            f"/api/v1/conversations/{conv_id}",
            json={"is_archived": True},
        )
        assert archive_resp.status_code == 200
        assert archive_resp.json()["is_archived"] is True

4. Docker Compose for Tests

# tests/docker-compose.test.yml
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: test
      POSTGRES_PASSWORD: test
      POSTGRES_DB: test
    ports:
      - "5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U test"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6380:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

Running Tests

# With testcontainers (auto-starts containers)
pytest tests/integration -v

# With docker-compose
docker-compose -f tests/docker-compose.test.yml up -d
DATABASE_URL=postgresql+asyncpg://test:test@localhost:5433/test \
REDIS_URL=redis://localhost:6380 \
pytest tests/integration -v