(http_protocol_cls: type[HTTPProtocol])
| 369 | |
| 370 | |
| 371 | async def test_post_request(http_protocol_cls: type[HTTPProtocol]): |
| 372 | async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable): |
| 373 | body = b"" |
| 374 | more_body = True |
| 375 | while more_body: |
| 376 | message = await receive() |
| 377 | assert message["type"] == "http.request" |
| 378 | body += message.get("body", b"") |
| 379 | more_body = message.get("more_body", False) |
| 380 | response = Response(b"Body: " + body, media_type="text/plain") |
| 381 | await response(scope, receive, send) |
| 382 | |
| 383 | protocol = get_connected_protocol(app, http_protocol_cls) |
| 384 | protocol.data_received(SIMPLE_POST_REQUEST) |
| 385 | await protocol.loop.run_one() |
| 386 | assert b"HTTP/1.1 200 OK" in protocol.transport.buffer |
| 387 | assert b'Body: {"hello": "world"}' in protocol.transport.buffer |
| 388 | |
| 389 | |
| 390 | async def test_keepalive(http_protocol_cls: type[HTTPProtocol]): |
nothing calls this directly
no test coverage detected