pytestとmotoで始めるLambda単体テスト入門

introduction-python-lambda-testing/
├── chapter1/   # テスタブルな設計の基本（カバレッジを含む）
│   ├── pyproject.toml
│   ├── src/
│   │   └── order_handler.py
│   └── tests/
│       ├── conftest.py
│       └── test_order_handler.py
├── chapter2/   # 現実的なファイル分割（全層DI・モック）
├── chapter3/   # リポジトリ層をmotoでテストする
├── chapter4/   # エラーハンドリングとバリデーションのテスト
└── chapter5/   # 外部API呼び出しのモック

[project]
name = "chapter1"
version = "1.0.0"
requires-python = ">=3.13"

[dependency-groups]
dev = [
    "boto3>=1.42.83",
    "moto[dynamodb]>=5.1.22",
    "pytest>=9.0.2",
    "pytest-cov>=7.1.0",
]

[tool.pytest]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = ["--cov=src", "--cov-report=term-missing"]

$ uv sync --dev

import boto3
import os

# モジュールレベルで初期化（コールドスタート最適化）
_dynamodb = boto3.resource("dynamodb")
_TABLE_NAME = os.environ["TABLE_NAME"]

def handler(event, context):
    items = event.get("items", [])

    # ビジネスロジックとAWS操作が混在している
    subtotal = sum(item["price"] * item["quantity"] for item in items)
    total = int(subtotal * 0.9) if subtotal >= 10000 else subtotal

    table = _dynamodb.Table("orders")
    table.put_item(Item={"order_id": event["order_id"], "items": items, "total": total})

    return {"statusCode": 200, "body": {"total": total}}

import boto3
import os

# モジュールレベルで初期化（コールドスタート最適化）
_dynamodb = boto3.resource("dynamodb")
_TABLE_NAME = os.environ["TABLE_NAME"]

def calculate_total(items: list[dict]) -> int:
    """注文明細から合計金額を計算する（10,000円以上は10%割引）"""
    subtotal = sum(item["price"] * item["quantity"] for item in items)
    if subtotal >= 10000:
        return int(subtotal * 0.9)
    return subtotal

def save_order(order_id: str, items: list[dict], total: int) -> None:
    table = _dynamodb.Table(_TABLE_NAME)
    table.put_item(Item={"order_id": order_id, "items": items, "total": total})

def handler(event, context):
    items = event.get("items", [])
    total = calculate_total(items)
    save_order(event["order_id"], items, total)
    return {"statusCode": 200, "body": {"data": {"total": total}}}

@pytest.fixture
def dynamodb_table():
    # ① セットアップ: テーブルを作成
    ...
    yield dynamodb   # ② テスト関数に渡す
    # ③ クリーンアップ: yieldの後が自動実行される（mock_aws()のwithブロックを抜ける）

def test_save_order(dynamodb_table):  # ← 引数に書くだけで自動注入される
    ...

import os

import boto3
import pytest
from moto import mock_aws

os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
os.environ["AWS_DEFAULT_REGION"] = "ap-northeast-1"
os.environ["TABLE_NAME"] = "orders"

@pytest.fixture
def dynamodb_table():
    with mock_aws():
        dynamodb = boto3.resource("dynamodb", region_name="ap-northeast-1")
        dynamodb.create_table(
            TableName="orders",
            KeySchema=[{"AttributeName": "order_id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "order_id", "AttributeType": "S"}],
            BillingMode="PAY_PER_REQUEST",
        )
        yield dynamodb

from order_handler import calculate_total, save_order, handler

def test_empty_items():
    # 明細が空のとき合計0円になること
    assert calculate_total([]) == 0

def test_no_discount_below_threshold():
    items = [{"price": 3000, "quantity": 2}]  # 合計6,000円 → 割引なし
    assert calculate_total(items) == 6000

def test_exact_threshold():
    items = [{"price": 10000, "quantity": 1}]  # ちょうど10,000円 → 10%割引が適用される
    assert calculate_total(items) == 9000

def test_discount_applied_at_threshold():
    items = [{"price": 6000, "quantity": 2}]  # 合計12,000円 → 10%割引
    assert calculate_total(items) == 10800

def test_multiple_items():
    items = [
        {"price": 2000, "quantity": 3},  # 6,000円
        {"price": 1000, "quantity": 5},  # 5,000円
    ]
    # 合計11,000円 → 10%割引 → 9,900円
    assert calculate_total(items) == 9900

def test_calculate_total_truncation():
    """割引後の端数は切り捨てられる"""
    items = [{"price": 10001, "quantity": 1}]
    assert calculate_total(items) == 9000

def test_save_order(dynamodb_table):
    items = [{"price": 3000, "quantity": 3}]
    save_order("order-001", items, 9000)

    table = dynamodb_table.Table("orders")
    response = table.get_item(Key={"order_id": "order-001"})
    assert response["Item"]["items"] == items
    assert response["Item"]["total"] == 9000

def test_handler(dynamodb_table):
    event = {
        "order_id": "order-001",
        "items": [{"price": 5000, "quantity": 3}],  # 合計15,000円 → 10%割引
    }
    result = handler(event, {})
    assert result["statusCode"] == 200
    assert result["body"]["data"]["total"] == 13500

$ uv run pytest -v

============================ tests coverage ============================
__________ coverage: platform darwin, python 3.13.12-final-0 ___________

Name                   Stmts   Miss  Cover   Missing
----------------------------------------------------
src/order_handler.py      16      0   100%
----------------------------------------------------
TOTAL                     16      0   100%
========================== 8 passed in 0.40s ===========================

src/
├── controller.py                        # eventを受け取り、usecaseを呼び出す
└── app/
    ├── dependencies.py                  # AWSクライアントの初期化と提供
    ├── models/order.py                  # データモデル（dataclass）
    ├── repositories/order_repository.py # DynamoDBアクセス
    ├── services/order_service.py        # ビジネスロジック（純粋関数）
    └── usecases/order_usecase.py        # ユースケース（サービスとリポジトリの組み合わせ）

from dataclasses import dataclass

@dataclass
class OrderItem:
    price: int
    quantity: int

@dataclass
class Order:
    order_id: str
    items: list[OrderItem]
    total: int = 0

from app.models.order import OrderItem

def calculate_total(items: list[OrderItem]) -> int:
    """10,000円以上は10%割引"""
    subtotal = sum(item.price * item.quantity for item in items)
    if subtotal >= 10000:
        return round(subtotal * 0.9)
    return subtotal

from app.models.order import OrderItem
from app.services.order_service import calculate_total

def test_calculate_total_no_discount():
    items = [OrderItem(price=3000, quantity=2)]
    assert calculate_total(items) == 6000

def test_calculate_total_exact_threshold():
    items = [OrderItem(price=10000, quantity=1)]
    assert calculate_total(items) == 9000

def test_calculate_total_with_discount():
    items = [OrderItem(price=6000, quantity=2)]
    assert calculate_total(items) == 10800

def test_calculate_total_rounding():
    """割引後の端数は切り捨てでなく四捨五入される"""
    items = [OrderItem(price=10001, quantity=1)]
    assert calculate_total(items) == 9001  # int()なら9000になる

from app.models.order import Order

class OrderRepository:
    def __init__(self, dynamodb, table_name: str):
        self._table = dynamodb.Table(table_name)

    def save(self, order: Order) -> None:
        self._table.put_item(Item={
            "order_id": order.order_id,
            "items": [{"price": item.price, "quantity": item.quantity} for item in order.items],
            "total": order.total,
        })

from unittest.mock import MagicMock

from app.models.order import Order, OrderItem
from app.repositories.order_repository import OrderRepository

def test_save():
    mock_dynamodb = MagicMock()
    mock_table = MagicMock()
    mock_dynamodb.Table.return_value = mock_table

    items = [OrderItem(price=3000, quantity=3)]
    order = Order(order_id="order-001", items=items, total=9000)
    repo = OrderRepository(mock_dynamodb, "orders")
    repo.save(order)

    mock_dynamodb.Table.assert_called_once_with("orders")
    mock_table.put_item.assert_called_once_with(Item={
        "order_id": "order-001",
        "items": [{"price": 3000, "quantity": 3}],
        "total": 9000,
    })

from app.models.order import Order, OrderItem
from app.repositories.order_repository import OrderRepository
from app.services.order_service import calculate_total

def create_order(items: list[OrderItem], order_id: str, repository: OrderRepository) -> Order:
    total = calculate_total(items)
    order = Order(order_id=order_id, items=items, total=total)
    repository.save(order)
    return order

from unittest.mock import MagicMock

from app.models.order import Order, OrderItem
from app.usecases.order_usecase import create_order

def test_create_order():
    mock_repo = MagicMock()
    order = create_order([OrderItem(price=5000, quantity=3)], "order-001", mock_repo)
    assert order.total == 13500
    assert order.order_id == "order-001"
    expected_order = Order(order_id="order-001", items=[OrderItem(price=5000, quantity=3)], total=13500)
    mock_repo.save.assert_called_once_with(expected_order)

import os
from functools import cache

import boto3
from app.repositories.order_repository import OrderRepository

@cache
def get_order_repository() -> OrderRepository:
    return OrderRepository(boto3.resource("dynamodb"), os.environ["TABLE_NAME"])

from app.dependencies import get_order_repository

def test_get_order_repository_returns_same_instance():
    repo1 = get_order_repository()
    repo2 = get_order_repository()
    assert repo1 is repo2

from app.dependencies import get_order_repository
from app.models.order import OrderItem
from app.usecases.order_usecase import create_order

def handler(event, context):
    order_id = event["order_id"]
    items = [OrderItem(**item) for item in event.get("items", [])]
    order = create_order(items, order_id, get_order_repository())
    return {"statusCode": 200, "body": {"data": {"total": order.total}}}

from unittest.mock import MagicMock, patch

from controller import handler

def test_handler():
    mock_repo = MagicMock()
    with patch("controller.get_order_repository", return_value=mock_repo):
        event = {"order_id": "order-001", "items": [{"price": 5000, "quantity": 3}]}
        result = handler(event, {})
    assert result["statusCode"] == 200
    assert result["body"]["data"]["total"] == 13500

import os

import pytest

os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
os.environ["AWS_DEFAULT_REGION"] = "ap-northeast-1"
os.environ["TABLE_NAME"] = "orders"

@pytest.fixture(autouse=True)
def _clear_dependency_cache():
    from app.dependencies import get_order_repository
    get_order_repository.cache_clear()
    yield
    get_order_repository.cache_clear()

def test_save():
    mock_dynamodb = MagicMock()
    mock_table = MagicMock()
    mock_dynamodb.Table.return_value = mock_table

    items = [OrderItem(price=3000, quantity=3)]
    order = Order(order_id="order-001", items=items, total=9000)
    repo = OrderRepository(mock_dynamodb, "orders")
    repo.save(order)

    mock_table.put_item.assert_called_once_with(Item={
        "order_id": "order-001",
        "items": [{"price": 3000, "quantity": 3}],
        "total": 9000,
    })

# motoを使うと、モック設定なしでそのまま操作・検証できる
def test_save(dynamodb_table):
    repo = OrderRepository(dynamodb_table, "orders")
    order = Order(order_id="order-001", items=[OrderItem(price=3000, quantity=3)], total=9000)
    repo.save(order)
    response = dynamodb_table.Table("orders").get_item(Key={"order_id": "order-001"})
    assert response["Item"]["total"] == 9000

import os

import boto3
import pytest
from moto import mock_aws

os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
os.environ["AWS_DEFAULT_REGION"] = "ap-northeast-1"
os.environ["TABLE_NAME"] = "orders"

@pytest.fixture(autouse=True)
def _clear_dependency_cache():
    from app.dependencies import get_order_repository
    get_order_repository.cache_clear()
    yield
    get_order_repository.cache_clear()

@pytest.fixture
def dynamodb_table():
    with mock_aws():
        dynamodb = boto3.resource("dynamodb", region_name="ap-northeast-1")
        dynamodb.create_table(
            TableName="orders",
            KeySchema=[{"AttributeName": "order_id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "order_id", "AttributeType": "S"}],
            BillingMode="PAY_PER_REQUEST",
        )
        yield dynamodb

from app.models.order import Order, OrderItem
from app.repositories.order_repository import OrderRepository

def test_save(dynamodb_table):
    repo = OrderRepository(dynamodb_table, "orders")
    order = Order(order_id="order-001", items=[OrderItem(price=3000, quantity=3)], total=9000)
    repo.save(order)
    table = dynamodb_table.Table("orders")
    response = table.get_item(Key={"order_id": "order-001"})
    assert response["Item"]["items"] == [{"price": 3000, "quantity": 3}]
    assert response["Item"]["total"] == 9000

src/
├── controller.py
└── app/
    ├── dependencies.py
    ├── exceptions.py                    # ドメイン例外の定義
    ├── inputs/order_input.py            # イベントのパースとバリデーション
    ├── models/order.py
    ├── repositories/order_repository.py
    ├── services/order_service.py
    └── usecases/order_usecase.py

class OrderError(Exception):
    """注文ドメインの基底例外"""

class OrderSaveError(OrderError):
    """注文の保存に失敗"""

class OrderItemValidationError(OrderError):
    """注文アイテムの入力値が不正"""

from app.exceptions import OrderItemValidationError
from app.models.order import OrderItem

def parse_order_items_from_event(data: object) -> list[OrderItem]:
    if not isinstance(data, list):
        raise OrderItemValidationError("items はリストである必要があります")
    return [parse_order_item(item) for item in data]

def parse_order_item(data: dict[str, object]) -> OrderItem:
    item = _parse(data)
    validate_order_item(item)
    return item

def _parse(data: dict[str, object]) -> OrderItem:
    try:
        return OrderItem(**data)
    except TypeError as e:
        raise OrderItemValidationError(f"不正なアイテムデータ: {e}") from e

def validate_order_item(item: OrderItem) -> None:
    if item.price <= 0 or item.quantity <= 0:
        raise OrderItemValidationError("price と quantity は1以上である必要があります")

from botocore.exceptions import ClientError

from app.exceptions import OrderSaveError
from app.models.order import Order

class OrderRepository:
    def __init__(self, dynamodb, table_name: str):
        self._table = dynamodb.Table(table_name)

    def save(self, order: Order) -> None:
        try:
            self._table.put_item(Item={
                "order_id": order.order_id,
                "items": [{"price": item.price, "quantity": item.quantity} for item in order.items],
                "total": order.total,
            })
        except ClientError as e:
            raise OrderSaveError(f"注文の保存に失敗しました: {e}") from e

from app.dependencies import get_order_repository
from app.exceptions import OrderItemValidationError, OrderSaveError
from app.inputs.order_input import parse_order_items_from_event
from app.usecases.order_usecase import create_order

def _ok(data: dict[str, object]) -> dict[str, object]:
    return {"statusCode": 200, "body": {"data": data}}

def _err(status_code: int, message: str) -> dict[str, object]:
    return {"statusCode": status_code, "body": {"error": message}}

def handler(event: dict[str, object], context: object) -> dict[str, object]:
    repo = get_order_repository()

    try:
        items = parse_order_items_from_event(event.get("items", []))
        order = create_order(items, event["order_id"], repo)
        return _ok({"total": order.total})

    except OrderSaveError as e:
        return _err(500, str(e))
    except OrderItemValidationError as e:
        return _err(400, str(e))
    except KeyError as e:
        return _err(400, f"必須パラメータが不足: {e}")

import pytest
from unittest.mock import MagicMock
from botocore.exceptions import ClientError

from app.exceptions import OrderSaveError
from app.models.order import Order, OrderItem
from app.repositories.order_repository import OrderRepository

def test_save(dynamodb_table):
    repo = OrderRepository(dynamodb_table, "orders")
    order = Order(order_id="order-001", items=[OrderItem(price=3000, quantity=3)], total=9000)
    repo.save(order)
    table = dynamodb_table.Table("orders")
    response = table.get_item(Key={"order_id": "order-001"})
    assert response["Item"]["total"] == 9000
    assert response["Item"]["items"] == [{"price": 3000, "quantity": 3}]

def test_save_raises_order_save_error_on_dynamodb_failure():
    mock_dynamodb = MagicMock()
    mock_table = MagicMock()
    mock_dynamodb.Table.return_value = mock_table
    mock_table.put_item.side_effect = ClientError(
        {"Error": {"Code": "ProvisionedThroughputExceededException", "Message": "スループット超過"}},
        "PutItem",
    )
    repo = OrderRepository(mock_dynamodb, "orders")
    order = Order(order_id="order-001", items=[OrderItem(price=3000, quantity=3)], total=9000)
    with pytest.raises(OrderSaveError, match="注文の保存に失敗しました"):
        repo.save(order)

import pytest
from unittest.mock import MagicMock

from app.exceptions import OrderSaveError
from app.models.order import OrderItem
from app.usecases.order_usecase import create_order

def test_create_order():
    mock_repo = MagicMock()
    order = create_order([OrderItem(price=5000, quantity=3)], "order-001", mock_repo)
    assert order.total == 13500
    mock_repo.save.assert_called_once_with(order)

def test_create_order_propagates_save_error():
    mock_repo = MagicMock()
    mock_repo.save.side_effect = OrderSaveError("保存失敗")
    with pytest.raises(OrderSaveError):
        create_order([OrderItem(price=1000, quantity=1)], "order-001", mock_repo)

[project]
dependencies = ["requests>=2.33.1"]

[dependency-groups]
dev = [
    "pytest>=9.0.2",
    "pytest-cov>=7.1.0",
    "responses>=0.26.0",  # HTTPモック用
]

class PaymentError(Exception):
    pass

class PaymentTimeoutError(PaymentError):
    pass

class PaymentAPIError(PaymentError):
    def __init__(self, status_code: int) -> None:
        self.status_code = status_code
        super().__init__(f"Payment API error: {status_code}")

from typing import Any

from app.dependencies import get_payment_gateway
from app.exceptions import PaymentAPIError, PaymentTimeoutError
from app.usecases.payment_usecase import process_payment

def handler(event: dict[str, Any], context: object) -> dict[str, Any]:
    try:
        order_id = event["order_id"]
        amount = event["amount"]
        result = process_payment(order_id, amount, get_payment_gateway())
        return {"statusCode": 200, "body": {"data": result}}
    except PaymentTimeoutError:
        return {"statusCode": 504, "body": {"error": "決済APIがタイムアウトしました"}}
    except PaymentAPIError as e:
        return {"statusCode": e.status_code, "body": {"error": "決済APIエラー"}}
    except KeyError as e:
        return {"statusCode": 400, "body": {"error": f"必須パラメータが不足: {e}"}}

import os

import pytest

PAYMENT_API_URL = "https://api.example.com/payments"

os.environ["PAYMENT_API_URL"] = PAYMENT_API_URL
os.environ["PAYMENT_API_TIMEOUT"] = "5"

@pytest.fixture(autouse=True)
def _clear_dependency_cache():
    from app.dependencies import get_payment_gateway
    get_payment_gateway.cache_clear()
    yield
    get_payment_gateway.cache_clear()

from unittest.mock import MagicMock

from app.usecases.payment_usecase import process_payment

def test_process_payment_success():
    """正常系: gatewayの結果をそのまま返す"""
    gw = MagicMock()
    gw.process.return_value = {"transaction_id": "txn-001", "status": "completed"}
    result = process_payment("order-001", 9000, gw)
    assert result["transaction_id"] == "txn-001"
    gw.process.assert_called_once_with("order-001", 9000)

from unittest.mock import MagicMock, patch

from app.exceptions import PaymentAPIError, PaymentTimeoutError
from controller import handler

_GET_GATEWAY = "controller.get_payment_gateway"

def _mock_gateway(side_effect=None, return_value=None):
    gw = MagicMock()
    if side_effect is not None:
        gw.process.side_effect = side_effect
    else:
        gw.process.return_value = return_value
    return gw

def test_handler_success():
    gw = _mock_gateway(return_value={"transaction_id": "txn-001", "status": "completed"})
    with patch(_GET_GATEWAY, return_value=gw):
        result = handler({"order_id": "order-001", "amount": 9000}, {})
    assert result["statusCode"] == 200
    assert result["body"]["data"]["transaction_id"] == "txn-001"

def test_handler_api_500():
    gw = _mock_gateway(side_effect=PaymentAPIError(500))
    with patch(_GET_GATEWAY, return_value=gw):
        result = handler({"order_id": "order-001", "amount": 9000}, {})
    assert result["statusCode"] == 500
    assert "決済APIエラー" in result["body"]["error"]

def test_handler_timeout():
    gw = _mock_gateway(side_effect=PaymentTimeoutError())
    with patch(_GET_GATEWAY, return_value=gw):
        result = handler({"order_id": "order-001", "amount": 9000}, {})
    assert result["statusCode"] == 504
    assert "タイムアウト" in result["body"]["error"]

def test_handler_missing_order_id():
    result = handler({"amount": 9000}, {})
    assert result["statusCode"] == 400
    assert "必須パラメータが不足" in result["body"]["error"]