実務でAIを運用していて「本番でどう壊れるか」が不安になることは多いはずです。特に外部APIや非同期パイプライン、異常入力に対する振る舞いは、想定外の障害を生みやすく、監視やRunbookと結びついた明確なテストが必要です。本記事では、現場で即使える耐障害（フェイルケース）テストのワークフローを、Pythonを使った自動化パターンとともに整理します。
狙いと適用範囲
狙いは「デプロイ前に実務上で致命的な失敗を早期発見し、運用で扱える形にする」ことです。技術的な細部だけでなく、監視・デプロイ・Runbookとの接続点を重視します。第49回（合成データ）で扱ったデータ生成と、第37回（デプロイ）で扱ったCI/CDの実装は本記事と直接つながります。Runbookの実際的な運用テンプレートは第46回を参照して補完してください。
実務ワークフロー（ステップバイステップ）
| ステップ | 目的 | 実施内容 | 成果物 |
|---|---|---|---|
| テスト要件定義 | 業務に致命的な障害を列挙 | 重要機能の失敗モード洗い出し（例：API不応答、遅延、想定外入力） | テスト要件ドキュメント |
| フェイルケース設計 | 代表的な障害シナリオの設計 | 遅延注入、APIエラー、部分欠損、極端入力、アドバーサリアル変換などを定義 | フェイルケース一覧 |
| 合成データでのケース生成 | 再現性のあるテストデータを用意 | 第49回の合成データを利用してエッジケース・ノイズを生成 | テストデータセット |
| 自動実行 | CIで再現性ある検証 | pytestやスケジュールジョブでフェイルケースを自動化 | 自動テスト結果（ログ・レポート） |
| 判定・レポート生成 | 影響の定量化と対応指示 | 閾値判定（例：応答時間、失敗率）とレポート自動生成 | 問題報告書・優先度付け |
| 修正→再実行 | 回帰防止と改善確認 | 修正をマージし、同じテストをCIで再実行 | 合格基準を満たすリリース |
フェイルケース設計（代表例）
ここでは具体的なケースと生成方法を示します。テーブルで原因、発生条件、合成方法、期待される判定基準を整理します。
| 失敗モード | 発生条件 | 合成方法 | 判定基準（例） |
|---|---|---|---|
| APIタイムアウト/遅延 | 外部APIが高遅延 | 遅延注入ライブラリで応答を遅らせる | タイムアウト発生・リトライ回数超過 |
| APIエラー（5xx/4xx） | 外部がエラー応答 | モックで5xx/4xxを返す | エラーハンドリング（フォールバック/アラート）動作 |
| 部分欠損データ | 入力フィールドが欠落 | 合成データで一部フィールドを削除 | 例外ではなく既定値で処理される |
| 極端入力・境界値 | 長大文字列、極端数値 | 合成データで境界値・長さ超過を生成 | 処理完了または適切なバリデーションエラー |
| アドバーサリアルノイズ | モデル入力が微小に改変される | ノイズを付与した合成データ生成 | 精度低下を定量化（閾値超過ならNG） |
| 外部依存の断絶 | S3やDBが一時的に不可 | 接続エラーをモック/インジェクト | エラーハンドルとリトライ挙動 |
合成データの活用（第49回との連携）
第49回で作成した合成データを活用して、以下を自動生成します。エッジケース、境界値、アドバーサリアルノイズはスクリプトで一括生成でき、テストカバレッジの測定にも使えます。
テストカバレッジ指標と測り方
| 指標 | 定義 | 測定方法 |
|---|---|---|
| 入力領域カバレッジ | 想定入力空間をどれだけ網羅しているか | カテゴリごとのサンプル比率と境界値の網羅率 |
| 失敗モードカバレッジ | 設計した失敗モードのうち何割をテストしているか | 失敗モードごとのテストケース数/総失敗モード数 |
| 回帰率 | 修正後に再発しない割合 | 修正後の再実行でのパス率 |
Pythonでの実装パターン
ここでは「テストハーネスの構成」「サンプルテスト」「簡易エラー注入ライブラリ」「ログ・メトリクス収集」の雛形を示します。実際にはプロジェクトの規模に応じて拡張してください。
推奨ディレクトリ構成
| パス | 役割 |
|---|---|
| tests/ | pytestテストケース |
| tests/fixtures/ | 合成データ・フィクスチャ |
| tests/mocks/ | 外部APIモック定義 |
| tools/fail_injector.py | 遅延・エラー注入ユーティリティ |
| tools/metrics.py | ログ・メトリクス集約 |
簡単なサンプルコード（同期API呼び出しのモックと遅延注入）
pytestとrequestsを使ったテスト例（pytestでrequestsをpatchする方法）。
# tests/test_api_failures.py
import time
from unittest.mock import patch
import requests
from tools.fail_injector import inject_delay, inject_status
def call_service(url):
r = requests.get(url, timeout=2)
return r.status_code, r.text
@patch('requests.get')
def test_api_timeout(mock_get):
# モックが遅延してタイムアウトとなるケース
def slow_get(*args, **kwargs):
inject_delay(3) # 3秒待つ
class R: status_code=200; text='ok'
return R()
mock_get.side_effect = slow_get
try:
status, _ = call_service('https://api.example')
except Exception as e:
assert 'timed out' in str(e).lower() or isinstance(e, TimeoutError)
非同期呼び出しの例（asyncio + aiohttp）
# tests/test_async.py
import asyncio
from aiohttp import ClientSession
async def fetch(session, url):
async with session.get(url, timeout=1) as resp:
return resp.status
async def test_async_timeout(aiohttp_client):
# 実際はaiohttpのサーバを立てて遅延応答を返すかモックする
async with ClientSession() as s:
try:
await fetch(s, 'http://slow.local')
except Exception:
assert True
簡易エラー注入ユーティリティ（tools/fail_injector.py）
# tools/fail_injector.py
import time
import random
def inject_delay(seconds):
time.sleep(seconds)
def inject_error(rate=0.1, code=500):
if random.random() < rate:
raise Exception(f'injected error {code}')
ログ・メトリクス収集の雛形（tools/metrics.py）
# tools/metrics.py
import logging
import json
logger = logging.getLogger('resilience')
handler = logging.FileHandler('resilience.log')
logger.addHandler(handler)
logger.setLevel(logging.INFO)
def record_event(name, payload):
entry = {'event': name, 'payload': payload}
logger.info(json.dumps(entry))
def record_metric(name, value, tags=None):
record_event('metric', {'name': name, 'value': value, 'tags': tags or {}})
上記は簡易例です。プロダクションではPrometheusやCloud Loggingと連携してください。
CI/CDと結びつける運用
テスト自動実行のタイミングと対応策をあらかじめ決めます。
- PR発行時：単体/統合のフェイルケースを実行（軽量）
- マージ前：主要な失敗モードを含むフルスイートを実行
- 定期バッチ（夜間）：長時間テストや確率的障害のシミュレーション
- カナリア連携：カナリアで観測された指標が悪化した場合にロールバックと該当テストの実行
失敗時の自動フロー（例）:
| 事象 | 自動対応 | 運用アクション |
|---|---|---|
| CIで致命的な失敗検出 | マージ禁止、Slack/メール通知、関連Runbookリンクを添付 | 担当者がRunbookに従い修正→再実行 |
| 本番カナリアで指標悪化 | 自動ロールバック、アラート、詳細テストのトリガー | インシデントハンドリング、原因分析 |
Runbook（簡易テンプレート）
| 項目 | 記載例 |
|---|---|
| 発見条件 | API応答時間>5s かつエラー率>5% |
| 初動対応 | 自動ロールバック実行、サービスの一部停止 |
| 詳細確認 | ログ収集、最近のデプロイ差分確認、関連テスト再実行 |
| 復旧手順 | ホットフィックス適用→テスト→段階的再デプロイ |
評価と改善のKPI、チェックリスト
実務で使うべき耐障害指標と導入前の準備チェックリストを示します。
| KPI | 意味 | 目安 |
|---|---|---|
| MTTR（平均復旧時間） | 障害から復旧までにかかる時間 | サービス特性により数分～数時間 |
| フェイル率 | テスト中に再現した致命的障害の割合 | 継続的に低下させる指標 |
| 検出カバレッジ | 設計した失敗モードのカバー率 | まずは50%→段階的に80%超を目標 |
導入前の準備チェックリスト
- テストデータ（合成データ）を用意しておく
- 外部APIのモック/サンドボックスの整備
- テスト環境と本番での権限・リソース分離
- 監視（メトリクス/ログ）にテスト用フックを追加
- Runbookに自動フローと手動手順を明記
まずこれだけやる（短い実行リスト）
- 主要APIに対して「タイムアウト」「5xx返却」「部分欠損」の3ケースをpytestで自動化する
- 合成データから境界値サンプルを10件作る
- CIに軽量スイートを登録してPR時に実行する
- 失敗時の通知とRunbookリンクをCIに組み込む
拡張例（中長期計画）
- 負荷試験と耐障害試験の連携（高負荷下での外部依存性の挙動確認）
- カオステスト（Chaos Engineering）の導入：段階的に本番近傍で実施
- 自動異常検知とテスト生成のループ（異常ログからフェイルケースを抽出し自動生成）
まとめ
耐障害・頑健性テストは単なる技術的要件ではなく、デプロイ・監視・Runbookと一体化して初めて効果を発揮します。この記事では、業務視点で必要なテストワークフロー、合成データの活用、Pythonでの自動化パターン、CI/CD連携、評価指標までを実務的に示しました。まずは「主要機能の3つの代表ケースを自動化してCIに組み込む」ことから始め、段階的にカバレッジを拡大してください。
参考: 第49回（合成データ）は合成ルールの作成、第37回（デプロイ）はCI/CD設計、第46回（Runbook）は運用手順のテンプレートをそれぞれ補完します。次回以降はカオステストの導入計画と自動生成されたフェイルケースのランク付け手法を扱う予定です。