Zenn
Closed6

pytest 用に非同期サーバーを別スレッドで建てる

Python
pytest
aiohttp
Uvicorn
httpx
まちゅけんまちゅけん

Context

httpx のような HTTP クライアントは同期 / 非同期 IO に対応して便利です。

httpx を使って何かの特定の Web API SDK を開発しているとき、pytest によるテストでモックサーバーにリクエストをしたいのですが、もちろんテストでも同期 / 非同期 IO の両方で同じ仕様のモックサーバーへテストリクエストしたいです。

また個人的な要件で HTTP リクエストだけでなく WebSocket API も利用します。 WebSocket に関してはラッパーライブラリである httpx-ws で実現できるのですが、これに関しても同じく同期 / 非同期でテストしたいです。

以上の背景から、このスクラップでは以下の要件を満たせるようなテスト用モックサーバーを建てる方法を探っていきます。

  1. pytest の Fixture でシンプルに実現できる
  2. 同期 / 非同期
  3. HTTP / WebSocket
  4. サーバーのパッケージが巨大な依存関係ではない
  5. ASGI
まちゅけんまちゅけん

Async server with threading

pytest の Fixture にて非同期サーバーを建ててクライアントからリクエストしようとすると、まず直面する問題点があります。

  • 非同期 Fixture でサーバーを建てると、テストの同期クライアントからリクエストするとイベントループが停止してサーバーがレスポンスを返せない

非同期サーバーでなく、同期サーバーを建てるとしても別の問題に直面します。

  • Fixture にて別スレッドでサーバーを実行する必要あり、別スレッドで立てたサーバーのシャットダウン処理がそこそこ難しい (シャットダウンができないと pytest が終了しない)
  • そもそも同期サーバーの多くは非同期が前提の WebSocket に対応していない

これを解決するには「別スレッドで非同期サーバーを建てる」という方法がよさそうです。 別スレッドでイベントループが動くので、テスト関数において同期リクエストを実行してもレスポンスが返されます。 また非同期サーバーであれば多くの場合 WebSocket を利用できます。

別スレッドで非同期サーバーを建てるには、anyio の blocking portal 機能を使うことで正しい手続きができます。 例えば Server() が特定の非同期サーバークラスだとして、以下のスニペットで実現できます。

from contextlib import asynccontextmanager
from anyio import from_thread

import httpx
import pytest

@asynccontextmanager
async def serve_test_server(...):
    server = Server()
    await server.start()
    yield server
    await server.shutdown()


@pytest.fixture(scope="session")
def my_test_server():
    with (
        from_thread.start_blocking_portal() as portal,
        portal.wrap_async_context_manager(serve_test_server(...)) as server,
    ):
        yield server


def test_client(my_test_server):
    with httpx.Client() as client:
        r = client.get(my_test_server.url)
    assert r.status_code == 200


@pytest.mark.anyio
async def test_client(my_test_server):
    async with httpx.AsyncClient() as client:
        r = await client.get(my_test_server.url)
    assert r.status_code == 200

こちらのスニペットにより、非同期サーバーがあれば同期 / 非同期クライアントによるリクエストをテストを実現できます。 テストが終わったあと、Fixture のシャットダウンも問題ありません。 以降はこれを元に、要件を満たす非同期サーバーを探していきます。

まちゅけんまちゅけん

aiohttp

✅ pytest Fixture
✅ threading
✅ HTTP / WebSocket
❌ ASGI

import aiohttp
import pytest
from aiohttp import web
from aiohttp.test_utils import TestServer
from anyio import from_thread

if TYPE_CHECKING:
    from collections.abc import Generator


@pytest.fixture(scope="session")
def aiohttp_server() -> Generator[TestServer, None, None]:
    async def http_handler(request: web.Request) -> web.Response:
        return web.json_response({"Hello": "World"})

    async def websocket_handler(request: web.Request) -> web.WebSocketResponse:
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        async for msg in ws:
            if msg.type == aiohttp.WSMsgType.TEXT:
                await ws.send_str(f"Message text was: {msg.data}")

        return ws

    app = web.Application()
    app.add_routes(
        [
            web.route("*", "/", http_handler),
            web.route("GET", "/ws", websocket_handler),
        ]
    )

    with from_thread.start_blocking_portal(
        backend="asyncio"
    ) as portal, portal.wrap_async_context_manager(TestServer(app)) as server:
        yield server


@pytest.fixture(scope="session")
def http_server(aiohttp_server: TestServer) -> str:
    return str(aiohttp_server.make_url("/"))


@pytest.fixture(scope="session")
def websocket_server(aiohttp_server: TestServer) -> str:
    return str(aiohttp_server.make_url("/ws"))


def test_websocket(websocket_server: str):
    with httpx_ws.connect_ws(websocket_server) as ws:
        ws.send_text("Hello, world!")
        assert ws.receive_text() == "Message text was: Hello, world!"


@pytest.mark.anyio
async def test_async_websocket(websocket_server: str):
    async with httpx_ws.aconnect_ws(websocket_server) as ws:
        await ws.send_text("Hello, world!")
        assert await ws.receive_text() == "Message text was: Hello, world!"

Public API として提供されている TestServer クラスが大変使いやすいです。 pytest-aiohttp の方を利用してしまうと、前述した同期リクエストの問題でレスポンスがブロックされてしまいますので、このスニペットが必要になります。

ただし ASGI ではないので、汎用性は少し落ちます。

まちゅけんまちゅけん

Uvicorn

✅ pytest Fixture
✅ threading
✅ HTTP / WebSocket
✅ ASGI

from collections.abc import Generator
from contextlib import asynccontextmanager
from socket import socket
from typing import AsyncGenerator

import httpx_ws
import pytest
import uvicorn
from anyio import from_thread
from fastapi import FastAPI, WebSocket
from uvicorn._types import ASGIApplication


@asynccontextmanager
async def serve_test_server(app: ASGIApplication) -> AsyncGenerator[socket, None]:
    server = uvicorn.Server(uvicorn.Config(app, port=0))

    config = server.config
    if not config.loaded:
        config.load()

    server.lifespan = config.lifespan_class(config)

    sock = config.bind_socket()

    await server.startup([sock])
    if server.should_exit:
        return

    yield sock

    await server.shutdown([sock])


@pytest.fixture(scope="session")
def uvicorn_server() -> Generator[socket, None, None]:
    app = FastAPI()

    @app.get("/")
    def read_root():
        return {"Hello": "World"}

    @app.websocket("/ws")
    async def websocket_endpoint(websocket: WebSocket):
        await websocket.accept()
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")

    with (
        from_thread.start_blocking_portal() as portal,
        portal.wrap_async_context_manager(serve_test_server(app)) as server,
    ):
        yield server


@pytest.fixture(scope="session")
def http_server(uvicorn_server: socket) -> str:
    host, port = uvicorn_server.getsockname()
    return f"http://{host}:{port}"


@pytest.fixture(scope="session")
def websocket_server(uvicorn_server: socket) -> str:
    host, port = uvicorn_server.getsockname()
    return f"http://{host}:{port}/ws"


def test_websocket(websocket_server: str):
    with httpx_ws.connect_ws(websocket_server) as ws:
        ws.send_text("Hello, world!")
        assert ws.receive_text() == "Message text was: Hello, world!"


@pytest.mark.anyio
async def test_async_websocket(websocket_server: str):
    async with httpx_ws.aconnect_ws(websocket_server) as ws:
        await ws.send_text("Hello, world!")
        assert await ws.receive_text() == "Message text was: Hello, world!"

ASGI 準拠の HTTP サーバーなので、FastAPI でも Quart でもなんでも利用できます。

だたし Publish API として説明されていない uvicorn.Server クラスの startup() などを利用している点に注意が必要です。

まちゅけんまちゅけん

Hypercorn

✅ pytest Fixture
✅ threading
✅ HTTP / WebSocket
✅ ASGI

Uvicorn 同様 ASGI 準拠のサーバーです。 urllib3 のテストケースを参考にしていたところ利用されていました。

https://github.com/urllib3/urllib3/blob/79a2a30f06b61127dc138158b82ffa791ab16733/test/conftest.py#L76-L90

https://github.com/urllib3/urllib3/blob/79a2a30f06b61127dc138158b82ffa791ab16733/dummyserver/hypercornserver.py

urllib3 ではそこそこ複雑な構成でサーバーを建てているようです。

まちゅけんまちゅけん

pytest-httpbin

✅ pytest Fixture
✅ threading
✅ HTTP
❌ WebSocket
❌ ASGI

こちらは WebSocket の要件を満たしませんが、HTTP に限っては非常に簡単に構築できるサーバーです。 公開サーバーである https://httpbin.org/ と同等のサーバーをローカルで実行します。

フィクスチャとして httpbin を利用するだけです。 httpcore のテストケースで利用されていました。

https://github.com/encode/httpcore/blob/acf7d151c715d0b18870ecc660a8677f57f0bcea/tests/test_api.py

このスクラップは4ヶ月前にクローズされました