テスト
テストクライアントを使用すると、httpx ライブラリを使用してAS GIアプリケーションのリクエストを送信できます。
starletteからHTMLResponseをインポート
starlette. testclientからTestClientをインポート
async def app(スコープ、受信、送信):
assert scope['type']=='http'
response=HTMLResponse('<html><body>Hello, world!</body></html>')
await応答(スコープ、受信、送信)
def test_app():
client=TestClient(アプリ)
response=client. get('/')
レスポンスをアサートします。ステータスコード==200
テストクライアントは、他の httpx セッションと同じインターフェイスを公開します。特に、リクエストを発行するための呼び出しは標準関数呼び出しであり、待機可能オブジェクトではないことに注意してください。
認証、セッションCookie処理、ファイルアップロードなど、任意の httpx 標準APIを使用できます。
たとえば、TestClientでヘッダーを設定するには、次のようにします。
client=TestClient(アプリ)
#将来のリクエストのためにクライアントにヘッダーを設定する
クライアントヘッダー= {"Authorization": "..."}
response=client. get("/")
#リクエストごとにヘッダーを個別に設定する
response=client. get("/",ヘッダー={"Authorization": "..."})
たとえば、TestClientを使用してファイルを送信します。
client=TestClient(アプリ)
#単一のファイルを送信する
with open("example. txt","rb")as f:
response=client.post("/form", files={"file": f})
#複数のファイルを送信する
with open("example. txt","rb")as f 1:
with open("example. png","rb")as f 2:
ファイル= {"file1": f1, "file2": ("filename", f2, "image/png")}
response=client.post("/form", files=files)
詳細については、httpx ドキュメントを参照してください。
デフォルトでは、TestClient はアプリケーションで発生するすべての例外をスローします。時々、クライアントがサーバー例外をスローすることを許可する代わりに、500エラーレスポンスの内容をテストしたい場合があります。その場合は、client = TestClient(app, raise_server_exceptions=False) を使用する必要があります。
注意する
TestClient に lifespan ハンドラを実行させたい場合、
TestClient をコンテキストとして使用する必要manager.Itます。
TestClient がinstantiated.Youときにトリガーされません
ここ。
非同期バックエンドを選択
TestClient は引数 backend(文字列)と backend_options(辞書)を受け取ります。これらのオプションは anyio.start_blocking_portal() に渡されます。受け入れ可能なバックエンドオプションの詳細については、anyioドキュメントを参照してください。デフォルトでは、asyncio とそのデフォルトオプションが使用されます。
申し訳ありませんが、原文の「Trio」と「backend="trio"」の意味が不明で、正確な翻訳ができません。この二つの記号についてもっと情報を提供していただければ、より正確な翻訳を提供できるように努力します。
defテストアプリ()
with TestClient(app, backend="trio")as client:
...
uvloop を使用して asyncio を実行するには、backend_options={"use_uvloop": True} を渡す必要があります。例:
defテストアプリ()
TestClient(app, backend_options={"use_uvloop": True})をクライアントとして使用します:
...
WebSocketセッションのテスト
テストクライアントを使用してWebSocketセッションをテストすることもできます。
httpx つまり、httpとwebsocketテストの間で同じ認証オプションと他のヘッダーを使用できます。
starlette. testclientからTestClientをインポート
starlette. websocketからWebSocketをインポート
async def app(スコープ、受信、送信):
assert scope['type']=='websocket'
websocket=WebSocket(スコープ、受信=受信、送信=送信)
websocketを待ちます。
websocket. send_text('こんにちは、世界!')を待ちます。
websocket. close()を待つ
def test_app():
client=TestClient(アプリ)
client. websocket_connect('/')をwebsocketとして使用します。
data=websocket。受信テキスト()
assert data=='こんにちは、世界!'
セッション上の操作は、待機可能なオブジェクトではなく、標準の関数呼び出しです。
コンテキスト管理された with ブロック内でセッションを使用することが重要です。これにより、AS GIアプリケーションが存在するバックグラウンドスレッドが正しく終了し、アプリケーション内で発生する例外が常にテストクライアントによってトリガーされるようになります。
テストセッションを作成します
.websocket_connect(url, subprotocols=None, **options)-httpx.get()と同じパラメーターのセットを使用します。
アプリケーションがWebSocket接続を受け入れない場合、starlette.websockets.WebSocketDisconnect が発生する可能性があります。
websocket_connect() は(with ブロック内で)コンテキストマネージャとして使用する必要があります。
!!! 注意 params パラメータは websocket_connect ではサポートされていません。クエリパラメータを渡す必要がある場合は、URLで直接ハードコーディングしてください。
```python
with client.websocket_connect('/path?foo=bar') as websocket:
...
```
データの送信
.send_text(data)-与えられたテキストをアプリケーションに送信します。.send_bytes(data)-与えられたバイトをアプリケーションに送信します。.send_json(data, mode="text")-与えられたデータをアプリケーションに送信します。使用mode="binary"JSONをバイナリデータフレームで送信します。
データの受信
.receive_text()-アプリケーションが送信した着信テキストを待ち、それを返します。.receive_bytes()-アプリケーションが送信した着信バイト列を待ち、それを返します。.receive_json(mode="text")・アプリケーションから送られてきた着信jsonデータを待って返します。を使ってmode="binary"JSONをバイナリデータフレームで受信します。
テキストは、"5月のレイズ starlette.websockets.WebSocketDisconnect .」は通常の意味表現にあまり合わないようで、正確な中国語翻訳が難しい。もしあなたがもっと文脈情報を持っていたり、テキスト内容の正確性を確認したりしたら、もっと満足できる翻訳を提供できるように努力します。現在、字義的に直訳すると、「向上する可能性がある starlette.websockets.WebSocketDisconnect しかし、この翻訳は実際の意味に合わないかもしれません。
接続を閉じる
.close(code=1000)・クライアントによるWebSocket接続のオフ操作を実行します。
非同期テスト
アプリケーションの外部で非同期操作を実行したい場合があります。たとえば、既存の非同期データベースクライアント/インフラストラクチャを使用してアプリケーションを呼び出した後、データベースの状態を確認したい場合があります。
これらの場合、TestClient を使用すると、独自のイベントループが作成され、データベース接続などの非同期リソースをイベントループ間で共有できないことがよくあります。
この問題を解決する最も簡単な方法は、テスト全体を非同期にし、httpx. AsyncClientなどの非同期クライアントを使用することです。
このようなテストの例は次のとおりです。
from httpxインポートAsyncClient
Starlette. applicationからStarletteをインポート
from starlette. routingインポートルート
インポートリクエスト
starletteからPlainTextResponseをインポート
def hello(request: Request)->PlainTextResponse:
return PlainTextResponse("Hello World!")
app=Starlette(Route=[Route("/", hello)])
pytestを使用している場合は、次のような非同期マーカーを追加する必要があります:
#@pytest. mark.anyio#https://github.com/agronholm/anyio
pytest-asyncioをインストールして設定する(https://github.com/pytest-dev/pytest-asyncio)
async def test_app()->None:
#ノート:あなた _必ず_ セット `base_url` "/"のような相対的なURLが機能するために
AsyncClient(app=app, base_url="http://test server")をクライアントとして非同期化します。
r=await client. get("/")
アサートr. status_code==200
assert r. text=="こんにちは、世界!"