https://github.com/Kludex/starlette/pull/1519#issuecomment-1236166180
()
| 1018 | |
| 1019 | |
| 1020 | def test_pr_1519_comment_1236166180_example() -> None: |
| 1021 | """ |
| 1022 | https://github.com/Kludex/starlette/pull/1519#issuecomment-1236166180 |
| 1023 | """ |
| 1024 | bodies: list[bytes] = [] |
| 1025 | |
| 1026 | class LogRequestBodySize(BaseHTTPMiddleware): |
| 1027 | async def dispatch( |
| 1028 | self, |
| 1029 | request: Request, |
| 1030 | call_next: RequestResponseEndpoint, |
| 1031 | ) -> Response: |
| 1032 | print(len(await request.body())) |
| 1033 | return await call_next(request) |
| 1034 | |
| 1035 | def replace_body_middleware(app: ASGIApp) -> ASGIApp: |
| 1036 | async def wrapped_app(scope: Scope, receive: Receive, send: Send) -> None: |
| 1037 | async def wrapped_rcv() -> Message: |
| 1038 | msg = await receive() |
| 1039 | msg["body"] += b"-foo" |
| 1040 | return msg |
| 1041 | |
| 1042 | await app(scope, wrapped_rcv, send) |
| 1043 | |
| 1044 | return wrapped_app |
| 1045 | |
| 1046 | async def endpoint(request: Request) -> Response: |
| 1047 | body = await request.body() |
| 1048 | bodies.append(body) |
| 1049 | return Response() |
| 1050 | |
| 1051 | app: ASGIApp = Starlette(routes=[Route("/", endpoint, methods=["POST"])]) |
| 1052 | app = replace_body_middleware(app) |
| 1053 | app = LogRequestBodySize(app) |
| 1054 | |
| 1055 | client = TestClient(app) |
| 1056 | resp = client.post("/", content=b"Hello, World!") |
| 1057 | resp.raise_for_status() |
| 1058 | |
| 1059 | assert bodies == [b"Hello, World!-foo"] |
| 1060 | |
| 1061 | |
| 1062 | @pytest.mark.anyio |
nothing calls this directly
no test coverage detected