StreamingResponseを返すFastAPIアプリケーションのテストの書き方を考える

はじめに

エンジニアニメ、ありがとうございました！ nikkieです。
さくらインターネットさんで櫻木真乃さんのお話をしたよ...

FastAPIのテストについての続編です。
現時点の思考のログといった趣です。
私は経験少ない（最近書き始めた）ので、「もっとよいやり方がある」という情報をお待ちしています🙏

動作環境は、はじめてTestClientの記事と同じです

StreamingResponseを返すFastAPIアプリ

fastapi.responses.StreamingResponseにイテレータを渡すだけです¹。
https://fastapi.tiangolo.com/advanced/custom-response/#streamingresponse

Takes an async generator or a normal generator/iterator and streams the response body.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/stream")
async def stream():
    def generator():
        yield b"data: Hello,\n\n"
        yield b"data: World!\n\n"

    return StreamingResponse(generator(), media_type="text/event-stream")

（本来テストを書きたいコードがasync defなので、awaitを使っていないですがasync defに揃えました²）

このエンドポイントにcurlしてみます（fastapi dev app/main.pyでアプリケーション立ち上げ）。

% curl -N http://127.0.0.1:8000/stream
data: Hello,

data: World!

HTTPXでstreaming responseをさばく

FastAPIのTestClientの前に、（curlの代わりに）HTTPXからどう扱うかを考えました。
ドキュメント「Streaming Responses」
https://www.python-httpx.org/quickstart/#streaming-responses

>>> import httpx
>>> client = httpx.Client()
>>> with client.stream("GET", "http://127.0.0.1:8000/stream") as response:
...   for line in response.iter_lines():
...     print(line)
... 
data: Hello,

data: World!

現時点の結論：HTTPX同様にTestClientでもstreamをさばいてテストする

.
└── app/
    ├── __init__.py
    ├── main.py
    └── test_main.py

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_Server_sent_events():
    with client.stream("GET", "/stream") as response:
        assert response.status_code == 200
        assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
        assert list(response.iter_lines()) == ["data: Hello,", "", "data: World!", ""]

先ほどのHTTPXのコードを元に、クライアントをTestClientに変更しただけです。
TestClientはHTTPXと同じように使えます³。

Use the TestClient object the same way as you do with httpx. (Using TestClient)

list(response.iter_lines())のところですが、HTTPXはいろいろな方法でStreaming Responsesを扱えるので⁴

assert response.read() == b"data: Hello,\n\ndata: World!\n\n"

のようにも書けました。
今回はstream呼び出し側でiter_lines()を使うので、こちらを採用しています。

Streaming Responseとして扱わなくてもテストは書けた

採用していませんが、気付きとして残しておきます。
TestClientでstream()メソッドを使わなくても書けました。

def test_Server_sent_events():
    response = client.get("/stream")
    assert response.status_code == 200
    assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
    assert response.text == "data: Hello,\n\ndata: World!\n\n"

単にget()メソッドだと、httpx.Responseが返ります。
https://www.python-httpx.org/api/#response
text属性はcurlでいうバッファのような動きなのですね（stream()のときはcurlのunbufferぽいですね）

FastAPIアプリを使うコードに合わせて、stream()メソッドで呼び出すテストコードを今回採用しました（この記事で意見ほしいポイントです）

終わりに

StreamingResponseを返すFastAPIアプリのテストを、stream()メソッドで考えました。
今回重視していたのは、FastAPIを使うコードに合わせてテストを書くことです。
TestClientはHTTPXと同じインターフェースで使え、streamの扱いも豊富というのが1つ学びです。

ジェネレータ（yieldを持つ関数）は、イテレータです ↩
awaitするならasync def、しないならdefと理解しています ↩
Testingのドキュメント内の「Technical Details」によると、TestClientはStarletteから来ているそうです↩
過去記事より ↩

nikkie-ftnextの日記

イベントレポートや読書メモを発信