DynamoDB localを使ったpytest

この記事は「もくもく会ブログリレー」 26 日目 の記事です!

以前、参加した「開発生産性Conference 2024」の和田卓人氏による「開発生産性の観点から考える自動テスト」の基調講演内で、DynamoDB localを使ってコンテナで動作させることができればテストサイズを小さくできるといったお話がありました。
元々DynamoDB localについて聞いたことがありましたが、実際に触ったことがなかったので、講演をきっかけに実際に触ってみました。
今回は、ローカル環境でpytestを実行するところまで試してみた手順をまとめてみました。

【開発生産性Conference 2024】基調講演「開発生産性の観点から考える自動テスト」を聴いてきました！

DynamoDB localとは

ダウンロード可能なバージョンの Amazon DynamoDB では、DynamoDB ウェブサービスにアクセスせずに、アプリケーションを開発してテストすることができます。代わりに、データベースはコンピュータ上で自己完結型となります。アプリケーションを本番稼働環境にデプロイする準備ができたら、コード内のローカルエンドポイントを削除します。その後、これは DynamoDB ウェブサービスを指します。

このローカルバージョンを使用することで、スループットやデータストレージ、データ転送料金を節約しやすくなります。また、アプリケーションを開発している間インターネットに接続しておく必要はありません。

https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/DynamoDBLocal.html

料金を節約しながら開発ができるなんて素敵ですね。ローカルで開発中に無限ループで知らないうちに料金が跳ね上がっていたなんてことも防げそうですね。
また、開発者のコンピューター上で開発を完結できるので、開発環境のデータベースが汚れないのも素敵です。

コンテナ（DynamoDB localなど）の設定

今回は、DynamoDB localを使ってバックエンドアプリケーションを開発するためのローカル開発環境を構築していきます。
VSCodeのDev Containersを使って環境を作っていきます。

.devcontainer/docker-compose.ymlを作成して下記を追加します。
アプリケーションとDynamoDB localのコンテナの設定をしています。

version: '3.8'
services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
    tty: true
    ports:
      - '9000:9000'
    working_dir: /workspace
    volumes:
      - ..:/workspace
    depends_on:
      - 'dynamodb'
    links:
      - 'dynamodb'
    environment:
      AWS_ACCESS_KEY_ID: 'DUMMYIDEXAMPLE' # DynamoDB localへアクセスするために必要。値はダミーのものでOK
      AWS_SECRET_ACCESS_KEY: 'DUMMYEXAMPLEKEY' # DynamoDB localへアクセスするために必要。値はダミーのものでOK
      AWS_DEFAULT_REGION: 'ap-northeast-1'

  dynamodb:
    command: '-jar DynamoDBLocal.jar -sharedDb -dbPath ./data'
    image: 'amazon/dynamodb-local:latest'
    ports:
      - '8000:8000'
    volumes:
      - './docker/dynamodb:/home/dynamodblocal/data'
    working_dir: /home/dynamodblocal

.devcontainer/Dockerfileを作成して下記を追加します。
pytyon3.12の環境のコンテナの設定をしていきます。

FROM python:3.12

ARG USERNAME=vscode
ARG USER_UID=1000
ARG USER_GID=$USER_UID

RUN groupadd --gid $USER_GID $USERNAME \
    && useradd -s /bin/bash --uid $USER_UID --gid $USER_GID -m $USERNAME \
    && apt-get update \
    && apt-get install -y sudo \
    && echo $USERNAME ALL=\(root\) NOPASSWD:ALL > /etc/sudoers.d/$USERNAME\
    && chmod 0440 /etc/sudoers.d/$USERNAME \
    && apt-get clean

.devcontainer/devcontainer.json

{
  "name": "Python Development",
  "dockerComposeFile": "docker-compose.yml",
  "workspaceFolder": "/workspace",
  "remoteUser": "vscode",
  "service": "app",
  "features": {
    "ghcr.io/devcontainers/features/aws-cli:1": {}
  }
}

ファイルがそれぞれ作成できたら、VSCodeの拡張機能のDev Containersでコンテナを起動します。

必要なライブラリのインストール

requirements.txtに下記を貼り付けてpip install -r requirements.txtでライブラリをインストールします。

fastapi==0.111.1
pytest==8.3.2
pytest-mock==3.14.0
pynamodb==6.0.1
pydantic==2.8.2
pydantic_core==2.20.1

DynamoDB localのテーブル作成

今回は1つのタスクテーブルのみ使用するので、下記をtask_table.jsonに貼り付けます。

{
    "TableName": "tasks",
    "KeySchema": [
      {
        "AttributeName": "task_id",
        "KeyType": "HASH"
      }
    ],
    "AttributeDefinitions": [
      {
        "AttributeName": "task_id",
        "AttributeType": "S"
      }
    ],
    "ProvisionedThroughput": {
      "ReadCapacityUnits": 1,
      "WriteCapacityUnits": 1
    }
  }

aws dynamodb create-table --cli-input-json file://task_table.json --endpoint-url http://dynamodb:8000
をターミナルで実行し、テーブルを作成します。
AWS環境のDynamoDBに対して操作するのと同じように、AWSのCLIでDynamoDB localに対して操作が可能です。

Web APIの実装

タスクの登録とタスク情報の取得ができる簡単なWeb APIを実装します。
後ほど、こちらのWeb APIに対してテストを書いていきます。

今回は、PythonのフレームワークであるFastAPIを用いてmain.pyに実装していきます。

import shortuuid

from fastapi import FastAPI, HTTPException
from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, BooleanAttribute
from pydantic import BaseModel


app = FastAPI()


class TaskModel(Model):
    """tasksテーブルのモデル"""

    class Meta:
        table_name = "tasks"
        host = "http://dynamodb:8000"

    task_id = UnicodeAttribute(hash_key=True)
    name = UnicodeAttribute()
    is_done = BooleanAttribute()


class Task(BaseModel):
    """タスクのスキーマ"""

    name: str
    is_done: bool


@app.post("/tasks/")
def create_task(task: Task):
    """タスクを作成する"""
    item = TaskModel(task_id=shortuuid.uuid(), name=task.name, is_done=task.is_done)
    item.save()
    return {"task_id": item.task_id, "name": item.name, "is_done": item.is_done}


@app.get("/tasks/{task_id}")
def read_task(task_id: str):
    """指定されたIDのタスクを取得する"""
    try:
        item = TaskModel.get(task_id)
    except TaskModel.DoesNotExist:
        raise HTTPException(status_code=404, detail="Task not found")

    return {"task_id": item.task_id, "name": item.name, "is_done": item.is_done}

下記の部分で、DynamoDB localにアクセスできるようにhostを指定しています。
AWS環境のDynamoDBにアクセスする際は不要なので、ローカル環境とその他の環境で動的に切り替わるようにしておくのが良いと思います。

class TaskModel(Model):
    """tasksテーブルのモデル"""

    class Meta:
        table_name = "tasks"
        host = "http://dynamodb:8000"

    task_id = UnicodeAttribute(hash_key=True)
    name = UnicodeAttribute()
    is_done = BooleanAttribute()

テストコードの記述

アプリケーションに対してのテストコードの全体が下記になります。
test_main.pyを作成して、記述していきます。

import pytest

from fastapi.testclient import TestClient
from main import app, TaskModel

client = TestClient(app)


@pytest.fixture()
def register_task():
    # テスト用のタスクを登録する
    task = TaskModel(
        task_id="kHyinuweTy5eoGYV7HpSZY", name="ブログを書く", is_done=False
    )
    task.save()
    yield
    task.delete()


@pytest.fixture()
def mock_shortuuid(mocker):
    # shortuuid.uuid()の返り値を固定する
    def mock_shortuuid():
        return "beerbeerbeerbeerbeer"

    mocker.patch("shortuuid.uuid", mock_shortuuid)


def test_read_task_指定されたIDのタスクが存在する場合にタスクの内容が返される(
    register_task,
):
    response = client.get("/tasks/kHyinuweTy5eoGYV7HpSZY")
    assert response.status_code == 200
    assert response.json() == {
        "task_id": "kHyinuweTy5eoGYV7HpSZY",
        "name": "ブログを書く",
        "is_done": False,
    }


def test_read_task_指定されたIDのタスクが存在しない場合に404エラーが返される(
    register_task,
):
    response = client.get("/tasks/saunasaunasaunasauna")
    assert response.status_code == 404
    assert response.json() == {"detail": "Task not found"}


def test_create_task_タスクが作成される(register_task, mock_shortuuid):
    response = client.post("/tasks/", json={"name": "PRのレビュー", "is_done": False})
    print(response.json())
    assert response.status_code == 200
    assert response.json() == {
        "task_id": "beerbeerbeerbeerbeer",
        "name": "PRのレビュー",
        "is_done": False,
    }

fixtureで各テストケースの実行前にDynamoDB localのテーブルにデータを登録し、テストケースの実行が完了するとデータを削除するようにして他のテストケースに影響が出ないようにしています。

@pytest.fixture()
def register_task():
    # テスト用のタスクを登録する
    task = TaskModel(
        task_id="kHyinuweTy5eoGYV7HpSZY", name="ブログを書く", is_done=False
    )
    task.save()
    yield
    task.delete()

pytestを実行すると全て成功します。

※VSCodeの拡張機能であるPython Test Explorer for Visual Studio Codeを使ってpytestを実行しています。

テスト用にタスクを登録しているfixtureのregister_taskメソッド内のtask.save()をコメントアウトしてテストを実行すると、以下のテストケースが失敗します。
DynamoDB localにデータがない状態で、データを取得しようとして404が返ってきているためですね。

最後に

初めてDynamoDB localを試してみましたが、DynamoDBと同じように扱えるのでとても良いなと感じました。
今までDynamoDB localを使わずに、DynamoDBのテーブルに対してデータを取得したり更新したりする処理の部分に対してモックを用意していましたが、規模が大きくなってくるとモックを修正しているのかテストを修正しているのかわからなくなってくる時があったので、そうならないようにするためにも効果的かなと思います。

明日の記事は、江塚さんの「AWS GameDay ~Secure Legends~ イベントレポート」です!