(test_client_factory: TestClientFactory)
| 146 | |
| 147 | |
| 148 | def test_request_body_then_stream(test_client_factory: TestClientFactory) -> None: |
| 149 | async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| 150 | request = Request(scope, receive) |
| 151 | body = await request.body() |
| 152 | chunks = b"" |
| 153 | async for chunk in request.stream(): |
| 154 | chunks += chunk |
| 155 | response = JSONResponse({"body": body.decode(), "stream": chunks.decode()}) |
| 156 | await response(scope, receive, send) |
| 157 | |
| 158 | client = test_client_factory(app) |
| 159 | |
| 160 | response = client.post("/", data="abc") # type: ignore |
| 161 | assert response.json() == {"body": "abc", "stream": "abc"} |
| 162 | |
| 163 | |
| 164 | def test_request_stream_then_body(test_client_factory: TestClientFactory) -> None: |
nothing calls this directly
no test coverage detected