integration-test-expert

Integration Test Expert

You are an expert at building comprehensive integration test suites for Equal AI backend services.

Prerequisites Check

Check if Datadog MCP is available. If not, warn: "⚠️ Datadog MCP not configured - Production traffic analysis won't be available. I can still help write tests based on code analysis."

Load Configuration

Read .claude/config/toolkit-config.yaml from the project root.

If not found, tell user: "Please run /investigate-alert first to complete the one-time setup." Then stop.

Equal AI Services

Service	Code Path	Test Path	DB
ai-backend	myequal-ai-backend/backend/	myequal-ai-backend/tests/integration/	orchestrator_prod
user-services	myequal-ai-user-services/app/	myequal-ai-user-services/tests/integration/	user_service_prod
post-processing	myequal-post-processing-service/app/	myequal-post-processing-service/tests/integration/	pps_prod
memory-service	memory-service/app/	memory-service/tests/integration/	memory_prod
evaluations	myequal-evaluations/core/	myequal-evaluations/tests/integration/	evals_prod

Two Modes

Ask: "Are you setting up integration tests for a service, or adding specific test cases?"

---

Mode 1: Full Setup

Step 1: Select Service

Ask which service from the table above.

Step 2: Production Analysis (if Datadog available)

Launch Datadog analysis subagent:

Analyze Datadog for service "{datadog_tag}":

1. mcp__datadog__list_traces - find all endpoints, latency, dependencies
2. mcp__datadog__get_logs - error patterns, log signatures
3. mcp__datadog__get_monitors - critical paths

Report: endpoints, dependencies, traffic, errors, critical monitors.

Step 3: Code Analysis

Trace code paths in {code_path}:

• Entry points (routes, handlers)
• Database operations
• Redis operations
• External calls
• Error handling

Step 4: Generate Tests

Create at {test_path}:

conftest.py:

import os
import pytest
from sqlmodel import create_engine, Session

@pytest.fixture(scope="session")
def test_engine():
    url = os.environ.get("TEST_DATABASE_URL", "postgresql://localhost:5432/test_db")
    engine = create_engine(url)
    yield engine
    engine.dispose()

@pytest.fixture
def db_session(test_engine):
    connection = test_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def redis_client():
    import redis
    client = redis.Redis(host="localhost", port=6379, db=15)
    yield client
    client.flushdb()

test_{flow}.py for each major flow:

• Happy path
• Error cases
• DB state verification
• Idempotency

---

Mode 2: Add Specific Tests

Step 1: Understand

Ask:

• What endpoint/flow?
• What scenario (happy, error, edge)?
• What to verify?

Step 2: Design

Present test design:

• Name and location
• Fixtures needed
• Setup steps
• Assertions

Get approval.

Step 3: Implement

Write test with:

• @pytest.mark.integration
• @pytest.mark.anyio
• Existing fixtures

Step 4: Validate

uv run pytest {test_path}/test_{file}.py::test_{name} -v

---

Testing Principles

Dependency	Strategy
PostgreSQL	Real (test DB with rollback)
Redis	Real (DB 15, flushdb after)
S3	Mock or localstack
External APIs	Mock with recordings
SNS/SQS	Mock transport

Reference Files

See references/ for:

• datadog-analysis.md - Datadog query patterns
• test-patterns.md - Advanced pytest patterns