Lambda × S3 で理解するPython DI・Pytest (初歩の初歩)

0. はじめに

生成AI に書いてもらった Lambda 関数のコードを読んでいたら、いきなりこんなクラスから始まりました。

from typing import Protocol

class S3Client(Protocol):
    def get_object(self, Bucket: str, Key: str) -> dict: ...

…処理が書いてない。

Python は大学時代に少しかじった程度で、業務効率化のスクリプトや小規模な BI（Streamlit）くらいしか書いたことがありません。(最近はもっぱらバイブコーディング)

dataclass？ Protocol？ MagicMock ? fixture ? parametrize？
ん～、見覚えあるのもあるけど、何がどんな役割してて必要なものなのか分からない。

生成AI が書いたコードを理解するたびにひとつひとつ調べるのも大変です。
さすがに基本くらいは押さえておこうということで、今回は DI とテストという切り口で、(生成AIにも頼りつつ)備忘録的に整理してみました。

この記事で (きっと) わかること

• DI（依存性注入）とは結局何なのか、なぜテストが楽になるのか
• dataclass / Protocol / pytest / MagicMock / fixture / parametrize の役割分担
• 本物の AWSリソース に繋がずに、Lambda をテストする具体的な書き方

1. なぜ Lambda はテストしにくいのか

まずは「ダメな例」から。
素人目には普通に見えるこういうコード、とてもテストしづらいです。

import boto3

def lambda_handler(event, context):
    s3 = boto3.client("s3")       # ← 関数の中で自分で作っている
    obj = s3.get_object(Bucket="my-bucket", Key=event["key"])
    return {"size": len(obj["Body"].read())}

この関数を動かすと、本物の S3 に接続しにいくからです。
テストのたびに本物のバケットが必要で、多少なりともお金がかかります。

関数が、自分の仕事道具（S3クライアント）を、自分の中で作っていることが原因です。
これを解決するのが DI(依存性注入) です。

DI（依存性注入）＝ 道具を「外から渡す」

端的にまとめると、

必要な道具を関数の中で作らず、引数で外から受け取る。

先ほどのコードをこう分けます。

import boto3

# 処理本体：s3 を「引数で受け取る」＝ 注入(injection)
def process(event, s3):
    obj = s3.get_object(Bucket="my-bucket", Key=event["key"])
    return {"size": len(obj["Body"].read())}

# エントリポイント：ここで本物を作って渡す（配線担当）
def lambda_handler(event, context):
    s3 = boto3.client("s3")
    return process(event, s3)

関数	役割
lambda_handler	S3クライアントを作って渡すだけの 配線担当
process	渡された道具を使うだけの ロジック担当

こうすると、テストでは process に本物の代わりに偽物を渡せます。
この偽物がのちほど出てくる MagicMock です。

2. 今回作るもの

あくまでDIやテストコードの書き方を理解するためなので、シンプルなお題にします。

お題: S3 の CSV を読み、行数を数えて、結果を別キーに JSON で書き戻す Lambda

完成形のコードを先に貼ります。コメントの「← ○○章」を目印に、以降で分解していきます。

# app.py
import json
from dataclasses import dataclass
from typing import Protocol

import boto3


class ValidationError(Exception):
    """入力イベントが不正なときに投げるカスタム例外"""


@dataclass
class LoadEvent:                      # ← dataclass章
    bucket: str
    key: str


class S3Client(Protocol):             # ← Protocol章
    def get_object(self, Bucket: str, Key: str) -> dict: ...
    def put_object(self, Bucket: str, Key: str, Body: bytes) -> dict: ...


def parse_event(event: dict) -> LoadEvent:
    if "bucket" not in event or "key" not in event:
        raise ValidationError("bucket と key は必須です")
    if not event["key"].endswith(".csv"):
        raise ValidationError("key は .csv で終わる必要があります")
    return LoadEvent(bucket=event["bucket"], key=event["key"])


def process(event: LoadEvent, s3: S3Client) -> dict:   # s3 を引数で受け取る＝DI
    obj = s3.get_object(Bucket=event.bucket, Key=event.key)
    body = obj["Body"].read().decode("utf-8")
    line_count = len([ln for ln in body.splitlines() if ln.strip()])

    result = {"source_key": event.key, "line_count": line_count}
    result_key = event.key.replace(".csv", ".result.json")
    s3.put_object(Bucket=event.bucket, Key=result_key,
                  Body=json.dumps(result).encode("utf-8"))
    return result


def lambda_handler(event: dict, context) -> dict:
    parsed = parse_event(event)
    s3 = boto3.client("s3")           
    return process(parsed, s3)

3. コードを整える道具

まずは実装側の2つ、dataclass と Protocol から見ていきます。
どちらも「無くても動く」補助ですが、あるとコードの意図が読みやすくなります。

3.1. dataclass ─ フィールドから特殊メソッドを自動生成するデコレーター

3.1.1. dataclass とは

dataclass は、標準ライブラリ dataclasses が提供するデコレーターです。
クラスに宣言したフィールド（型注釈付きのクラス変数）を読み取り、__init__・__repr__・__eq__ といった特殊メソッドを自動生成します。

デコレーターとは：@名前 の構文で、直後の関数・クラスの定義を引数として受け取り、加工した定義を返す仕組み（実体はただの関数）。@dataclass は次の糖衣構文（シンタックスシュガー）です。

@dataclass
class LoadEvent: ...

# ↑ は下と等価
class LoadEvent: ...
LoadEvent = dataclass(LoadEvent)   # クラスを受け取り、メソッドを足して返す

💭 デコレーターって美味しそうな名前ですよね。
Streamlitでデータをキャッシュしてくれるやつがあって、デコレーターって便利だなぁって第一印象でした。(小並感)

3.1.2. dataclassを書く目的

lambda_handlerに渡させる event は辞書型です 。辞書には弱点があります。

• event["bukcet"] と打ち間違えても、実行するまでエラーにならない
• どんなキーがあるべきか、コードから読み取れない
• 値が文字列なのか数値なのか、型が分からない

dataclass はこれを一気に解決します。

from dataclasses import dataclass

@dataclass
class LoadEvent:
    bucket: str
    key: str

@dataclass を付けるだけで、Python が __init__・__repr__・__eq__ を自動生成してくれます。

e = LoadEvent(bucket="my-bucket", key="data.csv")
e.bucket                                   # → "my-bucket"（e.bukcet は即エラー）
e == LoadEvent("my-bucket", "data.csv")    # → True（中身が同じなら等しい）

この __eq__ の自動生成が地味に効きます。
テストで
assert parse_event(...) == LoadEvent("b", "k")
と 1行で比較できるからです。

3.1.3. 注意点など

• 可変デフォルトの罠：リストや辞書を直接デフォルトにすると ValueError になります。

from dataclasses import dataclass, field

@dataclass
class LoadEvent:
    bucket: str
    key: str
    tags: list[str] = field(default_factory=list)   # ○ これが正解
    # tags: list[str] = []                          # × ValueError

[] を直接書くと全インスタンスが同じ1個のリストを共有してバグの温床になるため、Python が禁止しています。default_factory で「毎回新しく作る」ようにします。

• イミュータブル化：@dataclass(frozen=True) にすると代入禁止（不変）になり、作ったインタンスのフィールドを途中で書き換えられる事故を防げます。イベントのような「作ったら変えない値」に最適です。

3.2. Protocol ─ 構造的部分型でインターフェースを定義する型

3.2.1. Protocol とは

Protocol は typing モジュールが提供する型で、これを継承したクラスは「プロトコルクラス」になります。メソッドのシグネチャ（名前・引数・戻り値の型）だけを宣言し、実装は書きません。「このメソッド群を備えたオブジェクト」という型 ＝ インターフェースを定義するための仕組みです。

from typing import Protocol

class S3Client(Protocol):
    def get_object(self, Bucket: str, Key: str) -> dict: ...
    def put_object(self, Bucket: str, Key: str, Body: bytes) -> dict: ...

メソッド本体の ...（Ellipsis オブジェクト）は「シグネチャのみ・実装なし」を表します。これを process(event, s3: S3Client) のように型注釈（type hint）として使い、「s3 には get_object / put_object を持つオブジェクトを渡す」という契約をコードに明記します。

3.2.2. Protocolの仕組み： 引数の受け口に貼る「ラベル」

Protocol は単体では何もしません。実際に効くのは、関数の引数に型注釈として貼ったときです。

def process(event: LoadEvent, s3: S3Client) -> dict:
    #                    ▲ この受け口に「S3Client の形のモノを入れて」というラベル
    obj = s3.get_object(...)

s3: S3Client は、引数 s3 の受け口に貼ったラベルにすぎません。「ここには get_object / put_object を持つモノを入れてね」という条件書きであって、それ自体は何も実行せず、何も生成しません。

では何がそのラベルを見るのか。mypy / Pyright などの型チェッカー（エディタ内蔵）が、process を呼び出している場所で照合します。

process(parsed, boto3.client("s3"))
#               ▲ ここで渡す「引数」を、受け口のラベル S3Client と突き合わせる

型チェッカーの判断はこうです。

「process の2番目の受け口は S3Client（get_object / put_object が要る）。渡された boto3.client("s3") はその形をしている？ → OK」

形の合わないモノを渡せば、実行する前に警告が出ます。

process(parsed, "ただの文字列")   # str は get_object を持たない
# → 型チェッカー: 「S3Client の受け口に str を渡している」と赤線

Protocol の仕事は「受け口にラベルを貼る」だけで、点検するのは型チェッカーが“呼び出し場所”で、という分業です。

3.2.3. ダックタイピング

この照合ルールが 構造的部分型（structural subtyping） です。継承関係を明示的に宣言した型だけを仲間とみなす公称的部分型（nominal subtyping）と異なり、必要なメソッドを備えてさえいれば、継承の有無に関わらずその型とみなす方式です。動的言語で「ダックタイピング」と呼ばれてきた考え方を、静的型付けに持ち込んだものと言えます。

アヒルのように歩き、アヒルのように鳴くなら、それはアヒルだ。

これが効くのは、自作クラスを渡すときです。

class MyS3:                     # S3Client を継承していない
    def get_object(self, Bucket, Key): ...
    def put_object(self, Bucket, Key, Body): ...

process(parsed, MyS3())         # ✅ メソッドが揃っているので型チェッカーが認める

S3Client を継承していないのに通る、これが構造的部分型の本領です。

3.2.4. Protocol：何のために使う？

Protocol は実行時に何もせず、消してもプログラムは同じように動きます。
目的は開発中の支援です。

1. 渡し間違いを実行前に検出：S3 の代わりに無関係なモノを渡すミスを、動かす前に赤線で知らせてくれる。AWS 相手だと「デプロイして初めてエラー」は高くつくので、前倒しで気づける価値は大きい。
2. コードが自己説明する：s3: S3Client と書いてあれば、読む人は「S3 っぽいモノを渡す・get_object / put_object を使う」と一目で分かる。s3（型なし）だと中身を全部読むまで分からない。
3. エディタ補完が効く：s3. まで打つと get_object などの候補が出る。
4. 依存を「最小の契約」に絞れる：S3Client は「process が使うのは get_object / put_object の2つだけ」という宣言でもある。boto3 の巨大な API 全体ではなく、必要な2メソッドにだけ依存する形になるので、テストで代役（MagicMock や自作クラス）を用意しやすく、将来ほかの実装へ差し替えるときの指針にもなる。

Protocol は「動かすために必要」ではなく「安全・読みやすさのために付ける」補助です。

4. テストを検証する道具

ここではテスト側の4つを見ていきます。
pytest を土台に、MagicMock で依存を差し替え、fixture で準備を使い回し、parametrize でパターンを回します。

4.1. pytest ─ テストの記述・収集・実行を担うテストフレームワーク

4.1.1. pytest とは

pytest は、Python のサードパーティ製テストフレームワークです。

• 命名規則に従った関数を自動収集（テストディスカバリ）して実行
  .pyファイルやその中の関数をtest_で始まる名前にしておくだけで、
  ターミナルでpytestと打つと、「このフォルダに test_*.py は無いか？その中にtest_* 関数は無いか？」と自分で探し回って、実行します。

• 素の assert 文を書き換えて失敗時に詳細な差分を表示
  Python 標準の assertでは、失敗しても情報が乏しい。
  pytest 経由で同じ assert を書くと、失敗時に中身まで教えてくれます。

使い方のルールは3つ。

1. ファイル名は test_*.py
2. 関数名は test_*
3. 中で assert 条件 を書く

# tests/test_app.py
from app import parse_event, LoadEvent

def test_parse_event_ok():
    result = parse_event({"bucket": "b", "key": "data.csv"})
    assert result == LoadEvent(bucket="b", key="data.csv")

pytest -v

4.1.2. 異常系は pytest.raises で検証

「不正な入力を渡したら、ちゃんと例外を投げてくれるか？」の検証も重要です。
これを検証するのがpytest.raisesです。

import pytest
from app import parse_event, ValidationError

def test_missing_key():
    with pytest.raises(ValidationError):     # ブロック内で例外が出れば成功
        parse_event({"bucket": "b"})

def test_message():
    with pytest.raises(ValidationError, match="必須"):   # メッセージも検証できる(メッセージに必須という文字列が含まれているか)
        parse_event({"bucket": "b"})

4.1.3. マーカー（@pytest.mark）

@pytest.mark.○○ は、テストにタグ（目印）を付ける仕組みで、マーカーと呼びます。
○○ に名前を書き、pytest の挙動を制御したりテストを分類したりします。

組み込みのマーカーには次のようなものがあります。

• @pytest.mark.skip … このテストをスキップする
• @pytest.mark.xfail … 失敗して当然（失敗しても赤にしない）
• @pytest.mark.parametrize … 値を展開してテストを増やす（後述）

自分でタグを定義することもできます。

例えばテスト工程ごとに目印を付けたいとき:

@pytest.mark.unit          # unit という自作タグを付ける
def test_parse_event_ok(): ...

@pytest.mark.integration   # こっちは結合テスト、という別タグ
def test_with_real_s3(): ...

タグを付けておくと、-m でそのタグのテストだけを選んで実行できます。

pytest -m unit          # unit タグだけ実行（単体テストだけ回す）
pytest -m "not unit"    # unit 以外だけ実行

unit のような自作マーカーは登録が必要です。
未登録だと PytestUnknownMarkWarning（タイポ防止の警告）が出ます。
pyproject.toml に宣言しておきます。

[tool.pytest.ini_options]
markers = [
    "unit: 単体テスト",
    "integration: 結合テスト",
]

4.2. MagicMock ─ 呼び出しに動的応答し記録するモックオブジェクト

4.2.1. MagicMock とは

MagicMock は、標準ライブラリ unittest.mock が提供するモックオブジェクトです。

• どんな属性アクセス・メソッド呼び出しにも動的に応答する（未定義の属性でも、自動的に子 Mock を生成して返す）
  定義していない属性・メソッドを呼び出しても、その場で自動的に“子 MagicMock”を作って返す。

• 受けた呼び出し（引数・回数）をすべて記録し、後から検証できる

4.2.2. MagicMock を使う目的

テストから本物の依存を切り離すことです。
process は S3 に依存しますが、本物の S3 を使うとテストが遅く・不安定になり、課金も発生します。
そこで本物の代役（こうした代役を総称してテストダブルと呼びます）として MagicMock を注入し、「呼んだら決まった値を返す」、「どう呼ばれたか記録する」を本物なしで再現します。結果として、ネットワークにも AWS にもつながず、テストを高速・安定・無料で回せます。

特徴1：どんなアクセスにも動的に応答する

普通のオブジェクトは、定義していない属性やメソッドに触ると AttributeError になります。

class Foo: ...
f = Foo()
f.bar          # → AttributeError（bar なんて定義してない）
f.baz()        # → AttributeError

MagicMock は、未定義の属性・メソッドを呼び出してもエラーにならず、その場で自動的に子 MagicMock を作って返します。

from unittest.mock import MagicMock

m = MagicMock()
m.bar          # → エラーにならない。新しい MagicMock が返る
m.baz()        # → 呼べる
m.a.b.c.d()    # → 何段つなげても全部OK

この挙動の目的：本物の依存を再実装せずに代役を立てられるようにすること。本物の S3 クライアントは大量のメソッドを持ちますが、それを1つも定義せずに s3.get_object(...) といきなり本物のように呼べます。偽物を手作りしてメソッドを全部書く、という手間をなくすための仕様です。

特徴2：受けた呼び出しをすべて記録する

MagicMock は「自分がどう呼ばれたか（引数・回数）」を裏で全部記録しています。

s3 = MagicMock()
s3.get_object(Bucket="my-bucket", Key="data.csv")   # 呼ぶ

s3.get_object.call_count   # → 1（呼ばれた回数）
s3.get_object.call_args    # → call(Bucket='my-bucket', Key='data.csv')（引数の記録）

この記録を使って「期待どおり呼ばれたか」を検証できます。

s3.get_object.assert_called_once_with(Bucket="my-bucket", Key="data.csv")

この記録の目的：依存が「正しく使われたか」を検証できるようにすること。テストには「戻り値が正しいか（assert result == ...）」と「依存を正しく使ったか（正しい引数で呼んだか）」の2種類の検証があり、後者はこの呼び出し記録がないと成立しません。process が正しいバケット・キーで S3 を読みにいったか、を本物なしで確かめられます。

特徴1で「本物のように振る舞う偽物」を手間なく用意でき、特徴2で「その偽物がどう扱われたか」を検証できる。前者がスタブ（値を返す代役）、後者がモック（呼ばれ方を検証する代役）の顔で、MagicMock はこの両方をこなせます。

from unittest.mock import MagicMock
from app import process, LoadEvent

def test_process_counts_lines():
    # ニセ S3 を組み立てる
    s3 = MagicMock()
    body = MagicMock()
    body.read.return_value = b"a,1\nb,2\nc,3\n"      # read() の戻り値を仕込む
    s3.get_object.return_value = {"Body": body}      # get_object() の戻り値を仕込む

    # 実行：ニセ S3 を注入
    result = process(LoadEvent("my-bucket", "data.csv"), s3)

    # 検証その1：戻り値は正しいか（状態の検証）
    assert result == {"source_key": "data.csv", "line_count": 3}

    # 検証その2：S3 を正しく呼んだか（相互作用の検証）
    s3.get_object.assert_called_once_with(Bucket="my-bucket", Key="data.csv")
    s3.put_object.assert_called_once()

.return_value は、モックが呼ばれた時に返す値を指定しています。
assert_called_once_with(...) で指定した引数で、ちょうど1回だけ呼ばれたことを確認できます。

注意：return_value を仕込み忘れたメソッドは「また別の MagicMock」を返します。obj["Body"] が意図しない値になってテストが謎の失敗をするので、使う戻り値は必ず明示的に仕込むこと。

4.2.3. 例外を起こしたいなら side_effect

「S3に接続失敗したら？」という異常系は side_effect で作ります。

def test_process_raises_when_s3_down():
    s3 = MagicMock()
    s3.get_object.side_effect = Exception("S3 接続失敗")   # 呼ぶと例外を投げる
    with pytest.raises(Exception, match="接続失敗"):
        process(LoadEvent("b", "x.csv"), s3)

4.2.4. AWS をエミュレートする moto

MagicMock は「戻り値を手で仕込む」汎用の代役です。
もう1つの選択肢が moto です。moto は AWS サービスそのものの偽物を提供する外部ライブラリで、S3・DynamoDB などをメモリ上でシミュレートします。

import boto3
from moto import mock_aws

@mock_aws                       # この中では AWS が“偽物”に差し替わる
def test_process():
    # 本物と同じ boto3 コード。でも実際は moto の偽AWSに向かう
    s3 = boto3.client("s3", region_name="ap-northeast-1")
    s3.create_bucket(Bucket="my-bucket",
                     CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"})
    s3.put_object(Bucket="my-bucket", Key="data.csv", Body=b"a\nb\nc\n")

    result = process(LoadEvent("my-bucket", "data.csv"), s3)

    assert result["line_count"] == 3

ポイントは、put_object したものが get_object で返ってくること。
moto は「状態を持つ、実際に動く AWS の代替物」です。

	MagicMock	moto
正体	汎用のニセオブジェクト（スタブ/モック）	AWS 専用の偽実装（フェイク）
戻り値	手で全部仕込む	AWS の挙動を再現
状態	持たない	持つ
向く場面	依存が少なく「正しく呼んだか」を検証したいとき	複数の AWS 操作が連携する流れを検証したいとき

今回のように S3 を1〜2回呼ぶだけなら MagicMock で十分です。
「バケット作成 → 複数ファイル書き込み → 集計」のように AWS 操作が絡み合うなら moto が自然でラク、と使い分けます。

4.3. fixture ─ テストへ依存を供給する pytest の DI 機構

4.3.1. fixture とは

fixture は、@pytest.fixture デコレーターでマークした関数で、テストの前処理やリソース（テストデータ・モック・DB接続など）を用意して供給する仕組みです。テスト関数が引数名で fixture を要求すると、pytest がその依存を解決し、fixture の戻り値を注入します。
つまり pytest 自身による依存性注入（DI）の仕組みです。

テストを複数書く際に、先述の「ニセ S3 を組み立てる5行」を毎回書くのは面倒です。
fixture に切り出せば共有できます。

import pytest
from unittest.mock import MagicMock

@pytest.fixture
def s3_with_csv():
    s3 = MagicMock()
    body = MagicMock()
    body.read.return_value = b"a,1\nb,2\nc,3\n"
    s3.get_object.return_value = {"Body": body}
    return s3                       # ← ここで返した値がテストに渡る

def test_process(s3_with_csv):      # 引数名 = fixture 名 と書くだけ
    result = process(LoadEvent("my-bucket", "data.csv"), s3_with_csv)
    assert result["line_count"] == 3

「引数に書くだけで値が入る」のは、pytest が同名の fixture を探して呼び、戻り値を注入しているから。fixture 自体が pytest の DI 機構になっています。

後片付けもできる：return の代わりに yield を使うと、yield の前が準備、後が後片付け（ティアダウン）になります。テストが失敗しても後片付けは必ず実行されるので、一時ファイルや接続の解放漏れを防げます。

@pytest.fixture
def temp_file():
    path = Path("tmp.txt"); path.write_text("data")   # 準備
    yield path                                         # テストへ渡す
    path.unlink()                                      # 後片付け

複数ファイルで共有したくなったら conftest.py に置けば import 不要で使えます。

4.4. parametrize ─ 1つのテストを複数の引数セットに展開するマーカー

4.4.1. parametrize とは

@pytest.mark.parametrize は pytest のマーカー（先述のもの）の1つです。
parametrize は、1つのテスト関数に複数の引数セットを与え、pytest がそれぞれを独立したテストケースへ展開（パラメータ化）します。

4.4.2. parametrizeが解決する問題：似たテストの量産

こういう「中身は同じで値だけ違うテスト」の量産を防ぎます。

# ダメな例：検証ロジックは全部同じで、入力と期待値だけが違う
def test_case1():
    assert parse_event({"bucket": "b", "key": "a.csv"}) == LoadEvent("b", "a.csv")

def test_case2():
    assert parse_event({"bucket": "c", "key": "y.csv"}) == LoadEvent("c", "y.csv")

parametrize なら、入力と期待値の「表」を1つ渡すだけで、テスト関数は1つに減ります。

import pytest
from app import parse_event, LoadEvent

@pytest.mark.parametrize("event, expected", [
    ({"bucket": "b", "key": "a.csv"},     LoadEvent("b", "a.csv")),   # ケース1
    ({"bucket": "c", "key": "sub/y.csv"}, LoadEvent("c", "sub/y.csv")),   # ケース2
], ids=["単純キー", "サブパス"])
def test_parse_event_ok(event, expected):
    assert parse_event(event) == expected

テスト関数は1つでも、pytest は表の行の数だけ（この例では2回）実行します。
1件落ちても他は動き、どのケースが落ちたか個別に表示されます。

test_parse_event_ok[単純キー]   ← 実行1
test_parse_event_ok[サブパス]   ← 実行2

異常系の一覧化で特に強力です。バリデーションのテストが一気に書けます。

@pytest.mark.parametrize("bad_event", [
    {}, {"bucket": "b"}, {"key": "a.csv"}, {"bucket": "b", "key": "a.txt"},
], ids=["空", "key欠落", "bucket欠落", "拡張子違い"])
def test_parse_event_rejects_invalid(bad_event):
    with pytest.raises(ValidationError):
        parse_event(bad_event)

4.4.3. parametrizeとfixture との違い

fixture と混同しやすいですが、役割は別です。

	fixture	parametrize
役割	テストに準備・道具を渡す	テストを複数の値で繰り返す
提供するもの	ニセ S3・DB接続など1つのモノ	入力と期待値の表
テストの数	増やさない	増やす（行の数だけ実行）
たとえ	料理の材料を用意する係	同じレシピを具材を変えて何回も作る

• fixture＝縦の共有：1つの準備を、複数のテストに配る
• parametrize＝横の展開：1つのテストを、複数の値に増やす

役割が別なので併用が定番です。
fixture で道具を用意しつつ、parametrize で値を振ります。

5. テストファイル全体

上記の6つが1ファイルに全部そろった状態がこれです。

# tests/test_app.py
import pytest
from unittest.mock import MagicMock

from app import parse_event, process, LoadEvent, ValidationError


# fixture：ニセ S3 の準備を共有
@pytest.fixture
def s3_with_csv():
    s3 = MagicMock()
    body = MagicMock()
    body.read.return_value = b"a,1\nb,2\nc,3\n"
    s3.get_object.return_value = {"Body": body}
    return s3


# parse_event 正常系（parametrize + dataclass の ==）
@pytest.mark.parametrize("event, expected", [
    ({"bucket": "b", "key": "a.csv"},     LoadEvent("b", "a.csv")),
    ({"bucket": "c", "key": "sub/y.csv"}, LoadEvent("c", "sub/y.csv")),
], ids=["単純キー", "サブパス"])
def test_parse_event_ok(event, expected):
    assert parse_event(event) == expected


# parse_event 異常系（parametrize + pytest.raises）
@pytest.mark.parametrize("bad_event", [
    {}, {"bucket": "b"}, {"key": "a.csv"}, {"bucket": "b", "key": "a.txt"},
], ids=["空", "key欠落", "bucket欠落", "拡張子違い"])
def test_parse_event_rejects_invalid(bad_event):
    with pytest.raises(ValidationError):
        parse_event(bad_event)


# process 正常系（fixture + MagicMock）
def test_process_counts_lines(s3_with_csv):
    result = process(LoadEvent("my-bucket", "data.csv"), s3_with_csv)
    assert result == {"source_key": "data.csv", "line_count": 3}
    s3_with_csv.get_object.assert_called_once_with(Bucket="my-bucket", Key="data.csv")
    s3_with_csv.put_object.assert_called_once()


# process 異常系（side_effect で S3 障害を再現）
def test_process_raises_when_s3_down():
    s3 = MagicMock()
    s3.get_object.side_effect = Exception("S3 接続失敗")
    with pytest.raises(Exception, match="接続失敗"):
        process(LoadEvent("b", "x.csv"), s3)

pip install pytest
pytest -v

6. おわりに（所感）

生成AI 全盛の昨今、システムのアーキテクチャや非機能要件をどう組み立てるか、生成AI をどう制御するか（どんなコンテキストを渡すかを設計する、いわゆるコンテキストエンジニアリング）、木と森に例えるなら、森の構造をデザインする領域こそ、これから重点的に身につけていかないといけないところだと感じてます。

一方で、生成AI が書いたコード、今回扱った内容は、森の喩えでいえば枝葉レベルの細部かもしれません。それでも、最終的に品質を担保し、説明責任を負うのは（今のところ）人間です。せめて自分たちが作ったもの（これから作っていくもの）については、細部までちゃんと理解しておきたいなと感じる今日この頃です。

次の記事では、森の部分（Claude Codeのコンテキストエンジニアリングとか）を学んだことをアウトプットしたいと思います。