aiohttpを無理やりmockする

まずはよくあるケースとして、HTTPリクエストを送信せずに、送信したかのよ うに振る舞わせる。レスポンスを生成することが面倒なため、DummyResponse として偽のレスポンスオブジェクトを定義し、それに差し替えることにする。

import asyncio
from unittest import mock

import aiohttp
import pytest


async def send_request():
    async with aiohttp.ClientSession() as session:
        resp = await session.get('http://127.0.0.1:8234')
        resp.raise_for_status()

# mock
class DummyResponse(mock.MagicMock):
    def __await__(self):
        tmp = yield
        return self


class DummyClientSession:
    def __init__(self, resp: DummyResponse):
        self.resp = resp

    def __await__(self):
        tmp = yield
        return self

    def __aenter__(self):
        return self

    async def __aexit__(self, *args, **kwargs):
        await self.close()


    async def close(self, *args, **kwargs):
        pass

    async def get(self, *args, **kwargs):
        return self.resp


# testing
@pytest.mark.asyncio
async def test_simple():
    dummy_resp = DummyResponse()
    async with DummyClientSession(dummy_resp) as client:
        resp = await client.get()
        resp.raise_for_status()
        assert dummy_resp.raise_for_status.called


@pytest.mark.asyncio
async def test_aiohttp():
    with mock.patch("aiohttp.ClientSession") as ClientSession:
        dummy_resp = DummyResponse()
        ClientSession.return_value = DummyClientSession(dummy_resp)
        await send_request()
        assert dummy_resp.raise_for_status.called

実行

pytest test_aiohttp_mock.py

出力

============================= test session starts ==============================
platform darwin -- Python 3.8.0, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /srv/sximada/blogs/python-async
plugins: asyncio-0.10.0
collected 2 items

test_aiohttp_mock.py ..                                                  [100%]

=============================== warnings summary ===============================
/Users/sximada/.virtualenvs/py38/lib/python3.8/site-packages/aiohttp/helpers.py:107
  /Users/sximada/.virtualenvs/py38/lib/python3.8/site-packages/aiohttp/helpers.py:107: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead
    def noop(*args, **kwargs):  # type: ignore

-- Docs: https://docs.pytest.org/en/latest/warnings.html
========================= 2 passed, 1 warning in 0.20s =========================

上記でDummyResponseを定義したが型が異なるため、型ヒントを用いてチェッ クを行うとエラーする。現在のPythonは型ヒントの機能がかなり充実してきて いるため、常に型は本物を用いたほうが、コード上の問題にすばやく気付くこ とができる。しかしこのClientResponseは引数が多いため、所見だと仰々しく 感じる。ここではClientResponseをインスタンス化する方法を試す。

次の一覧は要求される引数だ。

• method
• url
• writer=None
• continue100
• timer
• request_info
• traces
• loop
• session

てっとり早くインスタンス化するためにNoneを設定できるが、urlとloopだけ はそれぞれきちんとした値を渡す必要がある。urlはyarl.URL()のインスタン スを生成して渡す。loopはイベントループなのでasyncio.get_event_loop()を 用いてイベントループを取得しそれを渡すこにした。

from asyncio import get_event_loop

from aiohttp import ClientResponse
from yarl import URL

url = URL("https://www.symdon.info")
loop = get_event_loop()

return ClientResponse(
    method="GET",
    url=url,
    writer=None,
    continue100=None,
    timer=None,
    request_info=None,
    traces=None,
    loop=loop,
    session=None,
)

<ClientResponse(https://www.symdon.info) [None None]>
None