エージェントパイプラインの統合テスト

エージェントの統合テストとは

エンドツーエンドテストの構成

import pytest

# Mark as integration test — skipped in fast unit test runs
@pytest.mark.integration
def test_research_agent_full_pipeline():
    from myagent import ResearchAgent

    agent = ResearchAgent(
        openai_api_key='YOUR_TEST_KEY',
        search_api_key='YOUR_TEST_KEY'
    )

    result = agent.run('What is the population of Tokyo?')

    # Structural assertions — not exact string matching
    assert isinstance(result, dict)
    assert result['status'] == 'completed'
    assert 'tokyo' in result['answer'].lower() or 'japan' in result['answer'].lower()
    assert len(result['sources']) >= 1

テストデータの分離

import os
import pytest

# Use a separate test database URL
@pytest.fixture(scope='session')
def test_db():
    test_db_url = os.environ.get(
        'TEST_DATABASE_URL',
        'postgresql://localhost/myagent_test'  # separate test DB
    )
    # Set up test schema
    from myagent.database import create_tables
    create_tables(test_db_url)
    yield test_db_url
    # Tear down after all tests in the session
    from myagent.database import drop_tables
    drop_tables(test_db_url)

各テスト後のクリーンアップ

import pytest

@pytest.fixture
def clean_agent_memory(test_db):
    # No setup needed — DB starts empty
    yield
    # Cleanup: delete any records created during this test
    from myagent.database import clear_conversation_history
    clear_conversation_history(test_db)

@pytest.mark.integration
def test_agent_stores_conversation(clean_agent_memory, test_db):
    from myagent import Agent
    agent = Agent(db_url=test_db)

    agent.run('Remember that my name is Alex')
    history = agent.get_history()

    assert len(history) > 0
    assert any('Alex' in str(msg) for msg in history)
    # clean_agent_memory fixture deletes these after the test

サンドボックス化されたサービス：テスト用APIキー

import os
import pytest

# Skip integration tests if test keys are not configured
def requires_integration_keys():
    return pytest.mark.skipif(
        not os.environ.get('OPENAI_TEST_KEY'),
        reason='Integration test keys not configured'
    )

@requires_integration_keys()
@pytest.mark.integration
def test_live_weather_tool():
    from myagent.tools import get_weather
    result = get_weather(city='London', unit='celsius')

    assert result['success'] is True
    assert 'temperature' in result
    assert isinstance(result['temperature'], (int, float))

サンドボックス化されたデータベースにDockerを使う

# conftest.py — docker-based test database
import subprocess
import pytest

@pytest.fixture(scope='session')
def docker_postgres():
    container_id = subprocess.check_output([
        'docker', 'run', '-d',
        '-e', 'POSTGRES_PASSWORD=test',
        '-e', 'POSTGRES_DB=agent_test',
        '-p', '5434:5432',  # use non-standard port to avoid conflicts
        'postgres:15'
    ]).decode().strip()

    import time
    time.sleep(2)  # wait for Postgres to start

    yield 'postgresql://postgres:test@localhost:5434/agent_test'

    subprocess.run(['docker', 'stop', container_id])
    subprocess.run(['docker', 'rm', container_id])

環境別のテスト設定

import os

def get_test_config() -> dict:
    env = os.environ.get('TEST_ENV', 'local')

    configs = {
        'local': {
            'db_url': 'postgresql://localhost/agent_test',
            'openai_key': os.environ.get('OPENAI_TEST_KEY', ''),
            'use_real_llm': False  # use mocks locally
        },
        'ci': {
            'db_url': os.environ.get('CI_DATABASE_URL', ''),
            'openai_key': os.environ.get('CI_OPENAI_KEY', ''),
            'use_real_llm': True  # use real LLM in CI integration tests
        },
        'staging': {
            'db_url': os.environ.get('STAGING_DATABASE_URL', ''),
            'openai_key': os.environ.get('STAGING_OPENAI_KEY', ''),
            'use_real_llm': True
        }
    }
    return configs[env]

config = get_test_config()

print(f"TEST_ENV not set -> using '{os.environ.get('TEST_ENV', 'local')}' config")
print(f"DB URL       : {config['db_url']}")
print(f"Use real LLM : {config['use_real_llm']}")

テスト選択のためのpytestマーカー

# pytest.ini
# [pytest]
# markers =
#     unit: Fast unit tests with mocked dependencies
#     integration: Slower tests with real or sandboxed services
#     expensive: Tests that make real LLM calls and cost money

# Run only unit tests (fast CI check):
# pytest -m unit

# Run only integration tests:
# pytest -m integration

# Run everything except expensive tests:
# pytest -m 'not expensive'

# In test files:
import pytest

@pytest.mark.unit
def test_tool_format():
    pass  # fast, no external calls

@pytest.mark.integration
@pytest.mark.expensive
def test_with_real_llm():
    pass  # slow, costs tokens

CIで統合テストを実行する

# .github/workflows/test.yml (abbreviated)
# name: Tests
# on:
#   push:
#     branches: [main, develop]
#   schedule:
#     - cron: '0 2 * * *'  # nightly integration tests
#
# jobs:
#   unit-tests:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v3
#       - run: pip install -r requirements.txt
#       - run: pytest -m unit --tb=short
#
#   integration-tests:
#     if: github.event_name == 'schedule'
#     env:
#       CI_OPENAI_KEY: ${{ secrets.CI_OPENAI_KEY }}
#       CI_DATABASE_URL: ${{ secrets.CI_DATABASE_URL }}
#     steps:
#       - run: pytest -m integration --tb=long

print('Unit tests on every push, integration tests nightly')

エージェントのテストカバレッジを測定する

# Install: pip install pytest-cov

# Run tests with coverage report:
# pytest --cov=myagent --cov-report=html -m unit

# This generates an HTML report showing which lines are untested
# Uncovered lines in the agent loop are high-risk areas

# Example coverage config in pyproject.toml:
# [tool.coverage.run]
# omit = ["tests/*", "scripts/*"]
#
# [tool.coverage.report]
# fail_under = 80  # fail if coverage drops below 80%

print('Coverage reports highlight untested code paths in your agent')

統合テストのベストプラクティスまとめ

理解度チェック：統合テスト

復習：エージェントパイプラインの統合テスト