とらすたのーと

試したこと。学んだこと。おぼえがき。

pytestでPython+SQLAlchemyのテストコードを書いてみた

概要

PythonとSQLAlchemyを使ってデータベースの操作をしているコードに対して、mock-alchemyを使ってテストコードを書く機会がありました。
その整理と備忘録を兼ねて、まとめようと思います。

やりたいこと

今回のテスト対象となるのは、データベースを操作する関数、つまりレポジトリ層の関数に対してのテストとなります。
テストの内容としては、以下のイメージです。

  • レポジトリ層の関数を呼び出した時、データベースに対して意図した操作(INSERT、SELECT、UPDATE、DELETEなど)が行われようとしているか。
  • データベース操作の結果が返ってきた時、その結果に基づいて、関数の戻り値が想定通りの内容になっているか。

つまり、関数へのインプットとアウトプットの確認となります。
整理して書くとこんな感じですね。

  • 関数呼び出し時のインプット:呼び出し方
  • 関数呼び出し時のアウトプット:データベースに対する操作
  • 関数が結果を返す時のインプット:データベースからの結果
  • 関数が結果を返す時のアウトプット:関数から呼び出し元への戻り値

尚、確認したいのは関数の動きなので、データベース自体の挙動はテストの対象外としてます。

これらを、Create(作成)、Read(読み取り)、Update(更新)、Delete(削除)の各操作で確認していきます。

事前準備

テストコードを書くためには、まずは、テストする元のコードが必要です。
ということで、以下を用意しました。

  • models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False, unique=True)

    def __init__(self, id=None, name=None, email=None):
        self.id = id
        self.name = name
        self.email = email
  • repository.py
from sqlalchemy.orm import Session
from models import User

class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def create(self, name: str, email: str):
        user = User(name=name, email=email)
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user

    def get(self, user_id: int):
        return self.db.query(User).filter(User.id == user_id).first()

    def update(self, user_id: int, name: str = None, email: str = None):
        user = self.get(user_id)
        if not user:
            return None
        if name:
            user.name = name
        if email:
            user.email = email
        self.db.commit()
        self.db.refresh(user)
        return user

    def delete(self, user_id: int):
        user = self.get(user_id)
        if not user:
            return False
        self.db.delete(user)
        self.db.commit()
        return True

今回は、CursorというAI Code Editorを使用して用意してみました。
Cursorを触ってみた感想は、また別のブログ記事にまとめようと思います。

やってみた

それでは、実際にテストコードを書いていきます。 テストコード用として下記を用意しました。

  • test_repository.py
import pytest
import mock
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, User
from repository import UserRepository
from unittest.mock import patch, MagicMock
from mock_alchemy.mocking import UnifiedAlchemyMagicMock

def get_session(data=None):
    session = UnifiedAlchemyMagicMock(data=data)
    # commit, refresh, add, delete などをMockで補う
    session.commit = MagicMock()
    session.refresh = MagicMock()
    session.add = MagicMock(side_effect=session.add)
    session.delete = MagicMock(side_effect=session.delete)
    return session

def test_create_user():
    pass

def test_get_user():
    pass

def test_update_user():
    pass

def test_delete_user():
    pass

CREATE

まずはCreate。
テスト内容としては、

  • 関数に対して登録したい値を渡した時、データベースに対してその値を登録しようとするか。
  • 関数の戻り値として、登録された結果が呼び出し元に返されるか。

となります。
実際に書いてみたのがこちら。

def test_create_user():
    # given
    session = get_session([([mock.call.query(User)], [])])
    repo = UserRepository(session)
    # when
    user = repo.create(name="Taro", email="taro@example.com")
    # then
    ## データベースに対してその値を登録しようとするか
    session.add.assert_called_once_with(user)
    called_user = session.add.call_args[0][0]
    assert called_user.name == "Taro"
    assert called_user.email == "taro@example.com"
    session.commit.assert_called_once()
    session.refresh.assert_called_once_with(user)
    ## 登録された結果が呼び出し元に返されるか
    assert user.name == "Taro"
    assert user.email == "taro@example.com"

session.addsession.commitが呼ばれているか、session.addをどんな値で呼んでいるか、を確認することで、どのように登録しようとしているかをテストしています。

READ

次はRead。
テスト内容としては、

  • 関数に対して取得したい値(条件)を渡した時、データベースに対してその値で取得しようとするか。
  • 関数の戻り値として、取得された結果が呼び出し元に返されるか。

となります。
実際に書いてみたのがこちら。

def test_get_user():
    # given: モックのセッションとクエリチェーンを用意
    session = MagicMock()
    mock_query = session.query.return_value
    mock_filter = mock_query.filter.return_value
    mock_filter.first.return_value = User(id=1, name="Jiro", email="jiro@example.com")
    repo = UserRepository(session)
    # when: getを呼び出す
    user = repo.get(1)
    # then
    ## データベースに対してその値で取得しようとするか
    ### query, filter, firstが呼ばれていること
    session.query.assert_called_once_with(User)
    ### filterの条件が正しいことをテストする
    args, kwargs = mock_query.filter.call_args
    assert len(args) == 1
    ### User.id == 1 という比較式になっているか確認
    condition = args[0]
    assert str(condition) == str((User.id == 1))
    mock_filter.first.assert_called_once()
    ## 取得された結果が呼び出し元に返されるか
    assert isinstance(user, User)
    assert user.name == "Jiro"
    assert user.email == "jiro@example.com"

取得対象、条件、取得件数を確認することで、どのように取得しようとしているかをテストしています。

UPDATE

次はUpdate。
テスト内容としては、

  • 関数に対して更新したい値を渡した時、データベースに対してその値で更新しようとするか。
  • 関数の戻り値として、更新された結果が呼び出し元に返されるか。

となります。
実際に書いてみたのがこちら。

def test_update_user():
    # given: モックのセッションとgetの返り値
    session = MagicMock()
    user_obj = User(id=2, name="Saburo", email="saburo@example.com")
    repo = UserRepository(session)
    repo.get = MagicMock(return_value=user_obj)
    # when: updateを呼び出す
    updated = repo.update(2, name="326", email="saburo2@example.com")
    # then
    ## データベースに対してその値で更新しようとするか
    ### user_objの値が更新されていること
    assert user_obj.id == 2
    assert user_obj.name == "326"
    assert user_obj.email == "saburo2@example.com"
    ### commit, refreshが呼ばれていること
    session.commit.assert_called_once()
    session.refresh.assert_called_once_with(user_obj)
    ## 更新された結果が呼び出し元に返されるか。
    assert updated.name == "326"
    assert updated.email == "saburo2@example.com"

どの値でcommitするか確認することで、どのように更新しようとしているかテストしています。

DELETE

次はDelete。
テスト内容としては、

  • 関数に対して削除したい値(条件)を渡した時、データベースに対してその値で削除しようとするか。
  • 関数の戻り値として、削除された結果が呼び出し元に返されるか。

となります。
実際に書いてみたのがこちら。

def test_delete_user():
    # given: getがUserを返すようにモック
    session = MagicMock()
    user_obj = User(id=3, name="Shiro", email="shiro@example.com")
    repo = UserRepository(session)
    repo.get = MagicMock(return_value=user_obj)
    # when: deleteを呼び出す
    result = repo.delete(3)
    # then
    ## データベースに対してその値で削除しようとするか。 
    ### delete, commitが呼ばれていること
    session.delete.assert_called_once_with(user_obj)
    # deleteがどのオブジェクトに対して呼ばれたか(条件)をテスト
    args, kwargs = session.delete.call_args
    assert args[0].id == 3
    session.commit.assert_called_once()
    ## 削除された結果が呼び出し元に返されるか。
    assert result is True

session.deletesession.commitが呼ばれてることを確認することで、どのように削除しようとしているかをテストしています。

まとめ

ひと通り、CRUD操作に対するテストを行うことは出来ました。ただ、この確認観点だと記述量が多くなるなというのが正直な感想です。
今回は、mock-alchemyを使ってみましたが、色々調べてみると、SQLiteを使う方法もあるそうです。
(っていうかそっちのほうが一般的なのかな?)
次回は、SQLiteを使ってテストコードを書く方法を試してみたいと思います。