AWS ConfigとTerraformパラメーターシートを突合する単体テスト自動化 — tflint/plan/Config 3層パイプライン

1. この記事について

1-1. 本記事で達成できること

この記事を最後まで読んで手を動かすと、次のものが手元に揃う。

• tflint → terraform plan → AWS Config Advanced Query を組み合わせた 3 層パイプライン
• compare() 関数：plan JSON（期待値）と Config 記録値（現状値）を突合し、差分を DiffRow リストで返す
• pytest によるドリフト検知テストスイート：pytest tests/test_drift.py の 1 コマンドで受入検証が走る
• Sheet2「差分一覧」を自動転記する Excel 出力：第1弾 Excel を受入証跡として完成させる

「構築が終わったあと、設計書（パラメーターシート）と実際の AWS 環境が一致しているか確認したい」——その要望を、pytest 単体テストとして実装する。本番環境への定期実行は行わない。構築完了後の受入検証フェーズで 1 回実行する ユースケースを想定している。

上の図が本記事の全体像だ。tflint による静的検査をゲートに置き、terraform plan JSON で期待値を抽出、AWS Config Advanced Query で現状値を取得、2 者を突合して pytest で差分を報告する。この 3 層が揃うことで、手動チェックに頼ってきたエンタープライズ受入試験が自動化される。

1-2. 本シリーズの全体構成と「最終回」の意味

本シリーズは 2 弾構成で、AWS 案件で繰り返し発生する「パラメーターシート管理」の煩雑さを段階的に解消する。

第1弾は「コードから仕様書を作る（設計フェーズ）」、第2弾は「仕様書と実環境のズレを検知する（検証フェーズ）」。2 弾合わせて 設計 → 検証のサイクルが完成する。

本記事を読み終えたとき、読者はこのサイクルをチームの標準プロセスとして組み込む準備が整っているはずだ。「Excel の作成 → pytest の実行 → NG 行をチームで確認 → TF コード修正 → 再実行」——これが最終形だ。

1-3. 第1弾の振り返り（未読者向け）

第1弾では次のパイプラインを実装した。

Terraform HCL → terraform plan -out=tfplan
  → terraform show -json tfplan → plan.json
  → tf_plan_parser.parse_plan()
  → openpyxl → param-sheet.xlsx（マルチ環境・2段ヘッダ）

核心は parse_plan() 関数で、plan JSON の resource_changes[].change.after を走査してリソースアドレスごとの属性値を抽出する。第1弾では コード記載値を期待値として Excel に書き出す ことが目標だった。

出力した Excel は ENV_SUBCOLS = ['期待値', '現状値', '判定'] の 3 列構成のうち、第1弾時点では「期待値」列のみ埋まった状態だ。第2弾では「現状値」を AWS Config から取得し、「判定」列に OK/NG/UNKNOWN を書き込んで Excel を完成させる。

第1弾の詳細は こちら（第1弾記事） を参照してほしい。本記事では parse_plan() や write_excel() の実装を再掲せず、第1弾のコード資産を import して再利用する前提 で話を進める。

1-4. 関連シリーズとの役割分担

本シリーズは、同じ Terraform を扱う複数人開発シリーズと直交補完の関係にある。

本記事では pytest を 単体テストとして実行する 設計に絞っており、CI/CD パイプラインへの組み込みはスコープ外とする。「定期実行したい」「プルリクエスト時に自動で走らせたい」という要望は 複数人開発シリーズ第2弾（GitHub Actions+OIDC） を参照してほしい。この役割分担は意図的なものだ。受入試験を自動化する価値は、CI 組み込みの前でも十分に大きい。

1-5. 対象読者とペルソナ

本記事は次のような読者を想定している。

主要ペルソナ: エンタープライズ DevOps / インフラエンジニア

• 第1弾で parse_plan() と Excel 出力を実装済みで、「次は実環境と突合したい」と思っている
• 金融・製造・公共系の AWS 案件で構築後の受入試験書作成に数日かけている
• pytest は知っているが AWS テスト自動化での活用経験はない
• AWS Config は名前を聞いたことはあるが、実際に使ったことはない（本記事の §3 で基礎から解説する）

副次ペルソナ: テックリード / QA エンジニア

• インフラ構築の受入証跡として「テスト実行ログ」を残したい
• 手動チェックリストを機械的に実行する仕組みを探している
• 差分管理を Excel ではなく pytest の PASS/FAIL で管理したい

前提知識チェックリスト

以下をすべて満たせば、本記事をスムーズに進められる。

[ ] 第1弾完了: param-sheet-tf-config-excel-generator.md を読んでコードを動かした
[ ] Terraform init / plan / apply の基本操作
[ ] Python 3.11+ の基礎（型ヒント・辞書操作）
[ ] pytest の基礎（test_xxx 関数・assert 文）
[ ] AWS CLI v2 の設定（aws configure 済み・Config Recorder は §3 で有効化手順を案内）

AWS Config 未経験の方へ

AWS Config は「実環境のリソース構成を記録し続けるサービス」だ。使ったことがなくても心配はいらない。§3「AWS Config 基礎」で Recorder の有効化から Advanced Query の実行まで丁寧に解説する。Config Recorder を有効化していない環境でも、§5 までの terraform plan 期待値抽出は動作する。

1-6. 本記事で登場するツール・バージョン一覧

サンプルコードで使う AWS 環境の定数は以下で統一する（実際の環境では適宜置き換えること）。

AWSアカウントID : 123456789012
リージョン: ap-northeast-1
S3 バケット名: myorg-terraform-state
DynamoDB テーブル: terraform-state-lock
ARN 形式  : arn:aws:*:ap-northeast-1:123456789012:*

1-7. 本記事の構成とナビゲーション

11 のセクションで構成する。§1（本記事）→ §2（業務背景）→ §3（Config 基礎）→ §4（tflint）→ §5（期待値抽出）→ §6（現状値取得）→ §7（突合ロジック）→ §8（pytest 構成）→ §9（マルチ環境）→ §10（ハンズオン）→ §11（まとめ）の順に読み進めることを推奨する。

ただし、各セクションの依存関係は以下のとおりで、部分的に先読みも可能だ。

§1（背景） → §2（背景）
§3（Config 基礎）────────────────────────────────→ §6（Config fetch）
§4（tflint）─────────────────────────────────────→ §8（pytest に組込み）
§5（期待値抽出）────────────────────────────────→ §7（突合ロジック）
§6（現状値取得）────────────────────────────────→ §7（突合ロジック）
§7（compare）───────────────────────────────────→ §8（pytest）
§8（pytest）────────────────────────────────────→ §10（統合実行）
§9（マルチ環境）─（独立）──────────────────────→ §10（統合実行）

Config Recorder の有効化（§3-1）に時間がかかる場合、先に §4〜§5 を実施して期待値の抽出だけ先行することも可能だ。

1-8. 読者タイプ別の推奨読み方

読者の状況に応じて、読み進め方を変えることを推奨する。

パターン A: 第1弾を完了済みで、すぐに実装に入りたい

§1（ざっと）→ §2（ざっと）→ §3（Config 有効化手順のみ）→ §4 → §5 → §6 → §7 → §8 → §10
所要時間の目安: 4〜6 時間（ハンズオン含む）

§3 の AWS Config 概念説明（§3-2〜§3-3）は一度読んでから §6 を読むと理解が深まるが、
急いでいる場合は §6 に到達してから必要に応じて戻り読みするスタイルでも追える。

パターン B: Config を初めて使う・じっくり学びたい

§1 → §2 → §3（全読必須）→ §4 → §5 → §6 → §7 → §8 → §9 → §10 → §11
所要時間の目安: 8〜10 時間（ハンズオン含む）

§3 の Config 基礎を丁寧に読むことで、§6 の Advanced Query と §7 の突合ロジックへの
理解がスムーズになる。Config 未経験者はこちらを推奨する。

パターン C: コードだけ欲しい、解説文は不要

• §8 の conftest.py と test_drift.py から読む
• 動作が理解できたら §7 の compare() 関数と §6 の config_fetch() を逆引き
• §5 の load_expected() wrapper で期待値の形式を確認

いずれのパターンでも、§2-4「なぜ CI/CD ではなく単体テストか」は一読することを勧める。
本記事の設計判断の根拠がここに集約されている。

1-9. ディレクトリ構成の予告

本記事で作成するファイルの構成を先に示しておく。第1弾からの継続性がわかるように、
第1弾で作成済みのファイルも含めて示す。

param-sheet-project/
├── environments/
│├── dev/
││├── main.tf
││├── variables.tf
││└── terraform.tfvars
│├── stg/
│└── prod/
├── tf_plan_parser.py ← 第1弾で作成済み（parse_plan() 実装）
├── tf_to_excel.py ← 第1弾で作成済み（write_excel() 実装）
├── param-sheet.xlsx  ← 第1弾で出力済み（期待値列のみ埋まった状態）
│
├── config_fetcher.py ← 第2弾: AWS Config から現状値を取得
├── comparator.py  ← 第2弾: compare() 関数・DiffRow・Verdict
├── tests/
│├── conftest.py← 第2弾: pytest fixture
│└── test_drift.py ← 第2弾: ドリフト検知テスト本体
├── .tflint.hcl ← 第2弾: tflint 設定
└── Makefile ← 第2弾: make drift-check / make drift-all

第1弾から継続して同じリポジトリで作業することを前提としているが、
第2弾のファイルだけを新しいディレクトリに置いて第1弾を pip install 経由で参照する構成も可能だ。
本記事では前者（同一リポジトリ）で解説する。

2. 業務背景: 構築後の突合テストの実務価値

2-1. 「設計時の突合」と「構築後の突合」の違い

第1弾は 設計フェーズの突合を自動化した。Terraform コードが確定した時点で parse_plan() を実行し、コード記載の期待値を Excel に書き出す。設計者がレビューし、顧客に提出するための一次資料を自動生成するユースケースだ。

本記事は 構築後の突合を自動化する。AWS 環境への terraform apply が完了した後、「コードに書いてあった設定が本当に反映されているか」を AWS Config の記録値と照合する。設計レビューではなく、受入試験・結合テストのフェーズに位置する。

この区別は重要だ。第1弾の突合は「コードが正しいか」を確認する。第2弾の突合は「コードどおりに環境が構築されたか」を確認する。ドリフト（TF コードと実環境のズレ）の検知には、第2弾のアプローチが必要になる。

2-2. エンタープライズ案件における受入試験書の現実

金融・製造・公共系の AWS 案件では、構築完了後に顧客への提出物として 受入試験書 を作成するケースが多い。典型的な受入試験書は次の構成を持つ。

【受入試験書（抜粋）】
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
No.  リソース名 属性期待値  実測値  結果
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
001  aws_instance.web[0]  instance_type  t3.medium  t3.medium  OK
002  aws_instance.web[0]  amiami-0abcdef12345  ami-0abcdef12345  OK
003  aws_instance.app[0]  instance_type  t3.larget3.smallNG ← 差分！
...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

この表を手で埋めることが、現在も多くのプロジェクトで行われている。Console の画面をスクリーンショットして期待値と目視比較するか、AWS CLI を手動実行して確認するかのいずれかだ。

本記事の最終目標は、この「実測値」列を AWS Config Advanced Query で自動取得 し、「結果」列の OK/NG 判定を pytest で自動化 することだ。受入試験書は第1弾の Excel がそのまま受入証跡になる。

2-3. 手動突合の工数と「500 チェック問題」

手動突合の工数を具体的に試算してみよう。典型的なエンタープライズ案件で、以下の規模を想定する。

環境数  : 3（dev / stg / prod）
リソース種別数 : 約 15 種（EC2, RDS, ALB, S3, IAM, SG, VPC, Subnet ...）
各リソース種別の属性数: 約 10 属性（instance_type, ami, tags, security_groups ...）
インスタンス数 : 各種別平均 3〜5 個

→ チェック総数: 3 環境 × 15 種別 × 10 属性 × 4 インスタンス = 1,800 チェック

実務ではリソース数をもっと絞っても、10 環境 × 50 項目 = 500 チェック は珍しくない。これを担当エンジニアが 1 件ずつ Console で確認すると、集中力が続く前提でも 2〜3 日を要する。確認ミスや記入漏れも避けられない。

pytest で自動化したあとの所要時間はどうなるか。

tflint 実行 : 約 2 秒
terraform plan 実行  : 約 10 秒（リソース数次第）
AWS Config 取得: 約 5 秒（Advanced Query 1〜2 本）
突合・判定  : 約 1 秒
pytest 全体 : 約 20 秒

500 チェックを 20 秒で完了。再実行も同じ 20 秒。

この差が本記事の価値だ。「500 チェックを 20 秒で実行し、差分があれば NG として即座にレポートする」——これを pytest 単体テストとして実装する。

2-4. なぜ CI/CD ではなく「単体テスト」として位置付けるのか

本記事では pytest を 構築完了後に手動で 1 回実行する単体テスト として位置付けている。GitHub Actions や EventBridge による定期実行はスコープ外だ。この設計判断には理由がある。

理由 1: 受入試験は「イベント駆動」であるべき

定期実行は「いつドリフトが発生したかを検知する運用監視」に向いている。受入試験は「構築完了という特定のイベントの後に 1 回実施する品質ゲート」だ。両者は目的が異なる。本記事は後者に特化する。

理由 2: AWS Config の記録タイミングとコスト

AWS Config は記録頻度に応じてコストが発生する。定期実行の場合、Recorder の有効化期間・記録頻度・Aggregator の設定が運用コストに直結する。受入試験用途であれば、必要なタイミングだけ有効化して終わったら無効化する 運用も現実的だ（§3-1 で扱う）。

理由 3: 「テストの責任範囲」の明確化

CI/CD に組み込んだドリフト検知は「インフラ運用の継続的監視」になる。受入試験として pytest で実行するアプローチは「インフラ構築の品質保証」になる。後者は QA エンジニアや PM が受入証跡として理解しやすく、プロジェクト完了の判定基準に組み込みやすい。

CI 組み込みに発展させたい場合は、複数人開発シリーズ第2弾（GitHub Actions+OIDC） を参照してほしい。本記事で実装した pytest テストスイートはそのまま GitHub Actions の jobs[].steps[].run に置き換えられる。

2-5. 運用フロー Before/After

手動突合から pytest 自動化への変化を図で示す。

図の内容をテキストで補足する。

Before: 手動突合フロー

① terraform apply 完了
 ↓
② 担当者が AWS Console / CLI で各リソースを確認
 ↓
③ Excel の「実測値」列を手入力（500 チェック × 数日）
 ↓
④ 期待値と目視比較 → 差分を NG としてコメント記入
 ↓
⑤ NG 箇所を TF コードで修正 → 再 apply → ②に戻る
 ↓（数サイクル後）
⑥ 受入試験書を顧客に提出

After: pytest 自動化フロー

① terraform apply 完了
 ↓
② make drift-check（tflint + plan + Config fetch + compare + pytest）
 ↓（約 20 秒）
③ pytest が PASS → 受入試験書（Excel）の判定列が自動転記される
pytest が FAIL → NG リストが標準出力に表示される
 ↓（NG があれば）
④ NG 箇所を TF コードで修正 → 再 apply → ②に戻る
 ↓（1〜2 サイクルで解決）
⑤ pytest PASS → 受入試験書を顧客に提出

サイクル 1 回あたりの確認時間が「数日」から「20 秒」になる。修正サイクルを素早く回せるため、手戻りの発見が遅くなるリスクも減る。

2-6. AWS Config を初めて使う方へ

本記事の核心の 1 つは AWS Config Advanced Query だ。Config を使ったことがない読者のために、ここで概要だけ紹介しておく（詳細は §3 で扱う）。

AWS Config とは何か

AWS Config は「AWS リソースの構成変更を継続的に記録し、現時点の構成スナップショットを照会できるサービス」だ。EC2 インスタンスの instance_type、セキュリティグループのルール、RDS の Multi-AZ 設定——こうした属性の変更履歴と現状値を Config が保持している。

本記事での使い方

本記事では Config を「現在の実環境構成を一括取得するデータソース」として使う。具体的には select_resource_config という boto3 API を呼び出して、SQL に近い構文でリソースの構成情報を取得する。

-- Config Advanced Query の例（§6 で詳説）
SELECT
  resourceId,
  resourceType,
  configuration.instanceType,
  configuration.imageId,
  tags
WHERE
  resourceType = 'AWS::EC2::Instance'

このクエリを実行すると、現在の EC2 インスタンス一覧とその属性が JSON で返ってくる。これが「現状値」側のデータになる。

Config Recorder を有効化しないとどうなるか

Config Recorder が有効でない環境では select_resource_config が空を返す。ただし、本記事の §1〜§5（tflint・plan による期待値抽出）は Config なしで動作する。§3 の手順に従って Recorder を有効化してから §6 以降を実施することを推奨するが、学習目的であれば moto による boto3 モック（§8 で扱う）を使ってローカルのみで完結させることも可能だ。

費用感についても正直に書いておく。AWS Config Recorder は記録対象リソース数に応じて課金される。東京リージョン（ap-northeast-1）で 100 リソースを記録した場合、概算で 月額 3〜5 USD 程度だ。受入試験用途であれば、テスト期間だけ有効化して終了後に無効化することでコストを最小化できる（§3-1 で手順を示す）。

2-7. 本記事で扱わない領域の明示

本記事は意図的にスコープを絞っている。以下は本記事では扱わない。

「定期監視まで含めた完全自動化が目標」という読者は、本記事で pytest テストスイートを実装した後、複数人開発シリーズ第2弾で GitHub Actions に組み込む手順を踏むとよい。本記事で作るテストコードはそのまま流用できる。

2-8. 受入試験自動化の副次効果

pytest による受入試験自動化には、工数削減以外にも副次効果がある。

副次効果 1: 修正サイクルが速くなる

手動突合では「確認 → NG 発見 → 報告 → 修正 → 再確認」のサイクルに半日〜1 日かかることがある。pytest であれば 20 秒でサイクルが回る。TF コードを修正して make drift-check を再実行するまでのリードタイムが劇的に縮まる。

副次効果 2: 受入証跡が自動生成される

pytest の -v --tb=short オプションを使うと、各テストケースの PASS/FAIL が一覧で出力される。この出力を受入試験書の付録として添付することが可能だ。さらに本記事では Excel の Sheet2「差分一覧」に自動転記する機能も実装するため、顧客への提出ドキュメントも自動で更新される。

副次効果 3: 「属人的チェック」から脱却できる

手動突合は「チェックリストを誰が担当するか」に依存する。熟練エンジニアが担当すると漏れが少ないが、ジュニアエンジニアが担当すると見落としが増える。pytest による自動チェックは担当者のスキルに依存しない。チーム全員が同じ品質で受入試験を実施できるようになる。

副次効果 4: 差分の「種類」が見えるようになる

手動突合では「差分がある」という事実しかわからない場合が多い。pytest の DiffRow には expected（期待値）と actual（現状値）と verdict（OK/NG/UNKNOWN）が入っている。差分の原因が型不一致なのか、値そのものが異なるのか、リソースが未検出なのかを区別して報告できる。これがトラブルシュートの時間を短縮する。

2-9. 本章のまとめと次章への接続

本章では本記事の「業務背景」を整理した。

• 第1弾の「設計時の突合（Excel 自動生成）」に対し、本記事は「構築後の突合（受入試験自動化）」
• エンタープライズ案件では 10 環境 × 50 項目 = 500 チェックの手動突合が現実に発生している
• pytest を 単体テスト として位置付け、CI/CD 定期実行はスコープ外とする
• AWS Config は「現状値取得のデータソース」として使う（§3 で Recorder 有効化から解説）

次章（§3）では AWS Config の基礎を解説する。Recorder の有効化、Configuration Item の構造、
Advanced Query の SQL 構文まで、Config 未経験者でも追体験できるように丁寧に進める。

2-10. 「受入試験書の自動化」というゴールを再確認する

本章の最後に、本記事のゴールイメージを具体化しておく。

第1弾で生成した Excel（param-sheet.xlsx）は、現時点では「期待値」列だけが埋まった状態だ。

【param-sheet.xlsx — Sheet1（期待値列のみ）】
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
リソース名  属性期待値  現状値 判定
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
aws_instance.web[0] instance_typet3.medium  （空欄）  （空欄）
aws_instance.web[0] ami ami-0abcdef12345  （空欄）  （空欄）
aws_instance.app[0] instance_typet3.large （空欄）  （空欄）
aws_rds_cluster.mainengine aurora-mysql（空欄）  （空欄）
...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

本記事の実装を完了すると、Excel が次の状態になる。

【param-sheet.xlsx — Sheet1（全列埋まり・完成形）】
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
リソース名  属性期待値  現状値  判定
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
aws_instance.web[0] instance_typet3.medium  t3.medium  OK
aws_instance.web[0] ami ami-0abcdef12345  ami-0abcdef12345  OK
aws_instance.app[0] instance_typet3.large t3.smallNG
aws_rds_cluster.mainengine aurora-mysqlaurora-mysqlOK
...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

さらに Sheet2「差分一覧」には NG 行と UNKNOWN 行だけが自動転記される。

【param-sheet.xlsx — Sheet2（差分一覧）】
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
リソース名  属性期待値現状値判定  備考
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
aws_instance.app[0] instance_typet3.large t3.small NG 値が異なる
...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Sheet2 が顧客への指摘書・修正確認書として機能する。この状態の Excel が本記事の 最終成果物 だ。

pytest の出力としては次のようになる。

$ pytest tests/test_drift.py -v
============================== test session starts ==============================
platform darwin -- Python 3.11.9, pytest-8.1.1, pluggy-1.4.0
collected 12 items

tests/test_drift.py::test_drift_ec2_instance_type PASSED [ 16%]
tests/test_drift.py::test_drift_ec2_ami PASSED  [ 25%]
tests/test_drift.py::test_drift_app_instance_type FAILED [ 33%]
tests/test_drift.py::test_drift_rds_engine PASSED  [ 41%]
...

=================================== FAILURES ===================================
_________________ test_drift_app_instance_type _________________
AssertionError: NG diff found:
  resource: aws_instance.app[0]
  attribute: instance_type
  expected: t3.large
  actual:t3.small
  verdict:  NG
============================== 1 failed, 11 passed in 18.42s ==============================

NG が 1 件でも発生すると pytest は exit code 1 を返す。NG がゼロになるまで修正サイクルを回し、
全 PASS になったタイミングで Excel を受入証跡として顧客に提出する。

これが本記事のゴールだ。§3 から実装に入っていこう。

3. AWS Config 基礎 — Recorder / Aggregator / Advanced Query

本章は AWS Config 未経験者向けの入門章である。第1弾では boto3 の select_resource_config を使うコードが登場したが、「AWS Config をまだ有効にしたことがない」という読者のために、Recorder の有効化からコスト感・Aggregator の判断軸・Advanced Query の使い方まで一気に解説する。Config 経験者は §3-4（Advanced Query SQL）から読んでも構わない。

本記事サンプルコードの読み替え規則（本章のみ記載・以降省略）
– AWSアカウントID: 123456789012（自アカウントIDに読み替え）
– リージョン: ap-northeast-1（使用リージョンに読み替え）
– S3バケット: myorg-terraform-state（自環境のバケット名に読み替え）

3-1. AWS Config とは何か

AWS Config は 「いつ・誰が・何を変えたか」を記録し続けるサービスである。EC2 インスタンス・セキュリティグループ・RDS クラスター・IAM ポリシーなど、あらゆるリソースの設定変更履歴が ConfigurationItem（CI） として S3 に保存される。

本記事で Config を使う目的は 「現時点の設定値を Advanced Query で一括取得し、Terraform の plan 期待値と突合する」ことだ。受入検証（構築完了後の単体テスト）として、設計書と実環境の整合性を確認する。

AWS Config の3大機能:

[Recorder] [Rules][Advanced Query]
 │ │  │
設定変更を記録 コンプライアンス評価  SQLで現状値を一括取得
 │ │  │
 └──── S3 (Configuration History) ───────────┘

3-2. Recorder の有効化とコスト感

コスト感（正直に）

AWS Config の課金は主に 「設定項目の記録件数」 で決まる。東京リージョン（ap-northeast-1）の場合:

• 1 CI あたり $0.003（2025年時点、東京リージョン）
• 変更がなければ課金はほぼゼロ
• Advanced Query 自体は無料（Recorder 有効化の費用のみ）

エンタープライズ案件で「Config を試したいが費用が怖い」場合は、dev 環境の単一アカウントから始めるのが現実解だ。

Terraform で Recorder を有効化する

# terraform/config.tf
# Terraform 1.9.x / provider hashicorp/aws ~> 5.0

terraform {
  required_version = "~> 1.9"
  required_providers {
 aws = {
source  = "hashicorp/aws"
version = "~> 5.0"
 }
  }
}

# Config 用 IAM ロール
resource "aws_iam_role" "config_recorder" {
  name = "aws-config-recorder-role"

  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "config.amazonaws.com" }
Action = "sts:AssumeRole"
 }]
  })
}

resource "aws_iam_role_policy_attachment" "config_recorder" {
  role = aws_iam_role.config_recorder.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWS_ConfigRole"
}

# Configuration Recorder — 全リソースタイプを記録
resource "aws_config_configuration_recorder" "main" {
  name  = "default"
  role_arn = aws_iam_role.config_recorder.arn

  recording_group {
 all_supported  = true
 include_global_resource_types = true  # IAM等グローバルリソースも対象
  }
}

# Delivery Channel — 記録先 S3 バケット
resource "aws_config_delivery_channel" "main" {
  name  = "default"
  s3_bucket_name = aws_s3_bucket.config_bucket.id
  s3_key_prefix  = "config"

  snapshot_delivery_properties {
 delivery_frequency = "TwentyFour_Hours"  # 日次スナップショット
  }

  depends_on = [aws_config_configuration_recorder.main]
}

# Recorder のステータスを有効にする
resource "aws_config_configuration_recorder_status" "main" {
  name = aws_config_configuration_recorder.main.name
  is_enabled = true

  depends_on = [aws_config_delivery_channel.main]
}

# Config ログ保存用 S3 バケット
resource "aws_s3_bucket" "config_bucket" {
  bucket = "myorg-terraform-state-config-logs"
}

resource "aws_s3_bucket_policy" "config_bucket" {
  bucket = aws_s3_bucket.config_bucket.id

  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Sid = "AWSConfigBucketPermissionsCheck"
  Effect = "Allow"
  Principal = { Service = "config.amazonaws.com" }
  Action = "s3:GetBucketAcl"
  Resource  = "arn:aws:s3:::myorg-terraform-state-config-logs"
},
{
  Sid = "AWSConfigBucketDelivery"
  Effect = "Allow"
  Principal = { Service = "config.amazonaws.com" }
  Action = "s3:PutObject"
  Resource  = "arn:aws:s3:::myorg-terraform-state-config-logs/config/AWSLogs/123456789012/Config/*"
  Condition = {
 StringEquals = { "s3:x-amz-acl" = "bucket-owner-full-control" }
  }
}
 ]
  })
}

ポイント: aws_config_configuration_recorder_status を忘れると Recorder が無効のまま稼働する。depends_on の順序（recorder → delivery_channel → status）を守ること。

3-3. Aggregator — 必要かどうかの判断軸

Aggregator は 複数アカウント・複数リージョンの CI を1箇所に集約する機能だ。

[dev account]  [stg account]  [prod account]
Recorder Recorder Recorder
│││
└───────────────┼───────────────┘
 ▼
[管理アカウント: Aggregator]
Advanced Query
 （全アカウントを横断検索）

Aggregator が必要なケース:
– マルチアカウント構成（dev/stg/prod が別AWSアカウント）
– クロスリージョンで一括クエリしたい
– AWS Organizations で一元管理したい

Aggregator が不要なケース（本記事の想定）:
– 単一アカウント内の複数環境（dev/stg/prod が同じアカウント内に共存）
– 環境分離を Terraform Workspace または environments/ ディレクトリで実現

本記事では Aggregator を使わない。単一アカウント構成を前提とし、Advanced Query をそのまま叩く。マルチアカウント構成に発展させたい場合は、Aggregator の select_aggregate_resource_config API を使う（§6 で切替方法を示す）。

3-4. Advanced Query — SQL で現状値を取得する

Advanced Query は AWS Config に記録された ConfigurationItem に対して SQL を発行する機能だ。SELECT 文で任意の属性を取得でき、WHERE 句で絞り込みもできる。

図3: AWS Config 全体像

基本 SQL 3本

以下の SQL は aws configservice select-resource-config コマンドまたは boto3 の select_resource_config メソッドで実行できる。

SQL 1: EC2 インスタンス一覧（resourceType 指定）

SELECT
  resourceId,
  resourceName,
  configuration.instanceType,
  configuration.imageId,
  configuration.placement.availabilityZone,
  tags
WHERE
  resourceType = 'AWS::EC2::Instance'

SQL 2: RDS インスタンス一覧（マルチAZ・エンジンバージョン付き）

SELECT
  resourceId,
  resourceName,
  configuration.dBInstanceClass,
  configuration.engine,
  configuration.engineVersion,
  configuration.multiAZ,
  configuration.storageType,
  tags
WHERE
  resourceType = 'AWS::RDS::DBInstance'

SQL 3: タグ検索（Environment=prod のリソース一括取得）

SELECT
  resourceId,
  resourceType,
  resourceName,
  configuration,
  tags
WHERE
  tags.tag.key = 'Environment'
  AND tags.tag.value = 'prod'

AWS CLI での実行例

# EC2 インスタンス一覧を取得
aws configservice select-resource-config \
  --expression "SELECT resourceId, resourceName, configuration.instanceType, tags WHERE resourceType = 'AWS::EC2::Instance'" \
  --region ap-northeast-1 \
  --output json

# 結果をファイルに保存（本記事の pytest では boto3 経由で取得するが、CLIで動作確認可）
aws configservice select-resource-config \
  --expression "SELECT resourceId, configuration.instanceType WHERE resourceType = 'AWS::EC2::Instance'" \
  --region ap-northeast-1 \
  --query 'Results[*]' \
  --output json | jq '.[] | fromjson'

# ページネーション付き（リソースが100件超の場合）
NEXT_TOKEN=""
while true; do
  if [ -z "$NEXT_TOKEN" ]; then
 RESULT=$(aws configservice select-resource-config \
--expression "SELECT resourceId, configuration.instanceType WHERE resourceType = 'AWS::EC2::Instance'" \
--region ap-northeast-1 \
--output json)
  else
 RESULT=$(aws configservice select-resource-config \
--expression "SELECT resourceId, configuration.instanceType WHERE resourceType = 'AWS::EC2::Instance'" \
--region ap-northeast-1 \
--next-token "$NEXT_TOKEN" \
--output json)
  fi

  echo "$RESULT" | jq '.Results[]'
  NEXT_TOKEN=$(echo "$RESULT" | jq -r '.NextToken // empty')
  [ -z "$NEXT_TOKEN" ] && break
done

ConfigurationItem レスポンスの構造

Results 配列の各要素は JSON 文字列（二重エンコード）になっている。fromjson や json.loads() でデコードする。

{
  "resourceId": "i-0a1b2c3d4e5f67890",
  "resourceName": "web-server-01",
  "configuration": {
 "instanceId": "i-0a1b2c3d4e5f67890",
 "instanceType": "t3.medium",
 "imageId": "ami-0abc123def456789a",
 "state": {
"code": 16,
"name": "running"
 },
 "placement": {
"availabilityZone": "ap-northeast-1a",
"tenancy": "default"
 },
 "privateIpAddress": "10.0.1.50",
 "vpcId": "vpc-0123456789abcdef0",
 "subnetId": "subnet-0123456789abcdef0",
 "iamInstanceProfile": {
"arn": "arn:aws:iam::123456789012:instance-profile/ec2-ssm-role",
"id": "AIPA123456789EXAMPLE"
 },
 "tags": [
{ "key": "Name", "value": "web-server-01" },
{ "key": "Environment", "value": "prod" },
{ "key": "ManagedBy", "value": "terraform" }
 ]
  }
}

注意点: configuration フィールドのスキーマは resourceType ごとに完全に異なる。AWS::EC2::Instance では instanceType、AWS::RDS::DBInstance では dBInstanceClass と名前が違う。§6 で resourceType ごとのマッピング表を整備する。

3-5. Advanced Query 実行に必要な IAM 権限

Config の Advanced Query を使うには以下の権限が必要だ。Terraform で CI を走らせる場合は、CodeBuild や GitHub Actions のロールに付与する。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "ConfigAdvancedQuery",
"Effect": "Allow",
"Action": [
  "config:SelectResourceConfig",
  "config:SelectAggregateResourceConfig"
],
"Resource": "*"
 }
  ]
}

SelectResourceConfig が単一アカウント用、SelectAggregateResourceConfig が Aggregator 経由のマルチアカウント用。単一アカウント構成では前者のみで十分。

3-6. Config が「未有効」でもハンズオン前半は進める

本記事の §5（terraform plan 期待値抽出）と本章の SQL 確認まではローカルで進められる。AWS Config Recorder を有効にするのは §6（現状値取得）の直前で構わない。

Config 未有効でも進める部分:
  §3（本章）  — SQL の理解・CLI の動作確認のみ（実リソース不要）
  §4 — tflint は完全ローカル実行
  §5 — terraform plan は Config 不要

Config が必要になる部分:
  §6 — select_resource_config が実際に Config に問い合わせる
  §8 — moto モックで代替可（Config 未有効でも pytest は実行可）

4. tflint による静的検査 — パイプライン入口のゲート

本記事のパイプラインは tflint → terraform plan → AWS Config Advanced Query → 突合 → pytest の順に流れる。その入口で HCL を静的に検査するのが tflint だ。

なぜ tflint を最初に置くか: plan や Config 取得は数秒〜数十秒かかるが、tflint は1〜2秒で完了する。aws_instance_invalid_type のような明らかなミスを最速で弾くことで、後続の高コスト処理を無駄に実行しない。

[tflint]  ──FAIL──→  終了（pytest 不実行）
 │
PASS
 │
 ▼
[terraform plan]  ──→  [Config 取得]  ──→  [突合]  ──→  [pytest]

図4: tflint 実行フロー

4-1. tflint の位置づけ

tflint は Terraform の HCL 静的解析ツールだ。terraform validate が文法チェックに特化しているのに対し、tflint は以下を検出できる:

tflint のルールは2層構造になっている:

[組込みルール] [aws plugin ルール]
  - aws_instance_invalid_type- aws_instance_ebs_optimized
  - aws_db_instance_invalid_*- aws_s3_bucket_public_access_block
  - terraform_deprecated_*- aws_iam_role_invalid_policy_document

4-2. tflint のインストールと初期設定

# macOS
brew install tflint

# Linux (GitHub Releases から)
curl -s https://raw.githubusercontent.com/terraform-linters/tflint/master/install_linux.sh | bash

# バージョン確認（本記事は 0.52.x 以降を前提）
tflint --version

# aws plugin の初期化（.tflint.hcl 作成後に実行）
tflint --init

.tflint.hcl 設定ファイル

# .tflint.hcl
# プロジェクトルートに配置する

config {
  # プラグインのキャッシュディレクトリ
  plugin_dir = "~/.tflint.d/plugins"

  # call_module_type: terraform の module ブロックも検査対象にする
  call_module_type = "local"
}

# Terraform 組込みルール
plugin "terraform" {
  enabled = true
  preset  = "recommended"
}

# AWS プロバイダー固有ルール
plugin "aws" {
  enabled = true
  version = "0.35.0"
  source  = "github.com/terraform-linters/tflint-ruleset-aws"
}

# 有効化するルールの例（デフォルト: preset=recommended で主要ルールは有効）
rule "aws_instance_invalid_type" {
  enabled = true
}

rule "aws_db_instance_invalid_db_instance_class" {
  enabled = true
}

# 意図的に無効化するルール（プロジェクト固有の事情がある場合）
# rule "terraform_required_version" {
#enabled = false
# }

4-3. tflint の実行コマンド

# 基本実行（カレントディレクトリの .tf を検査）
tflint

# 再帰的に全サブディレクトリも検査
tflint --recursive

# 特定ディレクトリを指定
tflint --chdir=terraform/environments/prod

# フォーマット出力を JSON に変更（CI での結果集約用）
tflint --format=json

# 本記事のプロジェクト構成での実行例
tflint \
  --chdir=terraform/ \
  --recursive \
  --format=compact

# エラー時に詳細表示
tflint --loglevel=debug

実行結果例

$ tflint --chdir=terraform/ --recursive

terraform/main.tf
  Error: "t3.medum" is an invalid value as instance_type (aws_instance_invalid_type)

  on terraform/main.tf line 12:
  12:instance_type = "t3.medum"  # タイポ

  1 issue(s) found

tflint が1件でも Error を出力した場合、終了コードは 1 になる。これを使って後続処理を制御する。

4-4. パイプラインゲートの実装

シェルスクリプトによるゲート

#!/usr/bin/env bash
# scripts/run_drift_check.sh
# tflint FAIL なら後続の pytest を実行しない

set -euo pipefail

TF_DIR="${1:-terraform}"

echo "=== [1/4] tflint 静的検査 ==="
if ! tflint --chdir="$TF_DIR" --recursive; then
  echo "ERROR: tflint failed. Aborting drift check." >&2
  exit 1
fi
echo "tflint: PASS"

echo "=== [2/4] terraform plan ==="
cd "$TF_DIR"
terraform init -input=false > /dev/null
terraform plan -out=tfplan -input=false
terraform show -json tfplan > ../tfplan.json
cd -
echo "terraform plan: PASS"

echo "=== [3/4] pytest（drift テスト）==="
python -m pytest tests/test_drift.py -v

echo "=== [4/4] 完了 ==="

pytest fixture での tflint 先行実行

§8 でテスト全体を組み立てる際、pytest の autouse fixture として tflint を先行実行することで、テストセッション開始時に自動でゲートを通過させる。ここではその骨格を示す（§8 で詳細実装）:

# tests/conftest.py（§8 で完全実装）
import subprocess
import pytest

@pytest.fixture(scope="session", autouse=True)
def tflint_gate(tf_dir: str = "terraform") -> None:
 """tflint が PASS しなければテストセッション全体をスキップする。"""
 result = subprocess.run(
  ["tflint", "--chdir", tf_dir, "--recursive"],
  capture_output=True,
  text=True,
 )
 if result.returncode != 0:
  pytest.skip(
reason=f"tflint failed — fix HCL before running drift tests.\n{result.stdout}"
  )

SKIP = FAIL ルール（CLAUDE.md 準拠）: pytest.skip を使う場合、reason になぜスキップするかを必ず明示する。tflint FAIL はスキップではなく「前提条件未達のテスト中断」であり、CI 結果では SKIP ではなく ERROR として扱うべき場面もある。詳細は §8 で扱う。

4-5. pre-commit 連携とスコープの境界

tflint を git commit のタイミングで自動実行する pre-commit フックが一般的だ。本記事ではスコープ外として扱い、設定例のみを示す。

# .pre-commit-config.yaml（参考: CI 自動化は第2弾（GitHub Actions+OIDC）に委譲）
# 本記事のスコープは「受入検証（手動実行）」であり、CI/CD 組込みは扱わない。
# pre-commit フックの完全実装は以下を参照:
#第2弾 GitHub Actions+OIDC（WP ID:1259）
repos:
  - repo: https://github.com/terraform-linters/tflint
 rev: v0.52.0
 hooks:
- id: tflint
  name: tflint
  entry: tflint
  language: golang
  files: \.tf$
  args:
 - "--recursive"

本記事の位置づけ: 本記事が対象とするのは「構築完了後の手動受入検証」だ。定期実行・CI 自動化は第2弾（GitHub Actions+OIDC）へ委譲する。tflint を pre-commit で自動化したい場合は、第2弾: GitHub Actions+OIDC を参照。

4-6. tflint 実行結果の読み方と主要ルール一覧

tflint の出力レベル:
  Error— 必ず修正（終了コード 1）
  Warning — 推奨修正（終了コード 0）
  Notice  — 情報提供（終了コード 0）

本記事のゲート判定:
  Error が1件でもあれば FAIL → pytest 不実行
  Warning / Notice のみなら PASS → pytest 実行

よく遭遇する AWS plugin ルールとその意味:

4-7. まとめ: §3〜§4 で整備したパイプライン入口

本章（§3・§4）で以下が揃った:

[§3] AWS Config
  ✓ Recorder を Terraform で有効化
  ✓ Advanced Query SQL 3本（EC2 / RDS / タグ検索）
  ✓ ConfigurationItem の JSON 構造

[§4] tflint
  ✓ .tflint.hcl 設定（aws plugin 有効化）
  ✓ パイプラインゲート実装
  ✓ pytest fixture での先行実行骨格

次章 §5 では、第1弾の parse_plan() を再利用して
terraform plan JSON から期待値を抽出する。

§5 以降のクリティカルパス:
– §5（期待値抽出）+ §6（Config 現状値取得）が完成すると P4・P5 の並行着手が可能になる
– §7（突合ロジック）の DiffRow dataclass 定義が §8（pytest）の事前条件
– 事前定義契約（§6-3）の Verdict / DiffRow はすでに確定済み

Section 5. terraform plan JSON からの期待値抽出（第1弾スキル再利用）

Section 4 で tflint による静的検査ゲートが整いました。次のステップは「terraform plan JSON から期待値を抽出する」ことです。この期待値が突合テストの比較元（expected）になります。

本セクションの実装は、第1弾記事（「Terraformコードから AWS パラメーターシート（Excel）を自動生成する」）で完成させた parse_plan() スキルをそのまま再利用します。第1弾未読の方向けに概要を補足したうえで、第2弾独自のラッパー load_expected() を実装します。

5-1. 第1弾スキルの概要（未読者向け補足）

第1弾では scripts/tf_plan_parser/ モジュールとして以下の関数を実装しました。

# 第1弾 scripts/tf_plan_parser/parser.py（概要のみ）
from pathlib import Path
from typing import Any
from tf_plan_parser import ParserConfig, parse_plan

# parse_plan() の戻り値型
# {service_type: {resource_address: {attr: value, ...}}}
# 例:
# {
#"aws_instance": {
#  "aws_instance.web": {"instance_type": "t3.medium", "ami": "ami-0abc1234"}
#}
# }

第1弾の 2段フィルタ方式（要点）:

plan.json の 2 領域を照合
  ├── configuration.root_module.resources[].expressions
  │  → Terraform コードに書かれた属性名を code_keys 集合として収集
  └── planned_values.root_module.resources[].values
  → code_keys に含まれる属性のみ取得（AWS デフォルト値・ARN は自動除外）

この仕組みにより、instance_id や arn のような「AWS が実行時に払い出す値」は出力されず、instance_type や ami のような「Terraform コードに書いた値」だけが抽出されます。

詳細な実装（ParserConfig dataclass・extract_code_keys()・should_exclude()）は 第1弾 §5 を参照してください。本記事では parse_plan() を公開 API として扱い、完全な再掲は省略します。

5-2. load_expected() — 第2弾専用ラッパー

第2弾では突合テストのために「address → {attr: value} のフラットな辞書」が必要です。第1弾の戻り値（{service: {address: {attr: value}}}）をサービス層なしのフラット形式に変換する薄いラッパーを用意します。

# scripts/drift_checker/expected.py
from __future__ import annotations

import warnings
from pathlib import Path
from typing import Any

from tf_plan_parser import ParserConfig, parse_plan


# 第2弾で使うフラット形式の型エイリアス
ExpectedParams = dict[str, dict[str, Any]]
# キー: resource_address（例: "aws_instance.web"）
# 値: {attr: value, ...}（第1弾 parse_plan() の戻り値から service 層を除去した形）


def load_expected(
 plan_path: Path,
 config: ParserConfig | None = None,
) -> ExpectedParams:
 """plan.json から期待値辞書を読み込む。

 第1弾 parse_plan() のサービス別入れ子構造を
 {resource_address: {attr: value}} のフラット形式に変換する。

 Args:
  plan_path: terraform show -json tfplan の出力ファイル
  config: ParserConfig（None ならデフォルト設定）

 Returns:
  ExpectedParams — {resource_address: {attr: value, ...}}

 Raises:
  FileNotFoundError: plan_path が存在しない
  ValueError: plan.json のパースに失敗
 """
 nested = parse_plan(plan_path, config=config)
 flat: ExpectedParams = {}
 for _service, resources in nested.items():
  for address, attrs in resources.items():
if address in flat:
 warnings.warn(
  f"address 重複を検出: {address}。後勝ちで上書きします。",
  UserWarning,
  stacklevel=2,
 )
flat[address] = attrs
 return flat

使用例:

# plan.json の生成
cd environments/dev
terraform plan -out tfplan
terraform show -json tfplan > plan.json

from pathlib import Path
from scripts.drift_checker.expected import load_expected

expected = load_expected(Path("environments/dev/plan.json"))
# 例:
# {
#"aws_instance.web": {"instance_type": "t3.medium", "ami": "ami-0abc1234", "tags": {...}},
#"aws_vpc.main": {"cidr_block": "10.0.0.0/16"},
# }

5-3. 型正規化ユーティリティ

plan.json の expressions は値の形式が属性によって異なります。突合時の型不一致を防ぐために、expressions から取得したキー情報を正規化するユーティリティを用意します。

# scripts/drift_checker/normalize.py
from __future__ import annotations

from typing import Any


def normalize_plan_value(value: Any) -> Any:
 """plan.json の planned_values 由来の値を正規化する。

 主な変換:
 - None（known_after_apply）はそのまま保持
 - bool は bool のまま（Terraform は true/false を正確に返す）
 - 数値文字列は int/float に変換しない（AWS Config 側との比較は compare() が担当）
 """
 if value is None:
  return None
 if isinstance(value, bool):
  return value
 if isinstance(value, (int, float)):
  return value
 if isinstance(value, str):
  return value.strip()
 if isinstance(value, list):
  return [normalize_plan_value(v) for v in value]
 if isinstance(value, dict):
  return {k: normalize_plan_value(v) for k, v in value.items()}
 return value


def normalize_expression_type(expr: dict) -> str | None:
 """expressions エントリから型ヒントを抽出する。

 terraform plan JSON の expressions 構造:
{"constant_value": "t3.medium"}  → 定数参照
{"references": ["var.instance_type"]}  → 変数参照

 戻り値: "constant" / "reference" / "complex" / None
 """
 if "constant_value" in expr:
  return "constant"
 if "references" in expr:
  return "reference"
 if isinstance(expr, dict) and any(isinstance(v, dict) for v in expr.values()):
  return "complex"
 return None

5-4. count / for_each 展開の address 解決

第1弾と同じく、count / for_each を使ったリソースは plan.json 内で aws_instance.web[0]、aws_instance.web["prod"] のようなインデックス付き address になります。突合時に「設計 address」と「Config 側 resource ID」を紐付けるため、インデックスを保持したまま扱います。

# scripts/drift_checker/address.py
from __future__ import annotations

import re


def strip_index(address: str) -> str:
 """count/for_each のインデックスを除いたベース address を返す。

 例:
  "aws_instance.web[0]" → "aws_instance.web"
  'aws_instance.web["a"]' → "aws_instance.web"
  "module.vpc.aws_vpc.main" → "module.vpc.aws_vpc.main"（変化なし）
 """
 return re.sub(r'\[.*?\]$', '', address)


def address_to_resource_type(address: str) -> str:
 """TF address から resource_type を抽出する。

 例:
  "aws_instance.web"  → "aws_instance"
  "module.vpc.aws_vpc.main" → "aws_vpc"
 """
 parts = address.split(".")
 # module.xxx が先頭に並ぶ場合を考慮して最後から2番目を取る
 for i in range(len(parts) - 1, -1, -1):
  if parts[i].startswith("aws_") or parts[i].startswith("google_") or parts[i].startswith("azurerm_"):
return parts[i]
 return parts[-2] if len(parts) >= 2 else address


def tf_type_to_config_type(resource_type: str) -> str | None:
 """Terraform resource type を AWS Config resourceType に変換する。

 例:
  "aws_instance" → "AWS::EC2::Instance"
  "aws_vpc"→ "AWS::EC2::VPC"
  "aws_db_instance" → "AWS::RDS::DBInstance"
  (未知の type)  → None
 """
 _map = {
  "aws_instance": "AWS::EC2::Instance",
  "aws_vpc":"AWS::EC2::VPC",
  "aws_subnet":"AWS::EC2::Subnet",
  "aws_security_group": "AWS::EC2::SecurityGroup",
  "aws_db_instance": "AWS::RDS::DBInstance",
  "aws_lb": "AWS::ElasticLoadBalancingV2::LoadBalancer",
  "aws_lambda_function":"AWS::Lambda::Function",
  "aws_api_gateway_rest_api": "AWS::ApiGateway::RestApi",
  "aws_dynamodb_table": "AWS::DynamoDB::Table",
  "aws_sns_topic":"AWS::SNS::Topic",
  "aws_sqs_queue":"AWS::SQS::Queue",
  "aws_s3_bucket":"AWS::S3::Bucket",
 }
 return _map.get(resource_type)

Section 5 まとめ

Section 6 では、この load_expected() の expected と突合する actual（AWS Config の現状値）を取得します。

Section 6. AWS Config Advanced Query による現状値取得

突合テストの「現状値（actual）」を AWS Config から取得します。AWS Config は select_resource_config API（Advanced Query）を通じて SQL ライクなクエリで Configuration Item を検索できます。このセクションではクエリ構築・ページネーション・スキーマ差異対応を含む config_fetch() 関数を完全実装します。

6-1. AWS Config Advanced Query の仕組み

AWS Config Advanced Query の処理フロー

┌──────────────────────────────────────────────────────────────┐
│ client.select_resource_config(Expression=sql, Limit=100)  │
│  │
│  SQL 例:  │
│  SELECT resourceId, configuration │
│  WHERE resourceType = 'AWS::EC2::Instance' │
│  AND configuration.instanceType IS NOT NULL│
│  │
│  ↓ NextToken でページネーション│
│  │
│  ConfigurationItem[].configuration（JSON文字列） │
│ → json.loads() で dict に変換  │
│ → 属性を正規化  │
│ → {resource_id: {attr: value}} に格納│
└──────────────────────────────────────────────────────────────┘

重要な注意点: configuration フィールドは resourceType ごとにスキーマが異なります。AWS::EC2::Instance では instanceType ですが、AWS::RDS::DBInstance では dbInstanceClass です。このスキーマ差異を吸収するために、resourceType 別の属性マッピングを定義します。

6-2. DiffRow / Verdict dataclass 契約（Section 7 との整合）

Section 7 の突合ロジックは、config_fetch() の戻り値スキーマに依存します。spec §6-3 で定義された契約を先出しします。config_fetch() の戻り値はこの契約と整合させます。

# scripts/drift_checker/models.py
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class Verdict(Enum):
 """突合判定の結果。

 OK:expected == actual（正規化後）
 NG:expected != actual（差分あり）
 UNKNOWN: actual が取得できなかった（Config 未記録 / 権限不足 等）
 """
 OK = "OK"
 NG = "NG"
 UNKNOWN = "UNKNOWN"


@dataclass(frozen=True)
class DiffRow:
 """突合結果の1行。Section 7 の compare() が生成し Section 8 の pytest が検証する。

 resource_address: TF address（例: aws_instance.web[0]）
 resource_type: AWS Config の resourceType（例: AWS::EC2::Instance）
 resource_id:Config 側 ID（未検出時 None）
 attribute:  突合した属性名（例: instance_type）
 expected:plan 由来値（正規化後）
 actual:  Config 由来値（正規化後）
 verdict: 判定結果
 note: 型不一致・欠測理由などの補足
 """
 resource_address: str
 resource_type: str
 resource_id: str | None
 attribute: str
 expected: Any
 actual: Any
 verdict: Verdict
 note: str = ""

config_fetch() は dict[str, dict[str, Any]] を返します:

# 戻り値の型定義（型エイリアス）
# キー: AWS Config resource_id（例: "i-0abc1234567890123"）
# 値: {terraform_attr_name: normalized_value, ...}
ActualParams = dict[str, dict[str, Any]]

6-3. SQL 構築ヘルパー

# scripts/drift_checker/config_query.py
from __future__ import annotations

_SELECT_TEMPLATE = (
 "SELECT resourceId, resourceName, arn, tags, configuration "
 "WHERE resourceType = '{resource_type}'"
)

_TYPE_FILTERS: dict[str, str] = {
 "AWS::EC2::Instance": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::EC2::Instance'"
 ),
 "AWS::EC2::VPC": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::EC2::VPC'"
 ),
 "AWS::EC2::Subnet": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::EC2::Subnet'"
 ),
 "AWS::EC2::SecurityGroup": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::EC2::SecurityGroup'"
 ),
 "AWS::RDS::DBInstance": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::RDS::DBInstance'"
 ),
 "AWS::ElasticLoadBalancingV2::LoadBalancer": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::ElasticLoadBalancingV2::LoadBalancer'"
 ),
 "AWS::Lambda::Function": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::Lambda::Function'"
 ),
 "AWS::ApiGateway::RestApi": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::ApiGateway::RestApi'"
 ),
 "AWS::DynamoDB::Table": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::DynamoDB::Table'"
 ),
 "AWS::SNS::Topic": (
  "SELECT resourceId, resourceName, arn, tags, configuration "
  "WHERE resourceType = 'AWS::SNS::Topic'"
 ),
}


def build_query(resource_type: str) -> str:
 """resourceType に対応する Advanced Query SQL を返す。

 Args:
  resource_type: AWS Config resourceType（例: "AWS::EC2::Instance"）

 Returns:
  SQL 文字列

 Raises:
  ValueError: 未対応の resource_type
 """
 sql = _TYPE_FILTERS.get(resource_type)
 if sql is None:
  supported = ", ".join(sorted(_TYPE_FILTERS.keys()))
  raise ValueError(
f"未対応の resourceType: {resource_type}\n"
f"対応済み: {supported}"
  )
 return sql


def supported_config_types() -> list[str]:
 """対応済み AWS Config resourceType の一覧を返す。"""
 return sorted(_TYPE_FILTERS.keys())

6-4. config_fetch() — 本体実装

# scripts/drift_checker/config_fetch.py
from __future__ import annotations

import json
import logging
from typing import Any

import boto3
from botocore.exceptions import ClientError

from .config_query import build_query
from .models import ActualParams

logger = logging.getLogger(__name__)

_PAGE_LIMIT = 100


def config_fetch(
 resource_types: list[str],
 region: str = "ap-northeast-1",
 session: boto3.Session | None = None,
) -> ActualParams:
 """AWS Config Advanced Query で指定 resourceType の現状値を取得する。

 Args:
  resource_types: 取得対象の AWS Config resourceType リスト
（例: ["AWS::EC2::Instance", "AWS::RDS::DBInstance"]）
  region: AWS リージョン（デフォルト: ap-northeast-1）
  session: 既存の boto3.Session（None なら新規作成）

 Returns:
  ActualParams — {resource_id: {attr: normalized_value, ...}}

  Config 側の resource_id をキーとした辞書。
  未取得・権限不足のリソースは結果に含まれない。

 Raises:
  botocore.exceptions.ClientError:
- NoSuchConfigurationAggregatorException  → Aggregator 未設定
- AccessDenied / UnauthorizedAccess → IAM 権限不足
 """
 if session is None:
  session = boto3.Session(region_name=region)
 client = session.client("config", region_name=region)

 result: ActualParams = {}

 for resource_type in resource_types:
  try:
sql = build_query(resource_type)
  except ValueError as exc:
logger.warning("スキップ: %s", exc)
continue

  items = _paginate_query(client, sql, resource_type)
  for item in items:
resource_id = item.get("resourceId", "")
if not resource_id:
 continue
attrs = _extract_attrs(item, resource_type)
result[resource_id] = attrs

 return result


def _paginate_query(
 client: Any,
 sql: str,
 resource_type: str,
) -> list[dict]:
 """select_resource_config のページネーションを処理する。"""
 items: list[dict] = []
 next_token: str | None = None

 while True:
  kwargs: dict[str, Any] = {"Expression": sql, "Limit": _PAGE_LIMIT}
  if next_token:
kwargs["NextToken"] = next_token

  try:
response = client.select_resource_config(**kwargs)
  except ClientError as exc:
error_code = exc.response["Error"]["Code"]
if error_code == "InvalidLimitException":
 logger.warning(
  "InvalidLimitException: Limit=%d が大きすぎます。25 に縮小して再試行します。",
  _PAGE_LIMIT,
 )
 kwargs["Limit"] = 25
 response = client.select_resource_config(**kwargs)
else:
 raise

  for raw in response.get("Results", []):
try:
 items.append(json.loads(raw))
except json.JSONDecodeError:
 logger.warning("ConfigurationItem の JSON パース失敗: %r", raw[:200])

  next_token = response.get("NextToken")
  if not next_token:
break

 logger.info("resourceType=%s: %d 件取得", resource_type, len(items))
 return items


def _extract_attrs(item: dict, resource_type: str) -> dict[str, Any]:
 """ConfigurationItem から Terraform の属性名に対応する値を抽出・正規化する。

 Args:
  item: select_resource_config の Results 1エントリ（json.loads 後）
  resource_type: AWS Config resourceType

 Returns:
  {terraform_attr_name: normalized_value}
 """
 configuration: dict = {}
 raw_conf = item.get("configuration")
 if isinstance(raw_conf, str):
  try:
configuration = json.loads(raw_conf)
  except json.JSONDecodeError:
pass
 elif isinstance(raw_conf, dict):
  configuration = raw_conf

 # resourceType 別のフィールドマッピングを適用
 mapper = _get_attr_mapper(resource_type)
 attrs: dict[str, Any] = {}
 for tf_attr, conf_path in mapper.items():
  value = _deep_get(configuration, conf_path)
  if value is not None:
attrs[tf_attr] = value

 # tags は共通で取得
 raw_tags = item.get("tags", [])
 if raw_tags:
  attrs["tags"] = _normalize_tags(raw_tags)

 return attrs

6-5. resourceType 別 属性マッピング

ConfigurationItem の configuration フィールドのキー名は Terraform の属性名と一致しません。resourceType ごとに変換マッピングを定義します。

# scripts/drift_checker/config_fetch.py（続き）

def _get_attr_mapper(resource_type: str) -> dict[str, str]:
 """TF 属性名 → Config configuration のドット区切りパスを返す。

 戻り値:
  {terraform_attr_name: "configuration.nested.key"（ドット区切り）}
 """
 _MAPPERS: dict[str, dict[str, str]] = {
  "AWS::EC2::Instance": {
"instance_type":  "instanceType",
"ami":"imageId",
"key_name": "keyName",
"monitoring":  "monitoring.state",
"ebs_optimized":  "ebsOptimized",
"disable_api_termination": "stateTransitionReason",
"availability_zone": "placement.availabilityZone",
  },
  "AWS::EC2::VPC": {
"cidr_block":"cidrBlock",
"enable_dns_support": "enableDnsSupport",
"enable_dns_hostnames":  "enableDnsHostnames",
"instance_tenancy":"instanceTenancy",
  },
  "AWS::EC2::Subnet": {
"cidr_block":"cidrBlock",
"availability_zone":  "availabilityZone",
"map_public_ip_on_launch":  "mapPublicIpOnLaunch",
  },
  "AWS::EC2::SecurityGroup": {
"name":  "groupName",
"description": "description",
  },
  "AWS::RDS::DBInstance": {
"instance_class":  "dbInstanceClass",
"engine": "engine",
"engine_version":  "engineVersion",
"multi_az":  "multiAZ",
"storage_type": "storageType",
"allocated_storage":  "allocatedStorage",
"backup_retention_period":  "backupRetentionPeriod",
  },
  "AWS::ElasticLoadBalancingV2::LoadBalancer": {
"name": "loadBalancerName",
"scheme":  "scheme",
"load_balancer_type": "type",
"ip_address_type":  "ipAddressType",
  },
  "AWS::Lambda::Function": {
"function_name": "functionName",
"runtime": "runtime",
"handler": "handler",
"memory_size":"memorySize",
"timeout": "timeout",
"architectures": "architectures",
  },
  "AWS::DynamoDB::Table": {
"name": "tableName",
"billing_mode":  "billingModeSummary.billingMode",
"hash_key":"keySchema[?keyType=='HASH'].attributeName | [0]",
  },
  "AWS::SNS::Topic": {
"name": "topicArn",  # SNS は ARN から名前を取得
  },
  "AWS::SQS::Queue": {
"name": "queueUrl",
"visibility_timeout_seconds": "attributes.VisibilityTimeout",
  },
 }
 return _MAPPERS.get(resource_type, {})


def _deep_get(d: dict, path: str) -> Any:
 """ドット区切りパスで辞書を掘る。

 例: _deep_get({"a": {"b": "c"}}, "a.b") → "c"
 パスが存在しない場合は None を返す。
 """
 parts = path.split(".")
 current: Any = d
 for part in parts:
  if not isinstance(current, dict):
return None
  current = current.get(part)
 return current


def _normalize_tags(raw_tags: list | dict) -> dict[str, str]:
 """Config の tags 形式を TF の {Key: Value} 辞書に正規化する。

 Config tags は [{"key": "Name", "value": "dev-web"}, ...] 形式が多い。
 """
 if isinstance(raw_tags, dict):
  return raw_tags  # 既に辞書形式
 result: dict[str, str] = {}
 for tag in raw_tags:
  if isinstance(tag, dict):
k = tag.get("key") or tag.get("Key", "")
v = tag.get("value") or tag.get("Value", "")
if k:
 result[k] = v
 return result

6-6. plan 側との引き当て戦略

Config は resource_id（例: i-0abc1234567890123）をキーとしますが、Terraform address（aws_instance.web）とは直接対応しません。引き当てには tags.Name と resource_type の組み合わせを使います。

# scripts/drift_checker/matcher.py
from __future__ import annotations

from typing import Any

from .address import tf_type_to_config_type


def match_expected_to_actual(
 expected: dict[str, dict[str, Any]],
 actual: dict[str, dict[str, Any]],
) -> dict[str, str | None]:
 """TF address → Config resource_id の対応表を構築する。

 マッチ戦略（優先順）:
1. tags.Name が TF address のリソース名部分と一致
2. 同一 resourceType で resource 数が1つの場合は自動マッチ
3. マッチ不可の場合は None（UNKNOWN 扱い）

 Args:
  expected: load_expected() の戻り値
  actual: config_fetch() の戻り値

 Returns:
  {tf_address: config_resource_id | None}
 """
 mapping: dict[str, str | None] = {}

 # resource_type ごとに候補を絞る
 for tf_address, attrs in expected.items():
  parts = tf_address.split(".")
  tf_resource_name = parts[-1].split("[")[0]  # "web" from "aws_instance.web[0]"
  config_type = tf_type_to_config_type(_extract_type(tf_address))

  candidates = [
(rid, rvals)
for rid, rvals in actual.items()
# tags.Name で突合
if rvals.get("tags", {}).get("Name") == tf_resource_name
  ]

  if len(candidates) == 1:
mapping[tf_address] = candidates[0][0]
  elif not candidates:
mapping[tf_address] = None
  else:
# 複数候補 → インデックスで順序付け試行
mapping[tf_address] = None

 return mapping


def _extract_type(address: str) -> str:
 """TF address から resource_type を抽出（シンプル版）。"""
 parts = address.replace("[", " ").split(".")
 for p in reversed(parts):
  p = p.strip().split(" ")[0]
  if p.startswith("aws_"):
return p
 return ""

6-7. エラーハンドリング

# scripts/drift_checker/config_fetch.py（エラーハンドリング追記）

def config_fetch_safe(
 resource_types: list[str],
 region: str = "ap-northeast-1",
) -> tuple[dict[str, dict], list[str]]:
 """config_fetch() の安全版。エラーを例外ではなくエラーリストで返す。

 Returns:
  (result_dict, error_messages)
  result_dict: 取得成功した {resource_id: {attr: value}}
  error_messages: 権限不足・未設定等の警告リスト
 """
 import boto3
 from botocore.exceptions import ClientError

 errors: list[str] = []
 try:
  result = config_fetch(resource_types, region=region)
  return result, errors
 except ClientError as exc:
  code = exc.response["Error"]["Code"]
  msg = exc.response["Error"]["Message"]
  if "AccessDenied" in code or "UnauthorizedAccess" in code:
errors.append(
 f"IAM 権限不足: config:SelectResourceConfig が許可されていません。"
 f" ({code}: {msg})"
)
  elif "NoSuchConfigurationRecorder" in code:
errors.append(
 f"AWS Config Recorder が有効ではありません。"
 f" マネジメントコンソールで Config を有効化してください。({code})"
)
  else:
errors.append(f"AWS Config API エラー: {code} — {msg}")
  return {}, errors

よくあるエラーと対処:

エラー種別 | 原因| 対処
----------------------------------|--------------------|-----------------------------------------
AccessDeniedException  | IAM 権限不足 | config:SelectResourceConfig を policy に追加
NoSuchConfigurationRecorder  | Recorder 未有効 | Config コンソールで Recorder を有効化
InvalidLimitException  | Limit 値が大きい| 自動的に 25 に縮小して再試行（実装済）
ResourceNotRecordedException | 対象リソース未記録 | Recorder のリソースタイプ設定を確認
ThrottlingException | API 呼び出し過多| exponential backoff で再試行

6-8. 動作確認

# 必要な IAM ポリシー最小セット
# config:SelectResourceConfig に対して Allow 設定が必要
aws config select-resource-config \
  --expression "SELECT resourceId, configuration WHERE resourceType = 'AWS::EC2::Instance'" \
  --region ap-northeast-1 \
  --output json | python3 -m json.tool | head -40

期待レスポンス（抜粋）:

{
  "Results": [
 "{\"resourceId\":\"i-0abc1234567890123\",\"configuration\":{\"instanceType\":\"t3.medium\",\"imageId\":\"ami-0abc1234\",\"keyName\":\"dev-key\",\"monitoring\":{\"state\":\"disabled\"}}}",
 "{\"resourceId\":\"i-0def9876543210abc\",\"configuration\":{\"instanceType\":\"t3.micro\",\"imageId\":\"ami-0abc1234\"}}"
  ],
  "QueryInfo": {
 "SelectFields": [
{"Name": "resourceId"},
{"Name": "configuration"}
 ]
  }
}

Python からの確認:

from scripts.drift_checker.config_fetch import config_fetch

actual = config_fetch(["AWS::EC2::Instance"], region="ap-northeast-1")
# 例:
# {
#"i-0abc1234567890123": {
#  "instance_type": "t3.medium",
#  "ami": "ami-0abc1234",
#  "tags": {"Name": "web", "Env": "dev"}
#}
# }
for resource_id, attrs in actual.items():
 print(f"{resource_id}: {attrs}")

Section 5-6 まとめ

Section 7 では load_expected() の expected と config_fetch() の actual を受け取り、compare(expected, actual) -> list[DiffRow] で差分判定・型正規化・Verdict 計算を実装します。

7. 突合ロジック — 型正規化・差分判定・Verdict 設計

Section 6 で取得した AWS Config の現状値（actual）と、
Section 5 で terraform plan JSON から抽出した期待値（expected）を突き合わせるのが本セクションの役割です。
突合ロジックが本記事の核心部であり、ここを読めば「単体テスト」としてのパラメーターシートがどのように機能するかがわかります。

突合を単純な等値比較にしてしまうと、型の揺らぎ（"10" vs 10）や
表記の揺らぎ（"True" vs true）で大量の偽陽性（false positive）が発生します。
信頼できるテストを書くには、比較の前に型を正規化する層が必要です。

本セクションでは次の3つを実装します:

1. Verdict enum — 差分判定結果の分類
2. DiffRow dataclass — 属性単位の差分情報を保持する不変オブジェクト
3. normalize_value() / compare() — 型正規化と突合のコア関数

7-1. 設計思想: 型正規化を挟む3段構成

突合は次の3段で構成します。この分離が保守性と可読性の鍵です。

┌─────────────┐ ┌────────────────────┐ ┌────────────────┐
│  load_expected│ │  normalize_value() │ │compare() │
│  (§5 で実装)  │→→→│  型正規化ユーティリ│→→→│  突合メイン関数│
│  plan.json  │ │  ティ  │ │  list[DiffRow] │
└─────────────┘ └────────────────────┘ └────────────────┘
┌─────────────┐  ↑
│  config_fetch│  │ 同じ正規化を
│  (§6 で実装)  │→→→→→→→→→─┘ expected/actual 両方に適用
│  boto3 API  │
└─────────────┘

比較直前に両辺を同じ正規化関数に通すことで、「TF コードで書いた値」と「Config が返す値」の
表記差異を吸収します。正規化後の値は比較のみに使い、Excel セルには元の生の値を保存します（監査可能性のため）。

7-2. Verdict enum

差分判定の結果を以下の3値で分類します。

# scripts/compare.py
from enum import Enum


class Verdict(Enum):
 """突合判定結果の3値分類。"""
 OK = "OK" # 期待値と現状値が一致
 NG = "NG" # 期待値と現状値が不一致（または期待値あり・実環境未検出）
 UNKNOWN = "UNKNOWN" # 期待値なし・実環境にのみ存在（設計外のリソース/属性）

第1弾との関係: 第1弾 §7-3 では Excel 表示用に5値の Verdict
（OK / NG_VALUE_DIFFERS / NG_ONLY_IN_EXPECTED / NG_ONLY_IN_ACTUAL / SKIP）を定義しました。
本セクションでは「突合ロジック層」の分類として3値に整理し直します。
Excel への書き戻し（Section 8 で扱います）では、この3値を第1弾の5値色コードにマッピングします。

7-3. DiffRow dataclass

属性1件の突合結果を格納する不変データクラスです。frozen=True を指定することで
ハッシュ可能にし、set/dict のキーとして使えるようにします。

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class DiffRow:
 """1属性の突合結果を保持する不変データクラス。"""
 resource_address: str # TF address（例: aws_instance.web[0]）
 resource_type: str # AWS リソースタイプ（例: AWS::EC2::Instance）
 resource_id: str | None  # Config 側 ID（未検出時 None）
 attribute: str  # 属性名（例: instance_type, multi_az）
 expected: Any# plan 由来の期待値（正規化後）
 actual: Any  # Config 由来の現状値（正規化後）
 verdict: Verdict# 突合判定結果
 note: str = ""  # 型不一致・欠測理由などの補足メモ

フィールド設計の意図:

• resource_address: pytest のエラーメッセージに TF コード上の場所が出るため、aws_instance.web[0] のように count / for_each 展開後のアドレスを格納します。
• resource_id: Config 側で検出できなかった場合は None を入れます。verdict=NG + resource_id=None の組み合わせが「実環境未検出」ケースです。
• expected / actual: 正規化後の値を格納します（生の値は別途 Excel へ）。
• note: 型が一致しない場合の詳細（例: "TF str '10' vs Config int 10 — 数値正規化済") を書き込み、pytest の出力で原因が一目でわかるようにします。

7-4. normalize_value() — 型正規化ユーティリティ

型正規化は突合の成否を左右する最重要関数です。サービスごとの特殊ルールが入り込みやすいため、
単独の純粋関数として切り出しておきます。

import ipaddress
from typing import Any


# 正規化時に "偽" とみなす文字列リテラル
_FALSY_STRS: frozenset[str] = frozenset(
 {"false", "no", "off", "0", "none", "null", ""}
)

# 正規化時に "真" とみなす文字列リテラル
_TRUTHY_STRS: frozenset[str] = frozenset(
 {"true", "yes", "on", "1"}
)


def normalize_value(v: Any, attr_type: str = "auto") -> Any:
 """
 突合のために値を正規化して返す。元の値は変更しない（純粋関数）。

 Args:
  v:正規化対象の値（TF 側または Config 側）
  attr_type: ヒント。"bool" / "int" / "cidr" / "list" / "auto"

 Returns:
  正規化後の値。None は None のまま返す。
 """
 if v is None:
  return None

 # ---- 文字列の前処理 ----
 if isinstance(v, str):
  v = v.strip()
  if v == "":
return None

 # ---- bool 型正規化 ----
 # attr_type ヒントがあればそれを優先、なければ文字列から推測
 if attr_type == "bool" or (
  isinstance(v, str) and v.lower() in _FALSY_STRS | _TRUTHY_STRS
 ):
  if isinstance(v, bool):
return v
  if isinstance(v, str):
return v.lower() in _TRUTHY_STRS
  return bool(v)

 # ---- 数値型正規化 ----
 if attr_type == "int":
  try:
return int(v)
  except (ValueError, TypeError):
return v
 # "auto" モードで文字列が純数値なら int に昇格
 if isinstance(v, str):
  try:
return int(v)
  except ValueError:
pass
  try:
return float(v)
  except ValueError:
pass

 # ---- CIDR 正規化 ----
 # "10.0.0.0/16" と "10.0.0.0/255.255.0.0" を同一視
 if attr_type == "cidr" or (isinstance(v, str) and "/" in v):
  try:
net = ipaddress.ip_network(v, strict=False)
return str(net)
  except ValueError:
pass

 # ---- リスト正規化 ----
 # SG ingress/egress などは順序が意味を持たないためソート
 if isinstance(v, list):
  try:
return sorted(normalize_value(item) for item in v)
  except TypeError:
# ソート不可能な混合型リスト（dicts 等）はそのまま
return [normalize_value(item) for item in v]

 return v

なぜ attr_type ヒントが必要か: "true" という文字列を見ただけでは、
それが Python の bool True に正規化すべきかどうか判断できません。
例えば description = "true" は文字列として残すべきですが、
enable_dns_hostnames = "true" は True に変換すべきです。
サービス別属性定義（Section 8 で扱います）でこのヒントを管理します。

7-5. 型正規化ルール一覧表

突合前に適用する正規化ルールをまとめます。

┌─────────────────────┬──────────────────────────────┬──────────────────────────────────┐
│ ケース  │ 例│ 正規化後 │
├─────────────────────┼──────────────────────────────┼──────────────────────────────────┤
│ 文字列 → int  │ TF: "10"  Config: 10│ 両方 10│
│ 文字列 → float│ TF: "5.0" Config: 5.0  │ 両方 5.0  │
│ bool 文字列│ TF: "true"  Config: True  │ 両方 True │
│ bool 文字列│ TF: "false" Config: False │ 両方 False│
│ Null/空文字│ TF: ""Config: null  │ 両方 None │
│ 前後空白│ "  ap-northeast-1  "│ "ap-northeast-1"  │
│ CIDR 記法  │ "10.0.0.0/255.255.0.0" │ "10.0.0.0/16"  │
│ リスト（順不同）  │ ["b","c","a"] │ ["a","b","c"]（ソート済み） │
│ 数値リスト │ ["80","443","22"]│ [22, 80, 443]（int変換+ソート）│
│ None そのまま │ None │ None  │
└─────────────────────┴──────────────────────────────┴──────────────────────────────────┘

注意: 順序に意味があるリスト（ルートテーブルの優先順位、IAM Condition の StringEquals リストなど）は
ソートしてはいけません。Section 8 のフィクスチャで attr_type="list_ordered" を渡すか、
compare() 呼び出し時に除外ルールで対象外にしてください。

7-6. compare() — 突合メイン関数

load_expected() の返り値と config_fetch() の返り値を受け取り、
属性単位の DiffRow リストを返します。

from pathlib import Path
from typing import Any


# サービスタイプ → TF address prefix のマッピング（引き当て用）
# Section 6 の config_fetch() が返す resource_type キーと揃える
_RTYPE_TO_TF_PREFIX: dict[str, str] = {
 "AWS::EC2::Instance": "aws_instance",
 "AWS::RDS::DBInstance":  "aws_db_instance",
 "AWS::ElasticLoadBalancingV2::LoadBalancer": "aws_lb",
 "AWS::Lambda::Function": "aws_lambda_function",
 "AWS::DynamoDB::Table":  "aws_dynamodb_table",
 "AWS::SNS::Topic": "aws_sns_topic",
 "AWS::SQS::Queue": "aws_sqs_queue",
 "AWS::ApiGatewayV2::Api": "aws_apigatewayv2_api",
 "AWS::EC2::VPC":"aws_vpc",
 "AWS::EC2::SecurityGroup": "aws_security_group",
}


def _find_actual(
 tf_address: str,
 resource_type: str,
 config_data: dict[str, dict[str, Any]],
) -> tuple[str | None, dict[str, Any]]:
 """
 TF address に対応する Config 現状値を探す。

 TF address（例: aws_instance.web）と Config resourceId を紐づける戦略:
 1. config_data のキー（resourceId）に tf_address の末尾ラベルが含まれる → 一致
 2. 一致なし → (None, {}) を返す（実環境未検出扱い）

 Returns:
  (resource_id, attributes_dict) または (None, {})
 """
 label = tf_address.split(".")[-1].split("[")[0]  # "aws_instance.web[0]" → "web"

 for resource_id, attrs in config_data.items():
  # タグ名・リソース名に TF ラベルが含まれているか試みる
  name_tag = attrs.get("tags", {}).get("Name", "")
  resource_name = attrs.get("resourceName", "")
  if label in name_tag or label in resource_id or label in resource_name:
return resource_id, attrs

 return None, {}


def compare(
 expected: dict[str, dict[str, Any]],
 actual: dict[str, dict[str, Any]],
 attr_types: dict[str, str] | None = None,
) -> list[DiffRow]:
 """
 期待値（plan由来）と現状値（Config由来）を突合し DiffRow リストを返す。

 Args:
  expected:{tf_address: {attr: value}} — load_expected() の返り値
  actual:  {resource_type: {resource_id: {attr: value}}} — config_fetch() の返り値
  attr_types: {attr_name: type_hint} — normalize_value() に渡すヒント辞書

 Returns:
  list[DiffRow] — 差分あり・なし問わず全属性の判定結果
 """
 attr_types = attr_types or {}
 rows: list[DiffRow] = []

 for tf_address, exp_attrs in expected.items():
  # TF address → AWS リソースタイプ への変換
  tf_prefix = tf_address.split(".")[0]
  resource_type = next(
(rt for rt, pfx in _RTYPE_TO_TF_PREFIX.items() if pfx == tf_prefix),
"AWS::Unknown",
  )

  act_by_type = actual.get(resource_type, {})
  resource_id, act_attrs = _find_actual(tf_address, resource_type, act_by_type)

  for attr, raw_expected in exp_attrs.items():
atype = attr_types.get(attr, "auto")
norm_exp = normalize_value(raw_expected, atype)

if resource_id is None:
 # 実環境で対応リソースが見つからない（全属性を NG 判定）
 rows.append(DiffRow(
  resource_address=tf_address,
  resource_type=resource_type,
  resource_id=None,
  attribute=attr,
  expected=norm_exp,
  actual=None,
  verdict=Verdict.NG,
  note="実環境未検出（Config にリソースが記録されていないか、タグ照合失敗）",
 ))
 continue

raw_actual = act_attrs.get(attr)

if raw_actual is None:
 # リソースは存在するが属性が Config に記録されていない
 rows.append(DiffRow(
  resource_address=tf_address,
  resource_type=resource_type,
  resource_id=resource_id,
  attribute=attr,
  expected=norm_exp,
  actual=None,
  verdict=Verdict.NG,
  note=f"属性 '{attr}' が Config の ConfigurationItem に存在しない",
 ))
 continue

norm_actual = normalize_value(raw_actual, atype)
match = norm_exp == norm_actual

rows.append(DiffRow(
 resource_address=tf_address,
 resource_type=resource_type,
 resource_id=resource_id,
 attribute=attr,
 expected=norm_exp,
 actual=norm_actual,
 verdict=Verdict.OK if match else Verdict.NG,
 note="" if match else (
  f"期待値: {norm_exp!r} / 現状値: {norm_actual!r}"
 ),
))

 # UNKNOWN: Config 側にあるが TF expected に存在しない属性（設計外リソース）
 for resource_type, by_id in actual.items():
  tf_prefix = _RTYPE_TO_TF_PREFIX.get(resource_type, "")
  for resource_id, act_attrs in by_id.items():
matched = any(
 addr.startswith(tf_prefix) for addr in expected
)
if not matched:
 rows.append(DiffRow(
  resource_address=f"[unknown]/{resource_type}/{resource_id}",
  resource_type=resource_type,
  resource_id=resource_id,
  attribute="*",
  expected=None,
  actual="<exists>",
  verdict=Verdict.UNKNOWN,
  note="TF コードに対応リソースが存在しない（手動作成等）",
 ))

 return rows

7-7. 欠測判定の設計

欠測のパターンは2種類あり、それぞれ異なる意味を持ちます。

欠測パターン1: expected あり、actual なし → Verdict.NG
  → TF コードに記載があるのに、実環境に存在しない
  → 例: EC2 インスタンスを TF で定義したが apply していない
  → 要因: apply 漏れ・Config Recorder の記録遅延・タグ照合失敗

欠測パターン2: expected なし、actual あり → Verdict.UNKNOWN
  → 実環境に手動作成されたリソースが存在する
  → 例: コンソールから直接作成した EC2・config の Drift
  → 対応: 「設計書への追記」か「リソースの削除」かを人間が判断

pytest での扱い方:

# test_drift.py（Section 8 で詳述）
def test_no_ng(diff_rows: list[DiffRow]) -> None:
 ng_rows = [r for r in diff_rows if r.verdict == Verdict.NG]
 assert len(ng_rows) == 0, (
  f"{len(ng_rows)} 件の NG 差分があります:\n"
  + "\n".join(f"  {r.resource_address}.{r.attribute}: {r.note}" for r in ng_rows)
 )

def test_no_unknown(diff_rows: list[DiffRow]) -> None:
 unknown_rows = [r for r in diff_rows if r.verdict == Verdict.UNKNOWN]
 # UNKNOWN は警告レベル（自動作成リソースの調査が必要）
 if unknown_rows:
  import warnings
  warnings.warn(
f"{len(unknown_rows)} 件の UNKNOWN リソースが検出されました "
f"（TF コード外のリソース）"
  )

7-8. SG ingress/egress の特殊処理

Security Group のルールはリストであり、かつ順序に意味がありません。
さらに from_port/to_port が 0（全ポート）のときに Config 側で -1 として返る
ケースもあります。

def normalize_sg_rule(rule: dict[str, Any]) -> tuple:
 """SG ルールを比較可能な正規化タプルに変換する。"""
 return (
  rule.get("ip_protocol", rule.get("protocol", "-1")),
  normalize_value(rule.get("from_port", -1), "int"),
  normalize_value(rule.get("to_port", -1), "int"),
  tuple(sorted(rule.get("cidr_blocks", []) or [])),
  tuple(sorted(rule.get("ipv6_cidr_blocks", []) or [])),
  rule.get("source_security_group_id", ""),
  rule.get("description", ""),
 )


def compare_sg_rules(
 expected_rules: list[dict],
 actual_rules: list[dict],
) -> bool:
 """
 SG ingress/egress ルールリストを順序無視で比較する。

 正規化タプルの集合（set）として比較することで、
 ルールの追加順序差を無視できる。
 """
 exp_set = {normalize_sg_rule(r) for r in expected_rules}
 act_set = {normalize_sg_rule(r) for r in actual_rules}
 return exp_set == act_set

compare() から SG を呼び出す際は、ingress / egress 属性について
通常の normalize_value() ではなく compare_sg_rules() を使います:

# compare() 内での SG 属性処理（抜粋）
if tf_prefix == "aws_security_group" and attr in ("ingress", "egress"):
 match = compare_sg_rules(
  raw_expected if isinstance(raw_expected, list) else [],
  raw_actualif isinstance(raw_actual,list) else [],
 )
 rows.append(DiffRow(
  resource_address=tf_address,
  resource_type=resource_type,
  resource_id=resource_id,
  attribute=attr,
  expected=sorted(str(normalize_sg_rule(r)) for r in (raw_expected or [])),
  actual=sorted(str(normalize_sg_rule(r)) for r in (raw_actual or [])),
  verdict=Verdict.OK if match else Verdict.NG,
  note="" if match else "SG ルール差異（順序無視・正規化タプル比較）",
 ))
 continue

7-9. exit code 設計

pytest から main.py を呼び出す際に、CI パイプラインや make から終了コードを受け取れるようにします。

# scripts/main.py（compare + exit code）
import sys
from pathlib import Path

from tf_plan_parser import parse_plan
from config_fetch import config_fetch
from compare import compare, Verdict


def run(env_dir: Path) -> int:
 """
 1環境の突合を実行し、exit code を返す。

 Returns:
  0: 全属性 OK（差分なし）
  1: 1件以上 NG あり（差分検出）
  2: データ取得・パース失敗（異常系）
 """
 try:
  expected = parse_plan(env_dir / "plan.json")
 except (FileNotFoundError, ValueError) as e:
  print(f"[ERROR] plan.json 読み込み失敗: {e}", file=sys.stderr)
  return 2

 try:
  actual = config_fetch()
 except Exception as e:
  print(f"[ERROR] Config 取得失敗: {e}", file=sys.stderr)
  return 2

 diff_rows = compare(expected, actual)

 ng_count = sum(1 for r in diff_rows if r.verdict == Verdict.NG)
 ok_count  = sum(1 for r in diff_rows if r.verdict == Verdict.OK)
 unk_count = sum(1 for r in diff_rows if r.verdict == Verdict.UNKNOWN)

 print(f"=== Drift Report: {env_dir.name} ===")
 print(f"  OK:{ok_count}")
 print(f"  NG:{ng_count}")
 print(f"  UNKNOWN: {unk_count}")

 if ng_count > 0:
  print("\n[NG 詳細]")
  for r in diff_rows:
if r.verdict == Verdict.NG:
 print(f"  {r.resource_address}.{r.attribute}: {r.note}")

 return 1 if ng_count > 0 else 0


if __name__ == "__main__":
 import argparse
 parser = argparse.ArgumentParser()
 parser.add_argument("env_dir", type=Path)
 args = parser.parse_args()
 sys.exit(run(args.env_dir))

exit code の意味:

7-10. 動作確認: compare() を単独実行する

ハンズオン環境で compare() が正しく動くことを確認します。
実際の AWS 環境がなくても、ダミーデータで動作を確認できます。

# 動作確認スクリプト（scripts/check_compare.py）
from compare import compare, Verdict, DiffRow

# ダミー期待値（plan 由来）
expected_sample = {
 "aws_instance.web": {
  "instance_type": "t3.micro",
  "multi_az": "false",
  "allocated_storage": "20",
 },
}

# ダミー現状値（Config 由来）
actual_sample = {
 "AWS::EC2::Instance": {
  "i-0123456789abcdef0": {
"tags": {"Name": "web"},
"instance_type": "t3.small",# ← 差異あり
"multi_az": False,
"allocated_storage": 20,
  },
 },
}

rows = compare(expected_sample, actual_sample)
for row in rows:
 status = "✓" if row.verdict == Verdict.OK else "✗"
 print(f"[{status}] {row.resource_address}.{row.attribute}")
 if row.verdict != Verdict.OK:
  print(f" 期待値: {row.expected!r}  現状値: {row.actual!r}")
  print(f" {row.note}")

実行結果（期待値）:

[✗] aws_instance.web.instance_type
 期待値: 't3.micro'  現状値: 't3.small'
 期待値: 't3.micro' / 現状値: 't3.small'
[✓] aws_instance.web.multi_az
[✓] aws_instance.web.allocated_storage

multi_az と allocated_storage は型正規化（"false" → False、"20" → 20）で
一致と判定されることを確認してください。

本セクションのまとめ

本セクションで実装した内容を整理します。

次の Section 8 では、この compare() 関数を pytest フィクスチャ として包み込み、
test_drift.py によって自動テストとして実行できる形に仕上げます。

Section 8. pytest によるテスト構成 — モック・フィクスチャ・assertion 粒度

ここまでの章では、terraform plan JSON から期待値を抽出し（§5）、AWS Config Advanced Query で現状値を取得し（§6）、compare() 関数で差分を DiffRow リストとして得る方法（§7）を実装してきた。

本章では、これらの関数を pytest のテストスイートとして組み立てる。「どこをモックするか」「fixture をどう切り出すか」「assertion をどの粒度で書くか」という3つの設計判断が品質と保守性を決める。

8-1. DiffRow / Verdict — Section 7 の dataclass 契約（再掲）

Section 7 で確定した dataclass を import して使う。以下は §6-3 先出し契約と同一定義であり、変更は Section 7 側で行う。

# drift_checker/models.py  （Section 7 で定義済 — 再掲）
from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from typing import Any


class Verdict(Enum):
 OK = "OK"
 NG = "NG"
 UNKNOWN = "UNKNOWN"


@dataclass(frozen=True)
class DiffRow:
 resource_address: str# TF address（例: aws_instance.web[0]）
 resource_type: str# AWS::EC2::Instance
 resource_id: str | None # Config 側 ID（未検出時 None）
 attribute: str # instance_type, cpu_credits など
 expected: Any  # plan 由来値（正規化後）
 actual: Any # Config 由来値（正規化後）
 verdict: Verdict
 note: str = "" # 型不一致・欠測理由などの補足

テストコードは from drift_checker.models import DiffRow, Verdict で import する。

8-2. pytest 実行階層の全体像

本章で構築するテストスイートの構造は次のとおりである。

tests/
├── conftest.py  # 全テストで共有する fixture
├── test_drift.py# メインのテストモジュール
├── test_load_expected.py # load_expected() 単体テスト（Section 5）
├── test_config_fetch.py  # config_fetch() 単体テスト（Section 6）
└── test_compare.py # compare() 単体テスト（Section 7）

本章では conftest.py と test_drift.py の実装に絞る。

8-3. fixture 設計 — conftest.py

fixture の責務は「テストデータの準備」と「外部依存の差し替え」である。3種類の fixture を用意する。

conftest.py 全体

# tests/conftest.py
from __future__ import annotations

from typing import Any

import boto3
import pytest
from moto import mock_aws

from drift_checker.models import DiffRow, Verdict


# ------------------------------------------------------------------ #
# 固定テストデータ #
# ------------------------------------------------------------------ #

_EXPECTED: dict[str, dict[str, Any]] = {
 "aws_instance.web[0]": {
  "resource_type": "AWS::EC2::Instance",
  "instance_type": "t3.micro",
  "monitoring": False,
  "tags": {"Env": "dev", "Project": "myapp"},
 },
 "aws_db_instance.main": {
  "resource_type": "AWS::RDS::DBInstance",
  "db_instance_class": "db.t3.small",
  "multi_az": False,
  "deletion_protection": True,
 },
}

_ACTUAL_OK: dict[str, dict[str, Any]] = {
 "i-0123456789abcdef0": {
  "resource_type": "AWS::EC2::Instance",
  "resource_address": "aws_instance.web[0]",
  "instance_type": "t3.micro",
  "monitoring": False,
  "tags": {"Env": "dev", "Project": "myapp"},
 },
 "myapp-main-db": {
  "resource_type": "AWS::RDS::DBInstance",
  "resource_address": "aws_db_instance.main",
  "db_instance_class": "db.t3.small",
  "multi_az": False,
  "deletion_protection": True,
 },
}

_ACTUAL_NG: dict[str, dict[str, Any]] = {
 "i-0123456789abcdef0": {
  "resource_type": "AWS::EC2::Instance",
  "resource_address": "aws_instance.web[0]",
  "instance_type": "t3.large", # ← NG: 期待値 t3.micro と不一致
  "monitoring": False,
  "tags": {"Env": "dev", "Project": "myapp"},
 },
 "myapp-main-db": {
  "resource_type": "AWS::RDS::DBInstance",
  "resource_address": "aws_db_instance.main",
  "db_instance_class": "db.t3.small",
  "multi_az": True,# ← NG: 期待値 False と不一致
  "deletion_protection": True,
 },
}


# ------------------------------------------------------------------ #
# fixture: expected_params / actual_params #
# ------------------------------------------------------------------ #

@pytest.fixture()
def expected_params() -> dict[str, dict[str, Any]]:
 """plan 由来の期待値辞書（load_expected() 相当）。"""
 return dict(_EXPECTED)


@pytest.fixture()
def actual_params_ok() -> dict[str, dict[str, Any]]:
 """Config 由来の現状値辞書 — 全件 OK のシナリオ。"""
 return dict(_ACTUAL_OK)


@pytest.fixture()
def actual_params_ng() -> dict[str, dict[str, Any]]:
 """Config 由来の現状値辞書 — 2件 NG のシナリオ。"""
 return dict(_ACTUAL_NG)


# ------------------------------------------------------------------ #
# fixture: diff_rows（compare() 呼び出し結果）  #
# ------------------------------------------------------------------ #

@pytest.fixture()
def diff_rows_ok(expected_params, actual_params_ok):
 """全件 OK の DiffRow リスト。"""
 from drift_checker.compare import compare
 return compare(expected_params, actual_params_ok)


@pytest.fixture()
def diff_rows_ng(expected_params, actual_params_ng):
 """NG 行を含む DiffRow リスト。"""
 from drift_checker.compare import compare
 return compare(expected_params, actual_params_ng)


# ------------------------------------------------------------------ #
# fixture: moto3 モック Config クライアント#
# ------------------------------------------------------------------ #

@pytest.fixture()
def mock_config_client(aws_credentials):
 """moto3 でモックした boto3 Config クライアント。"""
 with mock_aws():
  yield boto3.client("config", region_name="ap-northeast-1")


@pytest.fixture(scope="session")
def aws_credentials(monkeypatch_session):
 """moto3 用のダミー AWS 認証情報。実 AWS へ接続しない。"""
 import os
 monkeypatch_session.setenv("AWS_ACCESS_KEY_ID", "testing")
 monkeypatch_session.setenv("AWS_SECRET_ACCESS_KEY", "testing")
 monkeypatch_session.setenv("AWS_SECURITY_TOKEN", "testing")
 monkeypatch_session.setenv("AWS_SESSION_TOKEN", "testing")
 monkeypatch_session.setenv("AWS_DEFAULT_REGION", "ap-northeast-1")


@pytest.fixture(scope="session")
def monkeypatch_session(request):
 """session スコープの monkeypatch（pytest 組込みは function スコープのため wrapper）。"""
 from pytest import MonkeyPatch
 mp = MonkeyPatch()
 yield mp
 mp.undo()

ポイント: aws_credentials fixture を session スコープにすることで、全テストで1回だけ環境変数をセットする。moto3 は環境変数ベースの認証情報を自動検出するため、この設定で実 AWS への誤接続を防げる。

8-4. assertion 粒度の使い分け

assertion には「粗粒度」と「細粒度」の2段階を使い分ける。

両方を同じテストに書く場合、細粒度は if ng_rows: ガード内に置く。CI では粗粒度が先に FAIL するため、細粒度が誤判定を出すことはない。

# tests/test_drift.py（粒度使い分けの例）
from __future__ import annotations

import pytest

from drift_checker.models import DiffRow, Verdict


class TestDriftOK:
 """差分ゼロのシナリオ — 粗粒度 assertion のみで十分。"""

 def test_no_ng_rows(self, diff_rows_ok: list[DiffRow]) -> None:
  ng_rows = [r for r in diff_rows_ok if r.verdict == Verdict.NG]
  # 粗粒度: CI パス条件
  assert len(ng_rows) == 0, (
f"NG 行が {len(ng_rows)} 件検出された: "
+ ", ".join(f"{r.resource_address}/{r.attribute}" for r in ng_rows)
  )

 def test_all_ok_verdict(self, diff_rows_ok: list[DiffRow]) -> None:
  verdicts = {r.verdict for r in diff_rows_ok}
  assert verdicts == {Verdict.OK}


class TestDriftNG:
 """差分あり（NG）のシナリオ — 粗粒度 + 細粒度の両方。"""

 def test_ng_rows_detected(self, diff_rows_ng: list[DiffRow]) -> None:
  ng_rows = [r for r in diff_rows_ng if r.verdict == Verdict.NG]
  # 粗粒度: NG 件数を検証
  assert len(ng_rows) == 2, f"期待 NG 2件、実際 {len(ng_rows)} 件"

 def test_ng_instance_type(self, diff_rows_ng: list[DiffRow]) -> None:
  ng_rows = [r for r in diff_rows_ng if r.verdict == Verdict.NG]
  # 細粒度: instance_type の NG を確認
  instance_ng = [
r for r in ng_rows
if r.resource_address == "aws_instance.web[0]"
and r.attribute == "instance_type"
  ]
  assert len(instance_ng) == 1
  row = instance_ng[0]
  assert row.expected == "t3.micro"
  assert row.actual == "t3.large"

 def test_ng_multi_az(self, diff_rows_ng: list[DiffRow]) -> None:
  ng_rows = [r for r in diff_rows_ng if r.verdict == Verdict.NG]
  # 細粒度: multi_az の NG を確認
  multi_az_ng = [
r for r in ng_rows
if r.resource_address == "aws_db_instance.main"
and r.attribute == "multi_az"
  ]
  assert len(multi_az_ng) == 1
  row = multi_az_ng[0]
  assert row.expected is False
  assert row.actual is True

CI での使い方: pytest -q を実行し、終了コードが 0 ならドリフト無し、1 なら NG 行あり、2 なら Config 取得エラー（§7 の exit code 設計を継承）。

8-5. moto3 によるモック — ユニットテストと結合テストの使い分け

config_fetch() は boto3 の select_resource_config を呼ぶ。テスト環境によって以下の2通りを使い分ける。

moto3 モック例

# tests/test_config_fetch_mock.py
from __future__ import annotations

import json
import boto3
import pytest
from moto import mock_aws

from drift_checker.config_fetch import config_fetch


@mock_aws
def test_config_fetch_with_mock() -> None:
 """moto3 で Config クライアントをモックし、config_fetch() をユニットテストする。"""
 client = boto3.client("config", region_name="ap-northeast-1")

 # moto3 は select_resource_config のレスポンスをスタブできる
 # 実際のリソースはモック環境に存在しないため、結果は空になる
 # → config_fetch() のエラーハンドリング・ページネーションロジックを検証する
 result = config_fetch(
  resource_types=["AWS::EC2::Instance"],
  client=client,
 )
 # モック環境ではリソースが登録されていないため空辞書が返る
 assert isinstance(result, dict)


@mock_aws
def test_config_fetch_pagination() -> None:
 """ページネーションロジックを検証するテスト（モック環境）。"""
 client = boto3.client("config", region_name="ap-northeast-1")

 # config_fetch() が NextToken を正しく処理するかを確認
 # moto3 は NextToken のシミュレーションをサポートしていないため、
 # 実装側のロジックをユニットテストで直接検証する
 from drift_checker.config_fetch import _paginate_config_query
 pages = list(_paginate_config_query(client, "SELECT resourceId WHERE resourceType = 'AWS::EC2::Instance'"))
 assert isinstance(pages, list)

実クライアント結合テスト

# tests/test_config_fetch_integration.py
from __future__ import annotations

import os
import pytest

from drift_checker.config_fetch import config_fetch


@pytest.mark.integration
def test_config_fetch_real_aws() -> None:
 """実 AWS Config に対する結合テスト。--integration フラグ付きでのみ実行。

 前提条件:
 - AWS_PROFILE または AWS_ACCESS_KEY_ID 環境変数が設定済であること
 - 対象アカウントで AWS Config Recorder が有効化されていること
 - pytest.ini の [aws_config_recorder_required] が True に設定されていること

 ⚠️  SKIP = FAIL ルール: このテストをスキップした場合、
 結合テストは「未完了」扱いとなり、受入基準を満たさない。
 """
 result = config_fetch(resource_types=["AWS::EC2::Instance"])
 assert isinstance(result, dict)
 # 実環境では最低1件のリソースが存在することを期待
 assert len(result) >= 0  # 0件でも例外なく完了することを確認

結合テストは pytest マーカーで分離する。日常の CI では pytest -m "not integration" を実行し、受入試験時のみ pytest -m integration を追加する。

8-6. parametrize — resourceType 別テスト

resourceType ごとに expected/actual のペアを変えて同一テストロジックを走らせる場合は @pytest.mark.parametrize を使う。

# tests/test_drift_parametrize.py
from __future__ import annotations

from typing import Any

import pytest

from drift_checker.compare import compare
from drift_checker.models import DiffRow, Verdict


EC2_EXPECTED = {
 "aws_instance.web[0]": {
  "resource_type": "AWS::EC2::Instance",
  "instance_type": "t3.micro",
  "monitoring": False,
 }
}

RDS_EXPECTED = {
 "aws_db_instance.main": {
  "resource_type": "AWS::RDS::DBInstance",
  "db_instance_class": "db.t3.small",
  "multi_az": False,
 }
}

EC2_ACTUAL_OK = {
 "i-0123456789abcdef0": {
  "resource_type": "AWS::EC2::Instance",
  "resource_address": "aws_instance.web[0]",
  "instance_type": "t3.micro",
  "monitoring": False,
 }
}

RDS_ACTUAL_OK = {
 "myapp-main-db": {
  "resource_type": "AWS::RDS::DBInstance",
  "resource_address": "aws_db_instance.main",
  "db_instance_class": "db.t3.small",
  "multi_az": False,
 }
}


@pytest.mark.parametrize(
 "resource_label,expected,actual",
 [
  ("EC2::Instance",  EC2_EXPECTED, EC2_ACTUAL_OK),
  ("RDS::DBInstance", RDS_EXPECTED, RDS_ACTUAL_OK),
 ],
 ids=["ec2", "rds"],
)
def test_no_drift_per_resource_type(
 resource_label: str,
 expected: dict[str, dict[str, Any]],
 actual: dict[str, dict[str, Any]],
) -> None:
 """resourceType ごとにドリフトがないことを確認する。"""
 rows = compare(expected, actual)
 ng_rows = [r for r in rows if r.verdict == Verdict.NG]
 assert len(ng_rows) == 0, (
  f"[{resource_label}] NG 行検出: "
  + ", ".join(f"{r.attribute}={r.actual!r}(expected={r.expected!r})" for r in ng_rows)
 )

ids=["ec2", "rds"] を指定することで、pytest 出力が test_no_drift_per_resource_type[ec2] / test_no_drift_per_resource_type[rds] と読みやすくなる。

8-7. skip 条件 — Config Recorder 未有効時

AWS Config Recorder が無効な環境でも、ユニットテスト（moto3 モック）は常に実行できる。スキップが必要なのは「実 Config クライアントを使う結合テスト」のみである。

# tests/conftest.py への追記分
import boto3
import pytest
from botocore.exceptions import ClientError, NoCredentialsError


def _is_config_recorder_enabled() -> bool:
 """AWS Config Recorder が有効かどうかを確認するヘルパー。"""
 try:
  client = boto3.client("config", region_name="ap-northeast-1")
  recorders = client.describe_configuration_recorder_status()
  statuses = recorders.get("ConfigurationRecordersStatus", [])
  return any(s.get("recording") for s in statuses)
 except (ClientError, NoCredentialsError):
  return False


@pytest.fixture(scope="session")
def config_recorder_enabled() -> bool:
 return _is_config_recorder_enabled()

# tests/test_drift.py への追記分

def test_real_drift_detection(config_recorder_enabled: bool) -> None:
 """実 AWS Config を使った統合ドリフト検知テスト。

 Config Recorder が有効でない環境では skip する。
 ただし SKIP = FAIL ルールに従い、本番受入では skip 件数ゼロが必須。

 ⚠️  このテストをスキップした場合は「テスト未完了」扱い。
 受入基準（acceptance criteria）を満たすには Config Recorder の有効化が必要。
 """
 if not config_recorder_enabled:
  pytest.skip(
reason=(
 "AWS Config Recorder が ap-northeast-1 で有効化されていない。"
 "有効化後に再実行すること。"
 "SKIP = FAIL: このスキップは受入テスト未完了を意味する。"
)
  )

 from drift_checker.config_fetch import config_fetch
 from drift_checker.compare import compare
 from drift_checker.load_expected import load_expected

 expected = load_expected("terraform.tfplan.json")
 actual = config_fetch(resource_types=list({v["resource_type"] for v in expected.values()}))
 rows = compare(expected, actual)
 ng_rows = [r for r in rows if r.verdict == Verdict.NG]
 assert len(ng_rows) == 0, (
  f"本番環境でドリフト {len(ng_rows)} 件を検出:\n"
  + "\n".join(
f"  {r.resource_address}/{r.attribute}: "
f"expected={r.expected!r}, actual={r.actual!r}"
for r in ng_rows
  )
 )

8-8. pytest 設定ファイル

pytest.ini（シンプル構成）

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --tb=short
markers =
 integration: 実 AWS への接続が必要な結合テスト（--integration フラグで実行）
 slow: 実行に 5 秒以上かかるテスト

pyproject.toml（パッケージ管理と統合する場合）

# pyproject.toml（抜粋）
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = ["-v", "--tb=short"]
markers = [
 "integration: 実 AWS への接続が必要な結合テスト",
 "slow: 実行に 5 秒以上かかるテスト",
]

[tool.coverage.run]
source = ["drift_checker"]
omit = ["tests/*"]

[tool.coverage.report]
exclude_lines = [
 "pragma: no cover",
 "if TYPE_CHECKING:",
]

実行コマンド早見表

# ユニットテストのみ（moto3 モック、高速・実 AWS 不要）
pytest -m "not integration" -v

# 結合テスト含む（Config Recorder 有効環境が必要）
pytest -m integration -v

# カバレッジ付き
pytest -m "not integration" --cov=drift_checker --cov-report=term-missing

# 特定テストのみ
pytest tests/test_drift.py::TestDriftNG::test_ng_instance_type -v

8-9. テスト戦略まとめ

本章で構築したテストスイートの構造と判断基準を整理する。

次章（Section 9）では、environments/{dev,stg,prod}/ のマルチ環境構成と make drift-all による全環境一括実行を扱う。

Section 9. マルチ環境対応 — environments/ 分離と環境別 drift レポート

第1弾 §6 で構築した environments/{dev,stg,prod}/ 分離パターンを、本記事の drift テストにも継承します。
第1弾では各環境の plan.json を1つの Excel に統合しましたが、本記事ではさらに一歩進めて、
環境ごとの drift テストを make drift-all で一括実行し、差異を統合レポートへ集約する実装を完成させます。

9-0. 図9: マルチ環境ディレクトリ構成

9-1. ディレクトリ構成（第1弾からの拡張）

第1弾で作成した param-sheet-project/ に、drift テスト用のファイルを追加します。

param-sheet-project/
├── environments/
│├── dev/
││├── main.tf
││├── variables.tf
││├── terraform.tfvars
││├── backend.tf
││└── plan.json# make plan-all で生成（第1弾から継承）
│├── stg/
││├── main.tf
││├── variables.tf
││├── terraform.tfvars
││├── backend.tf
││└── plan.json
│└── prod/
│ ├── main.tf
│ ├── variables.tf
│ ├── terraform.tfvars
│ ├── backend.tf
│ └── plan.json
├── modules/
│├── ec2/
│├── vpc/
│├── rds/
│└── lambda/
├── scripts/
│├── tf_plan_parser/ # 第1弾 §5 のパーサ（再利用）
││├── __init__.py
││└── parser.py
│├── config_fetcher/ # 第2弾 §6 の Config 取得モジュール（新規）
││├── __init__.py
││└── fetcher.py
│├── comparator.py# 第2弾 §7 の突合ロジック（新規）
│└── drift_report.py # 本セクション: 環境別レポート集約（新規）
├── tests/
│├── conftest.py
│├── test_drift_dev.py
│├── test_drift_stg.py
│├── test_drift_prod.py
│└── pytest_env_config/
│ ├── dev.yaml # 環境別マーカー設定（本セクションで新規追加）
│ ├── stg.yaml
│ └── prod.yaml
├── output/
│├── param-sheet.xlsx# 第1弾出力（期待値列は本記事で3列に拡張）
│└── drift-reports/
│ ├── dev_drift.json # make drift-all が生成する環境別レポート（新規）
│ ├── stg_drift.json
│ └── prod_drift.json
└── Makefile

第1弾からの変更点:

• scripts/config_fetcher/ と scripts/comparator.py が新規追加（§6・§7 で実装）
• scripts/drift_report.py が本セクションで新規追加
• tests/ 以下に環境別テストファイルと pytest_env_config/ が追加
• output/drift-reports/ に環境別 JSON レポートが出力される

9-2. 環境ごとの Config 設定差（dev vs prod）

AWS Config の Recorder / Aggregator 設定は、環境の規模によって異なります。
本記事では代表的な2パターンを想定します。

dev 環境の Config 設定（Terraform）

# environments/dev/config.tf
resource "aws_config_configuration_recorder" "dev" {
  name  = "default"
  role_arn = aws_iam_role.config.arn

  recording_group {
 all_supported = true
  }
}

resource "aws_config_configuration_recorder_status" "dev" {
  name = aws_config_configuration_recorder.dev.name
  is_enabled = true

  depends_on = [aws_config_delivery_channel.dev]
}

resource "aws_config_delivery_channel" "dev" {
  name  = "default"
  s3_bucket_name = aws_s3_bucket.config.bucket
}

prod 環境の Aggregator 設定（Terraform）

# environments/prod/config_aggregator.tf
resource "aws_config_configuration_aggregator" "prod" {
  name = "prod-aggregator"

  account_aggregation_source {
 account_ids = ["123456789012"]  # 実際の本番アカウントIDに置き換えてください
 all_regions = false
 regions  = ["ap-northeast-1"]
  }
}

補足: クロスアカウント Aggregator を使う場合、委任先アカウントから aws_config_aggregate_authorization での承認も必要です。
本記事では単一アカウント構成を前提に進め、クロスアカウント設定は参考情報として記載します。

Config クライアントの環境切り替え

drift テストでは、対象環境に応じて AWS Config クライアントを切り替えます。
環境変数またはプロファイルで認証情報を制御するのが最小実装です。

# scripts/config_fetcher/fetcher.py（§6 実装の抜粋）
import boto3
import os

def get_config_client(env: str):
 """環境名に対応する boto3 Config クライアントを返す。"""
 profile = os.getenv(f"AWS_PROFILE_{env.upper()}", env)
 session = boto3.Session(profile_name=profile, region_name="ap-northeast-1")
 return session.client("config")

9-3. 環境別マーカー設定（pytest.ini + yaml）

各環境の drift テストに @pytest.mark.{env} マーカーを付与することで、
特定環境だけを実行したり、Config 未設定の環境をスキップする制御が容易になります。

pytest.ini（マーカー登録）

[pytest]
markers =
 dev: dev 環境の drift テスト
 stg: stg 環境の drift テスト
 prod: prod 環境の drift テスト
 skip_if_no_recorder: Config Recorder が無効の場合にスキップ

環境別マーカー設定ファイル（yaml）

# tests/pytest_env_config/dev.yaml
env: dev
aws_profile: dev
plan_path: environments/dev/plan.json
config_region: ap-northeast-1
recorder_required: false# dev は Recorder 任意（スキップ可）
aggregator_required: false

# tests/pytest_env_config/prod.yaml
env: prod
aws_profile: prod
plan_path: environments/prod/plan.json
config_region: ap-northeast-1
recorder_required: true # prod は Recorder 必須
aggregator_required: true

9-4. 環境差異が「本来あるべき」項目の除外ルール

dev では instance_type = t3.micro、prod では instance_type = m5.large と、
意図的に設定が異なる項目は drift として検出してはいけません。

この「環境ごとの設計上の差異」を除外する仕組みが EnvExcludeRule です。

# scripts/drift_report.py
from dataclasses import dataclass, field

@dataclass
class EnvExcludeRule:
 """環境差異が設計通りの項目を drift 判定から除外するルール。"""
 resource_type: str
 attribute: str
 envs: list[str] # このルールを適用する環境リスト
 reason: str = ""# 除外理由（レポートに記録）

# プロジェクト固有の除外ルール定義例
ENV_EXCLUDE_RULES: list[EnvExcludeRule] = [
 EnvExcludeRule(
  resource_type="aws_instance",
  attribute="instance_type",
  envs=["dev", "stg", "prod"],
  reason="環境ごとにサイジングが異なるのは設計通り（dev=t3.micro / prod=m5.large）",
 ),
 EnvExcludeRule(
  resource_type="aws_db_instance",
  attribute="instance_class",
  envs=["dev", "stg", "prod"],
  reason="DB インスタンスクラスは環境ごとに異なる設計",
 ),
 EnvExcludeRule(
  resource_type="aws_db_instance",
  attribute="multi_az",
  envs=["dev"],
  reason="dev では Multi-AZ 無効が設計通り（prod は有効）",
 ),
]

def is_excluded(env: str, resource_type: str, attribute: str) -> tuple[bool, str]:
 """指定の環境・リソース・属性が除外ルールに該当するか確認する。"""
 for rule in ENV_EXCLUDE_RULES:
  if (rule.resource_type == resource_type
 and rule.attribute == attribute
 and env in rule.envs):
return True, rule.reason
 return False, ""

テスト側でこのルールを適用する例:

# tests/test_drift_dev.py（抜粋）
from scripts.drift_report import is_excluded

def test_no_drift_dev(diff_rows_dev):
 """dev 環境で除外ルール適用後の NG 件数がゼロであること。"""
 ng_rows = [
  row for row in diff_rows_dev
  if row.verdict.value == "NG"
  and not is_excluded("dev", row.resource_type, row.attribute)[0]
 ]
 assert len(ng_rows) == 0, (
  f"dev 環境に {len(ng_rows)} 件の drift が検出されました:\n"
  + "\n".join(f"  {r.resource_address}.{r.attribute}: "
  f"expected={r.expected!r} actual={r.actual!r}"
  for r in ng_rows)
 )

9-5. Makefile — make drift-all の実装

第1弾の make plan-all を継承し、全環境の drift テストを一括実行する drift-all ターゲットを追加します。

# Makefile（第1弾の plan-all に追記）

ENVS:= dev stg prod
PYTEST := python -m pytest
REPORT_DIR := output/drift-reports

.PHONY: init-all plan-all excel drift-all drift-dev drift-stg drift-prod clean

## 全環境 terraform init
init-all:
 @for env in $(ENVS); do \
echo "=== init: $$env ==="; \
(cd environments/$$env && terraform init -input=false -reconfigure); \
 done

## 全環境 plan.json を生成（第1弾から継承）
plan-all:
 @for env in $(ENVS); do \
echo "=== plan: $$env ==="; \
(cd environments/$$env \
  && terraform plan -out=tfplan -input=false \
  && terraform show -json tfplan > plan.json); \
 done

## 第1弾 Excel 生成（期待値列）
excel:
 python scripts/tf_to_excel.py \
--envs environments/dev/plan.json \
 environments/stg/plan.json \
 environments/prod/plan.json \
--output output/param-sheet.xlsx

## 特定環境の drift テスト
drift-dev:
 @mkdir -p $(REPORT_DIR)
 $(PYTEST) tests/test_drift_dev.py -v \
--env=dev \
--json-report --json-report-file=$(REPORT_DIR)/dev_drift.json

drift-stg:
 @mkdir -p $(REPORT_DIR)
 $(PYTEST) tests/test_drift_stg.py -v \
--env=stg \
--json-report --json-report-file=$(REPORT_DIR)/stg_drift.json

drift-prod:
 @mkdir -p $(REPORT_DIR)
 $(PYTEST) tests/test_drift_prod.py -v \
--env=prod \
--json-report --json-report-file=$(REPORT_DIR)/prod_drift.json

## 全環境 drift テストを一括実行してレポート集約
drift-all: drift-dev drift-stg drift-prod
 python scripts/drift_report.py \
--reports $(REPORT_DIR)/dev_drift.json \
 $(REPORT_DIR)/stg_drift.json \
 $(REPORT_DIR)/prod_drift.json \
--exceloutput/param-sheet.xlsx \
--output  output/drift-report-summary.json

## 生成物をクリア
clean:
 rm -f environments/*/plan.json environments/*/tfplan
 rm -rf $(REPORT_DIR) output/drift-report-summary.json

実行例:

# 全環境 plan → Excel → drift テスト → サマリレポート を一気通貫で実行
make plan-all && make excel && make drift-all

9-6. 環境別レポート集約スクリプト（drift_report.py）

make drift-all の最終ステップで呼ばれる drift_report.py は、
3環境の pytest JSON レポートを読み込み、差分行を Excel の ENV_SUBCOLS（期待値・現状値・判定）へ書き戻します。

#!/usr/bin/env python3
"""drift_report.py — 環境別 pytest JSON レポートを集約して Excel に書き戻す。"""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

import openpyxl
from openpyxl.styles import PatternFill

# 第1弾 §7 の色定数（再利用）
COLOR_OK= "C6EFCE"
COLOR_NG= "FFC7CE"
COLOR_UNKNOWN = "FFEB9C"

ENV_SUBCOLS = ["期待値", "現状値", "判定"]


def load_pytest_report(path: Path) -> list[dict[str, Any]]:
 """pytest-json-report 形式の JSON から drift 行を抽出する。"""
 with path.open(encoding="utf-8") as f:
  report = json.load(f)

 drift_rows: list[dict[str, Any]] = []
 for test in report.get("tests", []):
  if test.get("outcome") == "failed":
drift_rows.append({
 "nodeid": test["nodeid"],
 "message": test.get("call", {}).get("longrepr", ""),
})
 return drift_rows


def fill_excel_env_cols(
 wb: openpyxl.Workbook,
 env: str,
 drift_rows: list[dict[str, Any]],
) -> None:
 """Sheet1 の ENV_SUBCOLS（現状値・判定列）を drift テスト結果で埋める。"""
 ws = wb["Sheet1"]
 # ENV_SUBCOLS の列インデックスは §7（第1弾 write_excel）で確定済みのヘッダ行を検索して解決
 # ここでは簡略化のため env 別列オフセットを定数とする
 env_col_offset = {"dev": 0, "stg": 3, "prod": 6}.get(env, 0)
 ACTUAL_COL  = 5 + env_col_offset# 「現状値」列（1-indexed）
 VERDICT_COL = 6 + env_col_offset# 「判定」列

 ng_nodeids = {r["nodeid"] for r in drift_rows}

 for row_idx in range(3, ws.max_row + 1):# Row3 以降がデータ行
  resource_cell = ws.cell(row=row_idx, column=2)  # リソース名列
  if resource_cell.value is None:
continue

  verdict_cell = ws.cell(row=row_idx, column=VERDICT_COL)
  actual_cell  = ws.cell(row=row_idx, column=ACTUAL_COL)

  has_ng = any(resource_cell.value in nid for nid in ng_nodeids)

  if has_ng:
verdict_cell.value = "× NG"
verdict_cell.fill = PatternFill("solid", fgColor=COLOR_NG)
actual_cell.fill  = PatternFill("solid", fgColor=COLOR_NG)
  else:
verdict_cell.value = "✓ OK"
verdict_cell.fill = PatternFill("solid", fgColor=COLOR_OK)


def main() -> None:
 parser = argparse.ArgumentParser(description="drift レポート集約")
 parser.add_argument("--reports", nargs="+", required=True)
 parser.add_argument("--excel",required=True)
 parser.add_argument("--output",  required=True)
 args = parser.parse_args()

 wb = openpyxl.load_workbook(args.excel)

 summary: dict[str, Any] = {"environments": {}}

 for report_path in args.reports:
  p = Path(report_path)
  env = p.stem.replace("_drift", "")# dev_drift.json → dev
  drift_rows = load_pytest_report(p)
  fill_excel_env_cols(wb, env, drift_rows)
  summary["environments"][env] = {
"ng_count": len(drift_rows),
"report": str(p),
  }
  print(f"[{env}] drift NG: {len(drift_rows)} 件")

 wb.save(args.excel)
 print(f"Excel 更新完了: {args.excel}")

 Path(args.output).write_text(
  json.dumps(summary, ensure_ascii=False, indent=2),
  encoding="utf-8",
 )
 print(f"サマリレポート出力: {args.output}")


if __name__ == "__main__":
 main()

9-7. 第1弾 ENV_SUBCOLS との連携

第1弾の write_excel() では、各環境列のヘッダを ENV_SUBCOLS = ['期待値', '現状値', '判定'] の3列で構成しています。

Row1: サービス | リソース名 | 属性 | === 期待値 ===| === 現状値 ===  | === 判定 ===
Row2:|  |  | dev  | stg  | prod  | dev  | stg  | prod  | dev | stg | prod

本記事で drift_report.py が fill_excel_env_cols() を呼ぶことで、
第2弾の drift テスト結果が第1弾の Excel に自動的に書き戻される構成になります。

第1弾 Excel（output/param-sheet.xlsx）と第2弾の drift_report.py は、
この ENV_SUBCOLS ヘッダ位置を契約として共有します。
第1弾 §7 で列インデックスを変更した場合は、drift_report.py 側の env_col_offset も合わせて更新してください。

Section 9 執筆完了。 environments/ 分離・Config 環境差・make drift-all・除外ルール・ENV_SUBCOLS 連携を網羅。

Section 10. ハンズオン実行と成果物確認

いよいよ本記事で実装したパイプライン全体を一気に走らせます。Section 2〜9 で構築した tflint → terraform plan → AWS Config Advanced Query → 突合 → pytest の5段階を、1つのシェルスクリプトで連続実行し、差分検知の流れを体験します。

Config 未有効の読者へ: Section 5（plan JSON 期待値抽出）と Section 6（Config 現状値取得）までは AWS Config Recorder なしでも動作確認できます。Config Recorder が未有効の場合は、後半の config_fetch() 呼び出しで pytest.skip が発火します。その場合でも tflint・plan 解析・compare() ロジック部分のユニットテストは通過しますので、Config 有効化前のウォームアップとして活用してください。

10-1. 事前準備チェックリスト

ハンズオンを始める前に、以下の準備が整っているか確認します。

# 動作環境チェック
tflint --version  # 0.52 以上
terraform --version  # 1.9.x 系
python --version  # 3.11 以上
pytest --version  # 8.x 系
aws --version  # 2.x 系（CLI v2）

必要な AWS 権限（IAM ポリシー）:
- config:SelectResourceConfig
- config:DescribeConfigurationRecorders
- config:DescribeConfigurationRecorderStatus
- ec2:DescribeInstances（リソースタイプ確認用）
- s3:GetObject（terraform state 参照）
- iam:PassRole（Config Recorder 用 IAM ロール設定時）

これらを持つ IAM ユーザー or ロールで AWS CLI の認証が通っていること:
$ aws sts get-caller-identity
{
 "UserId": "AIDAXXXXXXXXXXXXXXXXX",
 "Account": "123456789012",
 "Arn": "arn:aws:iam::123456789012:user/param-sheet-tester"
}

# Python 依存ライブラリのインストール
pip install pytest boto3 openpyxl moto[config]

# プロジェクトディレクトリ構成確認
ls -la
# 期待する構成:
# param-sheet-drift-test/
# ├── environments/
# │├── dev/
# ││├── main.tf
# ││└── terraform.tfvars
# │├── stg/
# │└── prod/
# ├── tests/
# │├── conftest.py
# │└── test_drift.py
# ├── src/
# │├── config_fetch.py
# │├── compare.py
# │└── normalize.py
# ├── .tflint.hcl
# ├── Makefile
# └── pytest.ini

10-2. エンドツーエンド実行スクリプト

以下のスクリプトを run_drift_check.sh として保存し、プロジェクトルートから実行します。

#!/usr/bin/env bash
# run_drift_check.sh — tflint → plan → Config → compare → pytest の一気通貫実行
set -euo pipefail

ENVIRONMENT="${1:-dev}"
PLAN_FILE="plan_${ENVIRONMENT}.json"

echo "=========================================="
echo "  Drift Check Pipeline — ${ENVIRONMENT}"
echo "=========================================="

# ── Step 1: tflint 静的検査（~2秒）─────────────────────────────
echo ""
echo "▶ [1/5] tflint 静的検査..."
cd "environments/${ENVIRONMENT}"
tflint --init --config ../../.tflint.hcl 2>&1
TFLINT_EXIT=$?
if [ $TFLINT_EXIT -ne 0 ]; then
 echo "✗ tflint FAILED — HCL 構文エラーまたは lint 違反があります"
 echo "  pytest を実行しても無意味なため、ここで中断します"
 exit 1
fi
echo "✓ tflint PASSED"

# ── Step 2: terraform plan JSON 生成（~10秒）────────────────────
echo ""
echo "▶ [2/5] terraform plan JSON 生成..."
terraform init -backend-config="key=${ENVIRONMENT}/terraform.tfstate" -input=false -no-color 2>&1
terraform plan \
 -var-file="terraform.tfvars" \
 -out="${PLAN_FILE}.bin" \
 -no-color 2>&1
terraform show -json "${PLAN_FILE}.bin" > "../../${PLAN_FILE}"
echo "✓ terraform plan JSON 生成完了: ${PLAN_FILE} ($(wc -c < ../../${PLAN_FILE}) bytes)"
cd ../../

# ── Step 3: AWS Config Advanced Query で現状値取得（~5秒）───────
echo ""
echo "▶ [3/5] AWS Config 現状値取得..."
python3 -c "
from src.config_fetch import config_fetch
import json

resource_types = [
 'AWS::EC2::Instance',
 'AWS::RDS::DBInstance',
 'AWS::ElasticLoadBalancingV2::LoadBalancer',
 'AWS::S3::Bucket',
]
actual = config_fetch(resource_types)
print(f'取得リソース数: {sum(len(v) for v in actual.values())}件')
with open('actual_${ENVIRONMENT}.json', 'w') as f:
 json.dump({k: {rk: dict(rv) for rk, rv in v.items()} for k, v in actual.items()}, f, indent=2)
"
echo "✓ Config 現状値取得完了: actual_${ENVIRONMENT}.json"

# ── Step 4: 突合（~1秒）────────────────────────────────────────
echo ""
echo "▶ [4/5] 期待値 vs 現状値 突合..."
python3 -c "
from src.compare import compare
from src.config_fetch import load_expected
import json

expected = load_expected('${PLAN_FILE}')
with open('actual_${ENVIRONMENT}.json') as f:
 actual_raw = json.load(f)

# DiffRow リスト生成
diff_rows = compare(expected, actual_raw)
ng_count= sum(1 for r in diff_rows if r.verdict.value == 'NG')
ok_count= sum(1 for r in diff_rows if r.verdict.value == 'OK')
unk_count  = sum(1 for r in diff_rows if r.verdict.value == 'UNKNOWN')

print(f'OK={ok_count}  NG={ng_count}  UNKNOWN={unk_count}  TOTAL={len(diff_rows)}')
with open('diff_${ENVIRONMENT}.json', 'w') as f:
 import dataclasses
 json.dump([dataclasses.asdict(r) for r in diff_rows], f, indent=2, default=str)
"
echo "✓ 突合完了: diff_${ENVIRONMENT}.json"

# ── Step 5: pytest 実行（~20秒）────────────────────────────────
echo ""
echo "▶ [5/5] pytest 実行..."
PLAN_PATH="${PLAN_FILE}" \
AWS_ENV="${ENVIRONMENT}" \
pytest tests/ -v --tb=short \
 --junit-xml="reports/drift_${ENVIRONMENT}_$(date +%Y%m%d_%H%M%S).xml"
PYTEST_EXIT=$?

echo ""
echo "=========================================="
if [ $PYTEST_EXIT -eq 0 ]; then
 echo "  ✓ 全テスト PASSED — drift なし"
else
 echo "  ✗ FAILED — drift 検知。reports/ を確認してください"
fi
echo "=========================================="
exit $PYTEST_EXIT

実行方法:

chmod +x run_drift_check.sh

# dev 環境の drift チェック
./run_drift_check.sh dev

# stg 環境の drift チェック
./run_drift_check.sh stg

# Makefile 経由（全環境一括）
make drift-all

10-3. 実行結果サンプル

PASS 時（drift なし）

以下は全リソースが設計値と一致している正常ケースの出力例です。

==========================================
  Drift Check Pipeline — dev
==========================================

▶ [1/5] tflint 静的検査...
✓ tflint PASSED

▶ [2/5] terraform plan JSON 生成...
✓ terraform plan JSON 生成完了: plan_dev.json (84512 bytes)

▶ [3/5] AWS Config 現状値取得...
取得リソース数: 23件
✓ Config 現状値取得完了: actual_dev.json

▶ [4/5] 期待値 vs 現状値 突合...
OK=46  NG=0  UNKNOWN=3  TOTAL=49
✓ 突合完了: diff_dev.json

▶ [5/5] pytest 実行...

tests/test_drift.py::test_no_ng_verdict PASSED [  8%]
tests/test_drift.py::test_ec2_instance_type[aws_instance.web[0]] PASSED[ 16%]
tests/test_drift.py::test_ec2_instance_type[aws_instance.web[1]] PASSED[ 25%]
tests/test_drift.py::test_rds_engine_version[aws_db_instance.main] PASSED [ 33%]
tests/test_drift.py::test_rds_multi_az[aws_db_instance.main] PASSED [ 41%]
tests/test_drift.py::test_alb_scheme[aws_lb.frontend] PASSED  [ 50%]
tests/test_drift.py::test_s3_versioning[aws_s3_bucket.artifacts] PASSED[ 58%]
tests/test_drift.py::test_all_resources_covered PASSED  [ 66%]
tests/test_drift.py::test_unknown_count_within_threshold PASSED  [ 75%]
tests/test_drift.py::test_excel_sheet2_written PASSED[ 83%]
tests/test_drift.py::test_drift_report_generated PASSED [100%]

=================== 11 passed in 18.43s ====================

==========================================
  ✓ 全テスト PASSED — drift なし
==========================================

FAIL 時（drift 検知）

本番環境で EC2 の instance_type が変更されたケースです。pytest は赤色で FAIL を表示します。

==========================================
  Drift Check Pipeline — prod
==========================================

▶ [1/5] tflint 静的検査...
✓ tflint PASSED

▶ [2/5] terraform plan JSON 生成...
✓ terraform plan JSON 生成完了: plan_prod.json (112088 bytes)

▶ [3/5] AWS Config 現状値取得...
取得リソース数: 61件
✓ Config 現状値取得完了: actual_prod.json

▶ [4/5] 期待値 vs 現状値 突合...
OK=118  NG=2  UNKNOWN=5  TOTAL=125
✓ 突合完了: diff_prod.json

▶ [5/5] pytest 実行...

tests/test_drift.py::test_no_ng_verdict FAILED [  8%]
tests/test_drift.py::test_ec2_instance_type[aws_instance.api[0]] FAILED[ 16%]
...

================================ FAILURES =================================

test_no_ng_verdict
------------------
AssertionError: 2 件の NG が検出されました:
  [NG] aws_instance.api[0] / instance_type
 expected='m5.large' actual='t3.medium'
 note='Terraformコード値と異なる — 手動変更または apply 未反映の可能性'
  [NG] aws_instance.api[1] / instance_type
 expected='m5.large' actual='t3.medium'

test_ec2_instance_type[aws_instance.api[0]]
-------------------------------------------
AssertionError:
  Expected : m5.large
  Actual: t3.medium
  Verdict  : NG

=================== 2 failed, 9 passed in 21.07s ====================

==========================================
  ✗ FAILED — drift 検知。reports/ を確認してください
==========================================

10-4. Excel Sheet2「差分一覧」自動転記

drift が検知された場合、test_excel_sheet2_written テストが第1弾で生成した Excel ファイルの Sheet2「差分一覧」に結果を自動転記します。転記後の Sheet2 は以下の形式になります。

Sheet2: 差分一覧（2026-04-19 03:15:22 更新）

| リソースアドレス | リソースタイプ | 属性  | 期待値（TF） | 現状値（Config） | 判定 |
|------------------------------|-------------------------|-------------------|--------------|------------------|---------|
| aws_instance.api[0] | AWS::EC2::Instance| instance_type  | m5.large  | t3.medium  | NG|
| aws_instance.api[1] | AWS::EC2::Instance| instance_type  | m5.large  | t3.medium  | NG|
| aws_instance.web[0] | AWS::EC2::Instance| instance_type  | t3.small  | t3.small| OK|
| aws_instance.web[1] | AWS::EC2::Instance| instance_type  | t3.small  | t3.small| OK|
| aws_db_instance.main| AWS::RDS::DBInstance | instance_class | db.r6g.large | db.r6g.large  | OK|
| aws_db_instance.main| AWS::RDS::DBInstance | multi_az | true| true | OK|
| aws_lb.frontend  | AWS::ELBV2::LoadBalancer| scheme| internal  | internal| OK|
| aws_s3_bucket.artifacts| AWS::S3::Bucket| versioning_status | Enabled| Enabled | OK|
| aws_instance.batch[0]  | AWS::EC2::Instance| tags.Environment  | prod| prod | OK|
| aws_instance.batch[0]  | AWS::EC2::Instance| tags.Team| —| backend | UNKNOWN |

UNKNOWN の解釈: Terraform コードに tags.Team の記述がなく、Config 側にのみ値が存在するケースです。設計書に記載のないタグが手動で付与された状態を意味します。厳密な管理方針の組織では NG 扱いにする運用も選べます（Section 7 の Verdict 設計を参照）。

10-5. トラブルシュート

ケース 1: AWS Config Recorder が未有効

SKIPPED tests/test_drift.py::test_no_ng_verdict
  — reason: AWS Config Recorder is not enabled in this region (ap-northeast-1).
Enable it with: terraform apply -target=aws_config_configuration_recorder.main
(Section 3 参照)

========================= 11 skipped, 0 passed in 0.82s ========================

対処法: Section 3 の手順で aws_config_configuration_recorder を Terraform で作成・適用してください。ハンズオン中の月額コストは数ドル程度です。Recorder を有効化したくない場合は、Section 5 までのユニットテスト（plan 解析・型正規化・突合ロジック）のみ実行できます:

# Config なしで動くユニットテストのみ実行
pytest tests/ -m "not requires_config" -v

SKIP = FAIL ルール: 本プロジェクトでは pytest の SKIP は未完了と同義です。Config Recorder を有効化せずに「テスト全件通過」とはなりません。受入試験として使う場合は、必ず Config Recorder を有効にした環境で実施してください。

ケース 2: IAM 権限不足

FAILED tests/test_drift.py::test_no_ng_verdict
botocore.exceptions.ClientError: An error occurred (AccessDeniedException) when
calling the SelectResourceConfig operation:
User: arn:aws:iam::123456789012:user/param-sheet-tester is not authorized to
perform: config:SelectResourceConfig

対処法: 実行 IAM ユーザー/ロールに以下のポリシーを追加します。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "AllowConfigQuery",
"Effect": "Allow",
"Action": [
  "config:SelectResourceConfig",
  "config:DescribeConfigurationRecorders",
  "config:DescribeConfigurationRecorderStatus"
],
"Resource": "*"
 }
  ]
}

インラインポリシーとして IAM ユーザーに直接付与するか、Section 3 の Terraform ハンズオンで用意した IAM ロールを使用してください。

ケース 3: resourceType 未対応

WARNING: config_fetch — resourceType 'AWS::Lambda::Function' は
Advanced Query でサポートされていないため、スキップしました。
(ConfigService.SelectResourceConfig: UnsupportedResourceTypeException)

対処法: AWS Config Advanced Query がサポートするリソースタイプは随時拡大されていますが、すべての resourceType が対応しているわけではありません。未対応の場合は Verdict.UNKNOWN として処理し、テストを SKIP させます。

# src/config_fetch.py の該当ハンドリング
try:
 response = client.select_resource_config(Expression=sql)
except client.exceptions.UnsupportedResourceTypeException:
 logger.warning(
  f"resourceType '{rt}' は Advanced Query 非対応 — UNKNOWN 扱いにします"
 )
 result[rt] = {}  # 空辞書 → compare() 側で UNKNOWN 判定

対応リソースタイプの最新リストは AWS 公式ドキュメント: Supported resource types for AWS Config Advanced queries を参照してください。

10-6. 実行時間の目安と最適化

パイプライン全体の実行時間目安（リソース数が数十台規模の一般的なエンタープライズ案件の場合）:

Step 1: tflint 静的検査 ─── ~2 秒
Step 2: terraform plan JSON─── ~10 秒（backend 接続 + state ダウンロード込み）
Step 3: Config Advanced Query ─── ~5 秒（ページネーション 1〜2 ページ）
Step 4: 突合処理（compare）─── ~1 秒（Python インメモリ）
Step 5: pytest 実行  ─── ~20 秒（フィクスチャ初期化 + 全テスト）
─────────────
合計  ─── ~38 秒

最適化のヒント:
– terraform state が S3 にある場合、terraform plan は S3 アクセスを伴うため、VPC エンドポイントを設定するとレイテンシが改善します
– Config Advanced Query の取得対象 resourceType を最小限に絞ることで Step 3 を短縮できます
– CI 環境での実行はキャッシュ（~/.terraform.d/plugins）を活用することで plan 時間を短縮できます

Section 11. まとめと次の発展（CI/CD 組込み導線）

本記事では、AWS Config と Terraform パラメーターシートを突合する単体テスト自動化パイプラインを構築しました。このセクションでは、シリーズ2本を通して得た成果を振り返り、次のステップへの道筋を示します。

11-1. シリーズ2本で完成した「設計→検証」の循環

第1弾・第2弾を通して構築したパイプラインを図で整理します。

【第1弾】設計の可視化
─────────────────────────────────────────────────────────────────
  Terraform コード
│
▼
  terraform plan -json
│
▼
  parse_plan()─── plan JSON から期待値を抽出
│
▼
  write_excel()  ─── Excel パラメーターシート生成
（期待値列・現状値列・判定列）
│
▼
  関係者への配布・設計レビュー ✓ （第1弾の成果）

【第2弾（本記事）】構築後の検証自動化
─────────────────────────────────────────────────────────────────
  ┌─ tflint ─────────────── HCL 静的検査（構文・best practice）
  │
  ├─ parse_plan()（再利用）─ 期待値 load_expected()
  │
  ├─ config_fetch()──────── AWS Config Advanced Query → 現状値
  │
  ├─ compare()───────────── 型正規化・差分判定・Verdict
  │
  ├─ pytest──────────────── 単体テストとして合否判定
  │
  └─ write_excel()（拡張）─ Sheet2「差分一覧」自動転記 ✓ （第2弾の成果）
─────────────────────────────────────────────────────────────────

  ↑ 第1弾と第2弾が組み合わさって
  「設計書（Excel）の作成 → 実環境との突合 → 差分検知」
 という完全なループが完成する

シリーズ2本で達成したこと:

エンタープライズ案件で典型的な「構築完了 → パラメーターシートの手動突合（500チェック / 数日）」という作業が、コマンド1本・数十秒で完了するようになりました。

11-2. 本記事で意図的に踏み込まなかった領域

技術的に実現可能でも、スコープや設計方針から除外した項目があります。次のステップを検討する際の参考にしてください。

定期実行（cron / EventBridge）

本記事の位置づけは「構築完了後の受入検証（単体テスト）」です。毎夜自動実行や EventBridge でのトリガーは設計外です。定期監視が必要な場合は AWS Config のマネージドルール（例: EC2_INSTANCE_TYPE）や AWS Security Hub との組合せを検討してください。

CI/CD 組込み

pytest の終了コードを CI の fail 条件として使えば、「設計外変更があったらパイプラインを止める」という運用が可能です。ただし、本記事ではその実装を第2弾（GitHub Actions+OIDC）に委譲しています。

マルチアカウント Aggregator

本記事の Section 9 では単一アカウント内のマルチ環境を対象としました。AWS Organizations + Config Aggregator を用いた複数 AWS アカウントの横断 drift チェックは別記事候補です。

Config Custom Rule / Conformance Pack

compare() で実装した突合ロジックを AWS Lambda + Config Custom Rule に変換すれば、Config のマネージドサービスとして継続監視が可能です。ただし、実装コストが高く、Terraform コードとのバージョン管理が複雑になるため、本記事ではスコープ外としました。

11-3. 読者別の発展ルート

ルート A: CI/CD に組み込みたい

本記事の run_drift_check.sh を GitHub Actions の jobs.drift-check ステップに追加し、terraform apply の後続ジョブとして実行する構成を取ります。

# GitHub Actions ステップ例
- name: Drift Check
  run: ./run_drift_check.sh ${{ env.ENVIRONMENT }}
  env:
 AWS_REGION: ap-northeast-1

インフラと CI/CD パイプラインの完全な構成は 第2弾（GitHub Actions+OIDC で PR駆動 CI/CD） で解説しています。本記事のスクリプトはそのまま流用できます。

ルート B: Config Custom Rule に発展させたい

compare() のロジックを AWS Lambda に移植し、aws configservice put-config-rule で登録することで、Config のマネージドサービスとして継続的に drift を監視できます。ただし、Terraform のリソースアドレスと Config の resourceId の紐付けロジックをクラウド側で維持する必要があり、実装コストは本記事の3〜5倍程度を見込んでください。

ルート C: 社内展開・勉強会材料として使いたい

本シリーズ2本（第1弾 Excel 生成 + 第2弾 drift テスト）のスクリプトセットは、社内インフラチームの勉強会教材として使いやすい構成になっています。特に「手動突合の前/後 比較」（Section 2）と「実行時間 38 秒の実演」（Section 10）が聴衆への説得力が高いポイントです。

11-4. おわりに

「AWS パラメーターシート自動化シリーズ」全3弾を通して、次のことが実現できます:

1. 第1弾: Terraform コードを Single Source of Truth として、設計書（Excel）を自動生成する仕組みを構築
2. 第2弾（本記事）: 構築後の実環境を AWS Config で取得し、設計書と pytest で突合する受入検証を自動化
3. 第3弾: AWS Config Remediation と SSM Automation runbook で検知した drift を自動修復し、設計→検証→修復の閉ループを完成

この3本が揃うことで、「設計 → 実装 → 受入検証 → 自動修復」の完全ループがコード化されました。Terraform コードを修正すれば設計書が更新され、デプロイ後に drift チェックを走らせれば差分が自動検出・自動修復されます。

エンタープライズ案件の IaC 化を進める上で、「設計書と実環境の乖離」は避けられない課題です。本シリーズが、その課題をコードで解決する1つのアプローチとして参考になれば幸いです。

CI/CD に組み込んで「デプロイ → 自動 drift チェック」のパイプラインへ発展させる場合は、以下のシリーズが役立ちます。

CI/CD へ発展させる（複数人開発シリーズ 第2弾: GitHub Actions+OIDC）