Zenn
🐍

pytestで非同期関数をテストする

2024/04/24に公開
Python
Python 3
FastAPI
pytest
async
tech

FastAPIを使っていると非同期関数を扱うことが増えます。
非同期関数のテストの書き方をメモとして記載します。

コード

下記のコードをテストするケースを考える。

repo.py
from sqlalchemy import select
from sqlalchemy.orm import Session


class UsersRepository:
    async def find_by_id(session: Session, id: int) -> User:
        result = await session.execute(select(User).where(User.id == id)
        return result.scalars().one_or_none()

テストコードはこんな感じになる。

test.py
import pytest
from .repo import UsersRepository


@pytest.mark.asyncio
async def test_find_by_id():
    result = await UsersRepository.find_by_id(session=session, id=1)
    assert isinstance(result, User)

解説

@pytest.mark.asyncioをつけずにテストを実行すると、下記のWarningが出る。

Waring
PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped.
  You need to install a suitable plugin for your async framework, for example:
    - anyio
    - pytest-asyncio
    - pytest-tornasync
    - pytest-trio
    - pytest-twisted
    warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))

pytest単体では非同期関数をサポートしていないのでテストがスキップされている。
非同期関数テストライブラリをインストールしてデコレータをつけるとテストが実行されるようになる。

zsh
poetry add -D pytest-asyncio

mock

ちなみに非同期関数のmockは同期関数と同様にできました。

service.py
from sqlalchemy.orm import Session


class UsersService:
    @staticmethod
    async def find_by_id(session: Session, id: int) -> User:
        return await UsersRepository.find_by_id(session=session, id=id)
test.py
from unittest.mock import patch

import pytest


@pytest.mark.asyncio
@patch("repo.UsersRepository.find_by_id")
async def test_find_by_id(self, find_by_id):
    find_by_id.return_value = User(id=1)
    result = await UsersService.find_by_id(session="dummy", id=1)

    find_by_id.assert_called_once_with(session="dummy", id=1)
    assert isinstance(result, User)
GitHubで編集を提案

Discussion