概要
PythonとSQLAlchemyを使ってデータベースの操作をしているコードに対して、mock-alchemyを使ってテストコードを書く機会がありました。
その整理と備忘録を兼ねて、まとめようと思います。
やりたいこと
今回のテスト対象となるのは、データベースを操作する関数、つまりレポジトリ層の関数に対してのテストとなります。
テストの内容としては、以下のイメージです。
- レポジトリ層の関数を呼び出した時、データベースに対して意図した操作（INSERT、SELECT、UPDATE、DELETEなど）が行われようとしているか。
- データベース操作の結果が返ってきた時、その結果に基づいて、関数の戻り値が想定通りの内容になっているか。
つまり、関数へのインプットとアウトプットの確認となります。
整理して書くとこんな感じですね。
- 関数呼び出し時のインプット：呼び出し方
- 関数呼び出し時のアウトプット：データベースに対する操作
- 関数が結果を返す時のインプット：データベースからの結果
- 関数が結果を返す時のアウトプット：関数から呼び出し元への戻り値
尚、確認したいのは関数の動きなので、データベース自体の挙動はテストの対象外としてます。
これらを、Create(作成)、Read(読み取り)、Update(更新)、Delete(削除)の各操作で確認していきます。
事前準備
テストコードを書くためには、まずは、テストする元のコードが必要です。
ということで、以下を用意しました。
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
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を触ってみた感想は、また別のブログ記事にまとめようと思います。
やってみた
それでは、実際にテストコードを書いていきます。
テストコード用として下記を用意しました。
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)
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():
session = get_session([([mock.call.query(User)], [])])
repo = UserRepository(session)
user = repo.create(name="Taro", email="taro@example.com")
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.add、session.commitが呼ばれているか、session.addをどんな値で呼んでいるか、を確認することで、どのように登録しようとしているかをテストしています。
READ
次はRead。
テスト内容としては、
- 関数に対して取得したい値(条件)を渡した時、データベースに対してその値で取得しようとするか。
- 関数の戻り値として、取得された結果が呼び出し元に返されるか。
となります。
実際に書いてみたのがこちら。
def test_get_user():
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)
user = repo.get(1)
session.query.assert_called_once_with(User)
args, kwargs = mock_query.filter.call_args
assert len(args) == 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():
session = MagicMock()
user_obj = User(id=2, name="Saburo", email="saburo@example.com")
repo = UserRepository(session)
repo.get = MagicMock(return_value=user_obj)
updated = repo.update(2, name="326", email="saburo2@example.com")
assert user_obj.id == 2
assert user_obj.name == "326"
assert user_obj.email == "saburo2@example.com"
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():
session = MagicMock()
user_obj = User(id=3, name="Shiro", email="shiro@example.com")
repo = UserRepository(session)
repo.get = MagicMock(return_value=user_obj)
result = repo.delete(3)
session.delete.assert_called_once_with(user_obj)
args, kwargs = session.delete.call_args
assert args[0].id == 3
session.commit.assert_called_once()
assert result is True
session.delete、session.commitが呼ばれてることを確認することで、どのように削除しようとしているかをテストしています。
まとめ
ひと通り、CRUD操作に対するテストを行うことは出来ました。ただ、この確認観点だと記述量が多くなるなというのが正直な感想です。
今回は、mock-alchemyを使ってみましたが、色々調べてみると、SQLiteを使う方法もあるそうです。
（っていうかそっちのほうが一般的なのかな？）
次回は、SQLiteを使ってテストコードを書く方法を試してみたいと思います。
