Claude API のReplay駆動テスト — 本番応答を録画して回帰テストを成立させる設計パターン

# tests/cassette.py
"""Claude API のレスポンスを録画・再生するカセット実装。
本物の API を1回だけ呼び、以後はファイルから再生する。
record_mode="once" で初回のみ録画、それ以降は再生する設計。
"""
 
import json
import hashlib
import os
from pathlib import Path
from typing import Any
from contextlib import contextmanager
from unittest.mock import patch
from anthropic import Anthropic
from anthropic.types import Message
 
CASSETTE_DIR = Path(__file__).parent / "cassettes"
 
def _request_key(messages: list[dict], model: str, tools: list | None = None) -> str:
    """リクエストの内容から決定的なキーを生成する。
    同じカセット内で複数リクエストを区別するために使う。
    """
    payload = json.dumps(
        {"messages": messages, "model": model, "tools": tools or []},
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
 
@contextmanager
def cassette(name: str, record_mode: str = "once"):
    """テストでカセットを使うためのコンテキストマネージャ。
 
    使い方:
        with cassette("test_basic_response"):
            client = Anthropic()
            res = client.messages.create(...)
 
    record_mode:
        "once":         カセットが無ければ録画。あれば再生のみ。
        "new_episodes": カセットがあっても新しいリクエストは追記録画。
        "none":         録画禁止。録画が無ければ AssertionError で失敗。
    """
    CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    cassette_path = CASSETTE_DIR / f"{name}.json"
 
    # 既存カセットを読み込み
    if cassette_path.exists():
        with cassette_path.open("r", encoding="utf-8") as f:
            recordings: dict[str, Any] = json.load(f)
    else:
        recordings = {}
 
    new_recordings: dict[str, Any] = {}
    real_create = Anthropic().messages.create
 
    def patched_create(self, **kwargs):
        messages = kwargs.get("messages", [])
        model = kwargs.get("model", "")
        tools = kwargs.get("tools")
        key = _request_key(messages, model, tools)
 
        # 再生モード
        if key in recordings:
            return Message.model_validate(recordings[key])
 
        # 録画モード
        if record_mode == "none":
            raise AssertionError(
                f"カセット {name} に key={key} の録画がありません。"
                f"record_mode='once' で再録画してください。"
            )
 
        response = real_create(**kwargs)
        new_recordings[key] = response.model_dump(mode="json")
        return response
 
    with patch.object(
        Anthropic.messages.__class__, "create", patched_create
    ):
        try:
            yield
        finally:
            if new_recordings and record_mode != "none":
                recordings.update(new_recordings)
                with cassette_path.open("w", encoding="utf-8") as f:
                    json.dump(recordings, f, ensure_ascii=False, indent=2)

# tests/test_summarizer.py
"""録画されたレスポンスでロジックの回帰テストを行う例。"""
 
from anthropic import Anthropic
from src.summarizer import summarize_long_text
from tests.cassette import cassette
 
def test_summarizer_truncates_long_input():
    """長文入力時に要約ロジックが期待通り動くことを検証する。"""
    long_text = "Anthropic は 2026 年に..." * 100  # 実際の長文を想定
 
    with cassette("summarizer_long_input"):
        result = summarize_long_text(long_text)
 
    # ロジック側の期待値を検証する（API 応答そのものではなく、
    # アプリケーション層での処理結果を検証する点が重要）
    assert len(result) < len(long_text) / 2
    assert "Anthropic" in result
    assert result.endswith("。")

# tests/streaming_cassette.py
"""Streaming（SSE）レスポンスを JSONL で録画・再生する。
content_block_delta などのイベント順序を完全に再現できる。
"""
 
import json
from pathlib import Path
from contextlib import contextmanager
from typing import Iterator
from anthropic import Anthropic
from anthropic.types import RawMessageStreamEvent
 
STREAM_CASSETTE_DIR = Path(__file__).parent / "cassettes" / "streams"
 
class CassetteStream:
    """録画されたイベント列を返すストリームのモック。
    本物の MessageStreamManager と同じインターフェースを提供する。
    """
 
    def __init__(self, events: list[dict]):
        self._events = events
 
    def __enter__(self):
        return self
 
    def __exit__(self, *args):
        return False
 
    def __iter__(self) -> Iterator[RawMessageStreamEvent]:
        for ev in self._events:
            yield RawMessageStreamEvent.model_validate(ev)
 
@contextmanager
def stream_cassette(name: str, record_mode: str = "once"):
    """Streaming API 用カセット。"""
    STREAM_CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    path = STREAM_CASSETTE_DIR / f"{name}.jsonl"
 
    if path.exists():
        events = [json.loads(line) for line in path.read_text("utf-8").splitlines()]
    else:
        events = []
 
    captured: list[dict] = []
    real_stream = Anthropic().messages.stream
 
    def patched_stream(self, **kwargs):
        if events:
            return CassetteStream(events)
 
        if record_mode == "none":
            raise AssertionError(
                f"Stream カセット {name} が見つかりません。"
            )
 
        # 本物のストリームをラップして全イベントを保存しながら yield
        original = real_stream(**kwargs)
 
        class RecordingStream:
            def __enter__(_self):
                _self._inner = original.__enter__()
                return _self
 
            def __exit__(_self, *a):
                return _self._inner.__exit__(*a)
 
            def __iter__(_self):
                for ev in _self._inner:
                    captured.append(ev.model_dump(mode="json"))
                    yield ev
 
        return RecordingStream()
 
    from unittest.mock import patch
 
    with patch.object(
        Anthropic.messages.__class__, "stream", patched_stream
    ):
        try:
            yield
        finally:
            if captured:
                with path.open("w", encoding="utf-8") as f:
                    for ev in captured:
                        f.write(json.dumps(ev, ensure_ascii=False) + "\n")

# tests/test_weather_agent.py
"""ツール呼び出しを含むエージェントの多段会話をカセットで検証する。
Claude が weather ツールを呼び、結果を受けて自然文を返すまでの全往復を録画する。
"""
 
from src.agents.weather_agent import WeatherAgent
from tests.cassette import cassette
 
def fake_get_weather(location: str) -> dict:
    """テスト用のツール実装。本番のツールロジックをモックしない方針なら
    本物の関数を呼んでも良いが、ここでは外部 API 依存を避けて固定値を返す。
    """
    return {"location": location, "temp_c": 18, "condition": "晴れ"}
 
def test_weather_agent_handles_japanese_location():
    """日本の地名でも正しくツール呼び出しと最終応答が成立することを検証。"""
    with cassette("weather_agent_kyoto"):
        agent = WeatherAgent(tool_impl=fake_get_weather)
        reply = agent.ask("京都の天気を教えてください")
 
    assert "晴れ" in reply
    assert "京都" in reply
    # ツールが正しく呼ばれたことの副次検証
    assert agent.tool_call_count == 1

# tests/conftest.py
"""pytest の全体設定。CI 上では録画モードを禁止する。"""
 
import os
import pytest
 
def pytest_addoption(parser):
    parser.addoption(
        "--record",
        action="store_true",
        default=False,
        help="カセットを録画する（ローカル開発専用）",
    )
 
@pytest.fixture
def record_mode(request):
    """テスト関数からこの fixture を受け取り、cassette() に渡す。
    CI 上では強制的に 'none' になる。
    """
    if os.getenv("CI"):
        return "none"
    return "once" if request.config.getoption("--record") else "none"

# 使い方の例
def test_with_record_guard(record_mode):
    with cassette("my_test", record_mode=record_mode):
        # ローカルで --record 付きで1回実行 → カセット作成
        # 以後 CI でも `record_mode='none'` で再生される
        ...

# scripts/replay_incident.py
"""本番障害の再現用スクリプト。trace_id から R2 のリクエストペアを引き、
カセット形式に変換してテスト用ディレクトリに配置する。
"""
 
import json
import sys
from pathlib import Path
import boto3
from tests.cassette import _request_key
 
INCIDENT_CASSETTE_DIR = Path("tests/cassettes/incidents")
R2_BUCKET = "claude-prod-traces"
 
def replay(trace_id: str) -> Path:
    """trace_id から本番リクエストペアを取得し、カセットファイルとして保存する。"""
    s3 = boto3.client(
        "s3",
        endpoint_url="https://<account>.r2.cloudflarestorage.com",
    )
    obj = s3.get_object(Bucket=R2_BUCKET, Key=f"requests/{trace_id}.json")
    pair = json.loads(obj["Body"].read())
 
    # pair = {"request": {...}, "response": {...}}
    key = _request_key(
        pair["request"]["messages"],
        pair["request"]["model"],
        pair["request"].get("tools"),
    )
    INCIDENT_CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    cassette_path = INCIDENT_CASSETTE_DIR / f"{trace_id}.json"
    cassette_path.write_text(
        json.dumps({key: pair["response"]}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"✅ カセット作成: {cassette_path}")
    return cassette_path
 
if __name__ == "__main__":
    replay(sys.argv[1])

// tests/cassette.ts
/**
 * Claude API カセット（TypeScript 版）
 * Vitest と組み合わせて使う前提
 */
import fs from "node:fs";
import path from "node:path";
import crypto from "node:crypto";
import Anthropic from "@anthropic-ai/sdk";
import type { Message, MessageCreateParams } from "@anthropic-ai/sdk/resources/messages";
 
const CASSETTE_DIR = path.join(__dirname, "cassettes");
 
function requestKey(params: MessageCreateParams): string {
  const payload = JSON.stringify({
    messages: params.messages,
    model: params.model,
    tools: params.tools ?? [],
  });
  return crypto.createHash("sha256").update(payload).digest("hex").slice(0, 16);
}
 
export async function withCassette<T>(
  name: string,
  fn: (client: Anthropic) => Promise<T>,
  recordMode: "once" | "none" = process.env.CI ? "none" : "once",
): Promise<T> {
  fs.mkdirSync(CASSETTE_DIR, { recursive: true });
  const cassettePath = path.join(CASSETTE_DIR, `${name}.json`);
  const recordings: Record<string, Message> = fs.existsSync(cassettePath)
    ? JSON.parse(fs.readFileSync(cassettePath, "utf-8"))
    : {};
  const newRecordings: Record<string, Message> = {};
 
  const realClient = new Anthropic();
  const proxiedClient = new Proxy(realClient, {
    get(target, prop, receiver) {
      if (prop === "messages") {
        return new Proxy(target.messages, {
          get(messagesTarget, messagesProp) {
            if (messagesProp === "create") {
              return async (params: MessageCreateParams) => {
                const key = requestKey(params);
                if (recordings[key]) return recordings[key];
                if (recordMode === "none") {
                  throw new Error(
                    `カセット ${name} に key=${key} の録画がありません`,
                  );
                }
                const res = await messagesTarget.create(params);
                newRecordings[key] = res as Message;
                return res;
              };
            }
            return Reflect.get(messagesTarget, messagesProp);
          },
        });
      }
      return Reflect.get(target, prop, receiver);
    },
  });
 
  try {
    return await fn(proxiedClient);
  } finally {
    if (Object.keys(newRecordings).length > 0 && recordMode !== "none") {
      fs.writeFileSync(
        cassettePath,
        JSON.stringify({ ...recordings, ...newRecordings }, null, 2),
      );
    }
  }
}

def _atomic_write_json(path: Path, data: dict) -> None:
    """並列実行でも安全に JSON を保存する。"""
    tmp = path.with_suffix(path.suffix + ".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)
    tmp.replace(path)  # POSIX で原子的なリネーム