Python: typing.Protocolを使った単体テスト用拡張作業例

概要

元々 Structual Subtyping の typing.Protocol と Nominal Subtyping の abc.ABC の違いについて書こうかと思っていましたが、止めました。今年もアドベントカレンダーが半ばを過ぎてネタが被っている上に元々書こうと思っていた内容よりもよくまとまっている記事をいくつも見かけたからです。

ということで、私の typing.Protocol のユースケースの話をします。

前提

仕事で Python のコードを書くときはいちから書くことはあまりなくて、既存のコードの修正の対応がほとんどです。プロジェクトによっては Type Hints がなかったり、要所外/テストしにくいコードの単体テストがなかったり、といった具合のところで、できる範囲で単体テストを足して修正をしていっています。

元々Go言語を多く書くので、Go言語の interface

相当のものがあれば自分にとって既存のコード修正前の単体テスト整備がやりやすいなと思っていました。

すると調べていなかっただけで Python 3.8 から typing.Protocol が導入されていました。新機能は何かと目にする機会もあるのですが、既にある機能は困ったタイミングまでなかなか見つけられないものだなと反省しました。

以降、既存実装があるが単体テストがない状態から、機能追加前にテスト追加していく工程を書いていきます。

対応

対応の流れ

今回の対応の流れは以下です。

1. 既存実装の確認
2. モンキーパッチを使って既存実装のテストを通す
3. typing.Protocol で制約対象を定義する
4. typing.Protocol を使ったテストを作る
5. typing.Protocol を使った実装をする
6. (詳細略記) 呼び出し元の差し替え、不要となった実装とテストを削除

なお、本Python プロジェクトは

• uv
• pytest

を使っているものとします。

既存実装の確認

今回の既存実装例は

• アプリケーション内で閉じている(外部公開してない)
• 単体テストがない

前提でアプリケーションに対しての外部サービス(例：データベースでもいいですし、クラウドサービスでもよいです)を取り扱うクラスのインスタンス化を関数内で行っている場合とします ( external_service.py )。 get_user_status のテストはステージング環境で結合テストをするようなイメージです。

# import 略

# 外部サービスへのリクエストのクラス
# この実装にはできるだけ手を入れたくない
class ExternalService:
    def fetch_user_data(self, user_id: int) -> dict:
        # 実際には外部サービスから値を取得して返却
        return {"id": user_id, "name": "山田太郎", "active": True}

# 今回のテスト対象
def get_user_status(user_id: int) -> str:
    # 関数内部でインスタンスを生成しているこの箇所が問題になりやすい
    service = ExternalService()
    user_data = service.fetch_user_data(user_id)

    if user_data.get("active"):
        return f"User {user_id} is active"

    return f"User {user_id} is inactive"

上記コードの get_user_status を戻り値は変更せずに、引数で typing.Protocol で fetch_user_data を持つ想定のクラスを受け取るように修正してテスト(だけではなく、別サービス切り替えのさいも対応できますが)への拡張性を高めていこうと思います。

モンキーパッチで既存実装のテストを通す

ExternalService クラスのインスタンス化が get_user_status 内で行われているので単体テストが書きづらい状態です。一旦、実装を変更せずに単体テストを実行できるようにします。

モンキーパッチでテスト実行時に ExternalService.fetch_user_data が実行される時の値を一時的に上書きします。

また、私がGoをよく書いている方なので、テストは Table Driven Tests で実施しています。

# import 略

# Table Driven Test 風
# テスト関数に対して引数に対応した名前で値をループして渡す
@pytest.mark.parametrize(
    ("user_id","is_active", "expected_result"),
    [
        (1, True,  "User 1 is active"),
        (2, False, "User 2 is inactive")
    ]
)
def test_get_user_status_active_with_monkeypatch(
        mocker,
        user_id: int,
        is_active: bool,
        expected_result: str
):
    # --- Arrange ---
    # モンキーパッチ作成
    # ExternalService.fetch_user_dataの値を `return_value` で上書き
    mock_fetcher = mocker.patch.object(
        ExternalService,
        'fetch_user_data',
        return_value={"id": user_id, "active": is_active}
    )

    # --- Act ---
    actual_result = get_user_status(user_id)

    # --- Assert ---
    assert actual_result == expected_result
    mock_fetcher.assert_called_once_with(user_id)

テストを実行します。

# オプションは記事に記載するさいに文字数が少なくなるように出力抑制しています
$ uv run pytest -q --disable-warnings --tb=no
..            　　　　　　　　　　　　  [100%]
2 passed in 0.04s

テストが通りました。

typing.Protocol で制約対象を定義する

テストが通ったので、 typong.Protocol で fetch_user_data を持つ制約の Protocol を作成します。名前は関数名から考えて UserFetcher とします(動作に対して -er / -or を付けたものを名前にするのも Go の interface の命名習慣からです)。

from typing import Protocol

class UserFetcher(Protocol):
    # 制約対象
    def fetch_user_data(self, user_id: int) -> dict: ...

typing.Protocol を使ったテストを作る

fetch_user_data を持つ UserFetcher を定義したので、これを使った get_user_status 用のテストを書きます。これは後に既存の実装をどう置き換えていくかの流れ次第ではあるのですが、ここでは get_user_status_v2 という別名の関数を作成してテストを作成します。

# `test_get_user_status_active` のコードが同じファイルにあるが略

# テストケースは `test_get_user_status_active` と一緒
@pytest.mark.parametrize(
    ("user_id","is_active", "expected_result"),
    [
        (1, True,  "User 1 is active"),
        (2, False, "User 2 is inactive")
    ]
)
def test_get_user_status_active_v2(
        mocker,
        user_id: int,
        is_active: bool,
        expected_result: str
):
    #  --- Arrange ---
    # 
    expected_user_data = {"id": user_id, "active": is_active}
    mock_fetcher = mocker.Mock(spec=UserFetcher)
    mock_fetcher.fetch_user_data.return_value = expected_user_data

    # --- Act ---
    actual_result = get_user_status_v2(mock_fetcher, user_id)

    # --- Assert ---
    assert actual_result == expected_result
    mock_fetcher.fetch_user_data.assert_called_once_with(user_id)

テストを実行します。

$ uv run pytest -q --disable-warnings --tb=no
..FF
================================================================================ short test summary info =================================================================================
FAILED tests/test_external_service.py::test_get_user_status_active_v2[1-True-User 1 is active] - NameError: name 'get_user_status_v2' is not defined
FAILED tests/test_external_service.py::test_get_user_status_active_v2[2-False-User 2 is inactive] - NameError: name 'get_user_status_v2' is not defined
2 failed, 2 passed in 0.06s

get_user_status_v2 は存在しない(かつ、存在しないので import もしていない)ので想定通り test_get_user_status_v2 のテストは失敗しました。

typing.Protocol を使った実装をする

それでは実装します。 fetch_user_data はこの実装では引数の UserFetcher ( fetch_user_data を持つもの ) を介して実行されるようになります。

# 既存の実装部分省略

# 第一引数に `UserFetcher` を取るようになった
def get_user_status_v2(fetcher: UserFetcher, user_id: int) -> str:
    # `fetch_user_data` は `UserFetcher` を介して実行される
    user_data = fetcher.fetch_user_data(user_id)

    if user_data.get("active"):
        return f"User {user_id} is active"

    return f"User {user_id} is inactive"

テスト側に実装した get_user_status_v2 の import を追加してテストを実行します。

$ uv run pytest -q --disable-warnings --tb=no
....            　　　　　　　　　　　　[100%]
4 passed in 0.32s

テストが通りました。

(詳細略記) 呼び出し元の差し替え、不要となった実装とテストを削除

ここからは略記ですが

• get_user_status の利用箇所の前で適切な UserFetcher のインスタンスを生成して get_user_status_v2 を呼び出すコードに修正 ( get_user_status の呼び出し箇所がなくなっていること)
• テスト実施をして修正後のコードのテストが通ることを確認
• 呼び出されていない get_user_status を削除
• テストを実行して、get_user_status が未定義による test_get_user_status の失敗だけになっていることを確認
• test_get_user_status を削除してテストが通ることを確認

としていきます。

おわりに

typing.Protocol を使った単体テストへの対応のための拡張の一例を書きました。

実際のところどれくらい丁寧/ざっとの対応にするかはここには記載していないようなプロジェクトの諸事情によると思うので何とも言えないところはあるのですが。

説明のために抽象化しようとしてうまくいっていないようなところもあるような気がしますが、やっていることを整理・言語化できたいい機会だと思います。