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

# tests/cassette.py
"""Claude API のレスポンスを録画・再生するカセット実装。
本物の API を 1 回だけ呼び、以後はファイルから再生します。
 
差し替え先は `Anthropic().messages` ではなくリソースクラス
`anthropic.resources.messages.Messages` です（理由は直後の節に書きました）。
"""
 
import json
import hashlib
from pathlib import Path
from typing import Any
from contextlib import contextmanager
from unittest.mock import patch
from anthropic.resources.messages import Messages
from anthropic.types import Message
 
CASSETTE_DIR = Path(__file__).parent / "cassettes"
 
# キーに含める引数。ここに入れ忘れた引数は「変えても同じ録画が返る」ことになります。
KEYED_FIELDS = (
    "messages",
    "model",
    "system",
    "tools",
    "tool_choice",
    "max_tokens",
    "temperature",
)
 
def _request_key(kwargs: dict[str, Any]) -> str:
    """リクエストの内容から決定的なキーを生成します。"""
    payload = json.dumps(
        {k: kwargs.get(k) for k in KEYED_FIELDS},
        sort_keys=True,
        ensure_ascii=False,
        default=str,  # NotGiven 等の番兵オブジェクトで落ちないように
    )
    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":         カセットが無ければ録画。あれば再生のみで、
                        未録画のリクエストが来たら AssertionError。
        "new_episodes": カセットがあっても未録画のリクエストは追記録画。
        "none":         録画禁止。録画が無ければ AssertionError で失敗。
    """
    CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    cassette_path = CASSETTE_DIR / f"{name}.json"
    had_cassette = cassette_path.exists()
 
    if had_cassette:
        recordings: dict[str, Any] = json.loads(
            cassette_path.read_text(encoding="utf-8")
        )
    else:
        recordings = {}
 
    new_recordings: dict[str, Any] = {}
    real_create = Messages.create  # 未束縛のまま退避する（self は呼び出し側から届きます）
 
    def patched_create(self, **kwargs):
        key = _request_key(kwargs)
 
        # 再生
        if key in recordings:
            return Message.model_validate(recordings[key])
 
        # 録画を許さない条件を、モードごとに明示する
        if record_mode == "none" or (record_mode == "once" and had_cassette):
            raise AssertionError(
                f"カセット {name} に key={key} の録画がありません。"
                f"ローカルで record_mode='new_episodes' を指定して録り足してください。"
            )
 
        response = real_create(self, **kwargs)
        new_recordings[key] = response.model_dump(mode="json")
        return response
 
    with patch.object(Messages, "create", patched_create):
        try:
            yield
        finally:
            if new_recordings:
                recordings.update(new_recordings)
                cassette_path.write_text(
                    json.dumps(
                        recordings, ensure_ascii=False, indent=2, sort_keys=True
                    ),
                    encoding="utf-8",
                )

# 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 unittest.mock import patch
from pydantic import TypeAdapter
from anthropic.resources.messages import Messages
from anthropic.types import RawMessageStreamEvent
 
STREAM_CASSETTE_DIR = Path(__file__).parent / "cassettes" / "streams"
 
# RawMessageStreamEvent は Annotated Union なので、クラスメソッドの
# model_validate は生えていません。TypeAdapter を 1 つ作って使い回します。
EVENT_ADAPTER = TypeAdapter(RawMessageStreamEvent)
 
class CassetteStream:
    """録画されたイベント列を返すストリームです。
    `with client.messages.stream(...) as s:` と同じ使い勝手にします。
    """
 
    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 EVENT_ADAPTER.validate_python(ev)
 
    @property
    def text_stream(self) -> Iterator[str]:
        """本物の MessageStream と同じく、本文デルタだけを流します。"""
        for ev in self:
            if ev.type == "content_block_delta" and ev.delta.type == "text_delta":
                yield ev.delta.text
 
@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 = Messages.stream  # ここも未束縛のまま退避します
 
    def patched_stream(self, **kwargs):
        if events:
            return CassetteStream(events)
 
        if record_mode == "none":
            raise AssertionError(f"Stream カセット {name} が見つかりません。")
 
        original = real_stream(self, **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(EVENT_ADAPTER.dump_python(ev, mode="json"))
                    yield ev
 
        return RecordingStream()
 
    with patch.object(Messages, "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

# 旧キー（messages + model + tools）と、system を含む新キーの比較
old = old_request_key(messages, model, tools=None)
new = _request_key({"messages": messages, "model": model,
                    "system": "簡潔に答えてください", "max_tokens": 64})
 
# system を足す前後で、旧キーは一致してしまう → 古い録画を再生する
# 新キーは別の値になる → 未録画として AssertionError で止まる

# 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");
 
// Python 版と同じく、応答を変えうる引数はすべてキーに入れます
function requestKey(params: MessageCreateParams): string {
  const payload = JSON.stringify({
    messages: params.messages,
    model: params.model,
    system: params.system ?? null,
    tools: params.tools ?? [],
    tool_choice: params.tool_choice ?? null,
    max_tokens: params.max_tokens,
    temperature: params.temperature ?? null,
  });
  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> = {};
 
  // API キーの無い CI でも生成できるよう、ダミーキーを許容します
  // （再生モードでは 1 度もネットワークへ出ません）
  const realClient = new Anthropic({
    apiKey: process.env.ANTHROPIC_API_KEY ?? "replay-only",
  });
  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} の録画がありません`,
                  );
                }
                if (!process.env.ANTHROPIC_API_KEY) {
                  throw new Error("録画には ANTHROPIC_API_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 で原子的なリネーム