閉じる

Pythonのunittest.mockで外部APIとDB接続をテストする

外部APIやDBに依存する処理はテストが難しく、遅く、不安定になりがちです。

本稿では、Python標準のunittest.mockを使い、API通信やDB接続を安全かつ高速に再現する具体的な手順を、初心者の方にもわかりやすく解説します。

実践的なコード例とともに、モックの基本からベストプラクティスまで丁寧に説明します。

unittest.mockの基本

モック

モックとは、本物のオブジェクトの代わりに振る舞いを模倣するテスト用の偽物を差し込む手法です。

外部APIやDBなどの外部要因に依存せず、速く再現性のあるテストを書けます。

Python標準のunittest.mockモジュールは、MockMagicMockpatchなどを提供し、関数の戻り値や例外、呼び出し回数などを柔軟に制御できます。

スタブやフェイクとの違い

用語は文脈で曖昧ですが、スタブは戻り値を決め打ちした簡易な置き換え、フェイクは簡易実装を持つ本物に近い置き換え、モックは呼び出し検証まで行う置き換え、と理解しておくと整理しやすいです。

MagicMockとMockの違い

MagicMock特殊メソッド(例: __len____iter____enter__など)を標準で持ち、コンテキストマネージャや演算などもモック化しやすいのが特徴です。

一方Mockは必要最小限で軽量です。

以下の表で違いを整理します。

比較項目MockMagicMock使いどころ
特殊メソッドなしありwith文やlen、イテレーションが必要ならMagicMock
既定の柔軟性高いとても高いとりあえず動かしたいならMagicMock
厳格性付与spec/spec_setで可能spec/spec_setで可能どちらも可能だが厳格にしたい時はspec系を付与

迷ったらまずはMock、特殊メソッドが必要になったらMagicMockと覚えると良いです。

MagicMockが必要な例

Python
# MagicMockは __enter__/__exit__ を持つので with 文でも動く
from unittest.mock import MagicMock

m = MagicMock()
with m as resource:
    resource.do_something()  # OK

patchの基本

patch名前を差し替える仕組みです。

差し替え対象は「使っている側のモジュール内の名前」である点が重要です。

デコレータ、コンテキストマネージャ、関数として利用できます。

Python
# patchの3つの使い方
import time
from unittest.mock import patch

# 1) コンテキストマネージャ
with patch("time.sleep", return_value=None) as m_sleep:
    time.sleep(10)
    m_sleep.assert_called_once_with(10)

# 2) デコレータ
@patch("time.sleep", return_value=None)
def work(_):
    time.sleep(1)
    return "done"

print(work())  # "done"

# 3) 手動で開始・終了
p = patch("time.sleep", return_value=None)
m_sleep = p.start()
try:
    time.sleep(3)
    m_sleep.assert_called_once()
finally:
    p.stop()
実行結果
done

return_valueとside_effectの使い分け

return_valueは戻り値を固定し、side_effect例外発生や連続した値の返却、コールバック関数の実行に使います。

Python
from unittest.mock import Mock
import itertools

m = Mock()

# 固定の戻り値
m.calc.return_value = 42
print(m.calc())  # 42

# 呼び出しのたびに値を変える
m.next.side_effect = [1, 2, 3]
print([m.next() for _ in range(3)])  # [1, 2, 3]

# 例外を発生させる
class MyError(Exception):
    pass

m.fail.side_effect = MyError("boom")
try:
    m.fail()
except MyError as e:
    print("caught:", e)

# コールバックで動的に決める
def multiply(x, y):
    return x * y

m.mul.side_effect = multiply
print(m.mul(3, 5))  # 15
実行結果
42
[1, 2, 3]
caught: boom
15

呼び出し検証

モックは実行後にどのように呼ばれたかを検証できます。

Python
from unittest.mock import Mock, call

m = Mock()

m.send("hello", to="alice")
m.send("bye", to="bob")

# 代表的なアサーション
m.send.assert_called()  # 少なくとも1回
m.send.assert_called_with("bye", to="bob")  # 直近の呼び出し
m.send.assert_any_call("hello", to="alice")  # どこかで呼ばれた
m.send.assert_has_calls(
    [call("hello", to="alice"), call("bye", to="bob")],
    any_order=False
)
print("count:", m.send.call_count)
print("calls:", m.send.mock_calls)
実行結果
count: 2
calls: [call('hello', to='alice'), call('bye', to='bob')]

外部API(HTTP)のテストをモックする

ここではrequestsを使う実装を想定し、HTTP通信を発生させずにテストする方法を解説します。

ディレクトリは例としてapp/配下に実装し、tests/配下にテストを書きます。

requests.getをpatchする

まずは実装例です。

APIキーは環境変数API_KEYから読みます。

Python
# app/services.py
import os
import requests

BASE_URL = "https://api.example.com"

def fetch_user(user_id: int) -> dict:
    """外部APIからユーザー情報を取得します。"""
    api_key = os.getenv("API_KEY")
    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
    url = f"{BASE_URL}/users/{user_id}"
    # タイムアウト付きでGET
    resp = requests.get(url, headers=headers, timeout=5)
    if resp.status_code != 200:
        raise RuntimeError(f"API error: status={resp.status_code}")
    return resp.json()

中身は空でいいので、appフォルダに__init__.pyを作成することを忘れないようにしてください。

テストでは実装が属するモジュール内の名前をpatchします。

ここではapp.services.requests.getです。

Python
# test_services_api.py
import unittest
from unittest.mock import patch, Mock
import requests  # 例外クラスの参照に使用

class TestFetchUser(unittest.TestCase):
    @patch("app.services.requests.get")  # 実装側の名前をpatch
    @patch("app.services.os.getenv", return_value="DUMMY_KEY")
    def test_fetch_user_success(self, m_getenv, m_get):
        # レスポンスのモックを作成
        m_resp = Mock()
        m_resp.status_code = 200
        m_resp.json.return_value = {"id": 1, "name": "Alice"}
        m_get.return_value = m_resp

        from app.services import fetch_user  # patch後にimportしてもOK
        user = fetch_user(1)

        self.assertEqual(user["name"], "Alice")
        m_get.assert_called_once_with(
            "https://api.example.com/users/1",
            headers={"Authorization": "Bearer DUMMY_KEY"},
            timeout=5,
        )

    @patch("app.services.requests.get", side_effect=requests.Timeout)
    def test_fetch_user_timeout(self, m_get):
        from app.services import fetch_user
        with self.assertRaises(requests.Timeout):
            fetch_user(1)

    @patch("app.services.requests.get")
    def test_fetch_user_non_200(self, m_get):
        m_resp = Mock(status_code=500)
        m_resp.json.return_value = {"error": "server"}
        m_get.return_value = m_resp

        from app.services import fetch_user
        with self.assertRaises(RuntimeError):
            fetch_user(1)

if __name__ == "__main__":
    unittest.main(verbosity=2)
実行結果
test_fetch_user_non_200 (__main__.TestFetchUser) ... ok
test_fetch_user_success (__main__.TestFetchUser) ... ok
test_fetch_user_timeout (__main__.TestFetchUser) ... ok

----------------------------------------------------------------------
Ran 3 tests in 0.00s

OK

ダミーのレスポンスを返す

上記のようにMock()status_codejson()を設定すれば十分です。

より厳密にするならspecを付けてrequests.Responseに近づけます。

Python
from unittest.mock import create_autospec
import requests

m_resp = create_autospec(requests.Response, instance=True)
m_resp.status_code = 200
m_resp.json.return_value = {"ok": True}

例外やタイムアウトを再現する

side_effectrequests.Timeoutrequests.RequestExceptionを指定します。

これでタイムアウト時のリトライやログなどの動作を安全に検証できます。

Python
@patch("app.services.requests.get", side_effect=requests.RequestException("network"))
def test_network_error(self, m_get):
    from app.services import fetch_user
    with self.assertRaises(requests.RequestException):
        fetch_user(1)

APIキーや環境変数をモックする

環境変数はpatch.dictで一時的に切り替えられます。

os.getenvを直接patchする方法でも構いません。

Python
import os
from unittest.mock import patch

def test_env_with_patch_dict():
    from app import services
    with patch.dict(os.environ, {"API_KEY": "K"}, clear=False):
        with patch("app.services.requests.get") as m_get:
            m_get.return_value.status_code = 200
            m_get.return_value.json.return_value = {}
            services.fetch_user(1)
            # ヘッダーにAPIキーが入っていることを確認
            args, kwargs = m_get.call_args
            assert kwargs["headers"]["Authorization"] == "Bearer K"

patch対象は実装側のインポート先にする

最も多い失敗patchの対象を間違えることです。

例えば実装でfrom requests import getとしているなら、patch対象はapp.services.getです。

実装でimport requestsならapp.services.requests.getです。

ライブラリそのものrequests.getを直接patchすると、別モジュールからの呼び出しまで意図せず影響する可能性があります。

DB接続のテストをモックする

DB接続層を分けてテストする考え方

DBは接続とクエリ実行を行う層を薄く分離し、上位のビジネスロジックはその層に依存する形にするとテストしやすくなります。

ここではsqlite3を例に、接続開始からコミット・ロールバック・クローズまでの呼び出しをモックで検証します。

sqlite3.connectをモックする

実装例を用意します。

Python
# app/db.py
import sqlite3
from typing import Optional

def get_user_name(user_id: int) -> Optional[str]:
    """ユーザー名を返します。見つからなければNone。"""
    conn = sqlite3.connect("app.db")
    try:
        cur = conn.cursor()
        cur.execute("SELECT name FROM users WHERE id = ?", (user_id,))
        row = cur.fetchone()
        return row[0] if row else None
    finally:
        conn.close()

def update_user_name(user_id: int, new_name: str) -> None:
    """ユーザー名を更新します。成功でcommit、失敗でrollbackします。"""
    conn = sqlite3.connect("app.db")
    try:
        cur = conn.cursor()
        cur.execute("UPDATE users SET name = ? WHERE id = ?", (new_name, user_id))
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

テスト側でapp.db.sqlite3.connectをpatchします。

Python
# tests/test_db.py
import unittest
from unittest.mock import patch, Mock

class TestDB(unittest.TestCase):
    @patch("app.db.sqlite3.connect")
    def test_get_user_name(self, m_connect):
        # connect() -> conn モック
        m_conn = m_connect.return_value
        m_cur = m_conn.cursor.return_value
        m_cur.fetchone.return_value = ("Alice",)

        from app.db import get_user_name
        name = get_user_name(1)

        self.assertEqual(name, "Alice")
        m_cur.execute.assert_called_once_with(
            "SELECT name FROM users WHERE id = ?", (1,)
        )
        m_conn.close.assert_called_once()

    @patch("app.db.sqlite3.connect")
    def test_update_user_name_commit(self, m_connect):
        m_conn = m_connect.return_value
        m_cur = m_conn.cursor.return_value

        from app.db import update_user_name
        update_user_name(1, "Bob")

        m_cur.execute.assert_called_once()
        m_conn.commit.assert_called_once()
        m_conn.rollback.assert_not_called()
        m_conn.close.assert_called_once()

    @patch("app.db.sqlite3.connect")
    def test_update_user_name_rollback(self, m_connect):
        m_conn = m_connect.return_value
        m_cur = m_conn.cursor.return_value
        # executeで例外を発生させる
        m_cur.execute.side_effect = RuntimeError("DB error")

        from app.db import update_user_name
        with self.assertRaises(RuntimeError):
            update_user_name(1, "Carol")

        m_conn.rollback.assert_called_once()
        m_conn.commit.assert_not_called()
        m_conn.close.assert_called_once()

if __name__ == "__main__":
    unittest.main(verbosity=2)
実行結果
test_get_user_name (__main__.TestDB) ... ok
test_update_user_name_commit (__main__.TestDB) ... ok
test_update_user_name_rollback (__main__.TestDB) ... ok

----------------------------------------------------------------------
Ran 3 tests in 0.00s

OK

cursor.execute/fetchoneを設定する

上の通りconnect.return_value.cursor.return_valueのようにチェーンした戻り値を順に設定します。

返したい行はfetchone.return_valueで指定できます。

該当のSQLが複数ある場合はassert_has_callsで順序も検証できます。

Python
from unittest.mock import call

# 複数のSQLを順序通りに呼んだか確認
expected = [
    call("BEGIN"),
    call("UPDATE users SET name = ? WHERE id = ?", ("Bob", 1)),
    call("COMMIT"),
]
# 例えば m_cur.execute.assert_has_calls(expected, any_order=False)

commit/rollback/closeの呼び出しを検証する

トランザクションの正しさはcommitとrollbackの呼び分けで担保されます。

例外を発生させてrollback()が呼ばれたかを確認し、正常系ではcommit()が呼ばれたことを確認します。

最後にclose()が必ず呼ばれることも大切です。

モックとインメモリDB(:memory:)の使い分け

ロジックの大半がSQLに依存する場合、sqlite3.connect(":memory:")実DBを使った高速テストを行う選択肢も有効です。

  • モックを使うべき場面: 失敗時の分岐や例外、接続エラー、コミット/ロールバックの呼び出し検証など制御フローのテスト
  • インメモリDBを使うべき場面: 実際のSQLの妥当性やJOINの動作などデータクエリの挙動

両者は対立ではなく補完関係です。

層ごとに使い分けると読みやすく保守しやすいテストになります。

よくあるミスとベストプラクティス

patchの対象を間違えない

patchは「使っている側のモジュール内の名前」を差し替えます

実装がimport requestsならapp.services.requests.getfrom requests import getならapp.services.getです。

ここを誤るとテストが意図通りに動きません。

autospec/spec_setで属性ミスを防ぐ

モックは何でも生えてしまうためタイプミスに気づきにくいです。

patch(..., autospec=True)create_autospec()spec_set=を使い、存在しない属性へのアクセスをエラーにしましょう。

Python
from unittest.mock import patch
import requests

@patch("app.services.requests.get", autospec=True)
def test_with_autospec(m_get):
    # m_getはrequests.getのシグネチャに合わせて厳格化される
    ...

with文/デコレータでスコープを管理する

patch()は使い終えたら確実に解除する必要があります。

基本はwithかデコレータでスコープを限定し、どうしても手動で開始する場合はtry/finallystop()を呼びます。

過度なモックを避ける

何でもモック化するとテストが実装詳細に強く依存して脆くなります。

外部境界(ネットワーク、ファイル、DB、時間)を中心にモックし、純粋なロジックは本物の関数でテストしましょう。

テストの読みやすさと実行速度のバランスが肝心です。

テストが読みやすくなる命名と配置

テストの可読性は保守性に直結します。

sut(System Under Test)やm_getなど役割が伝わる変数名にし、setup-exercise-verifyの流れがわかるよう段落を分けて書くと理解しやすくなります。

テストファイルはtests/配下で対象モジュールと同名に揃えると見つけやすいです。

まとめ

本稿では、unittest.mockで外部APIやDB接続を安全にテストする方法を、MockMagicMockの違い、patchの正しい適用先、return_value/side_effectの使い分け、呼び出し検証の実例を通じて解説しました。

HTTPではrequests.getを実装側の名前でpatchし、レスポンスやタイムアウトを自在に再現します。

DBではsqlite3.connectをモックし、executecommit/rollback/closeの呼び出しを検証しました。

加えてautospecで厳格性を高め、with文やデコレータでスコープを管理し、過度なモックを避けることで、速く壊れにくいテストを構築できます。

まずは外部境界からモックを適用し、必要に応じてインメモリDBも併用しながら、信頼できるテストスイートを育てていきましょう。

この記事を書いた人
エーテリア編集部
エーテリア編集部

人気のPythonを初めて学ぶ方向けに、文法の基本から小さな自動化まで、実際に手を動かして理解できる記事を書いています。

クラウドSSLサイトシールは安心の証です。

URLをコピーしました!