(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 = bclass="st">"" |
| 374 | more_body = True |
| 375 | while more_body: |
| 376 | message = await receive() |
| 377 | assert message[class="st">"type"] == class="st">"http.request" |
| 378 | body += message.get(class="st">"body", bclass="st">"") |
| 379 | more_body = message.get(class="st">"more_body", False) |
| 380 | response = Response(bclass="st">"Body: " + body, media_type=class="st">"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 bclass="st">"HTTP/1.1 200 OK" in protocol.transport.buffer |
| 387 | assert b&class="cm">#x27;Body: {class="st">"hello": class="st">"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