MCPcopy
hub / github.com/encode/starlette / test_pr_1519_comment_1236166180_example

Function test_pr_1519_comment_1236166180_example

tests/middleware/test_base.py:1020–1059  ·  view source on GitHub ↗

https://github.com/Kludex/starlette/pull/1519#issuecomment-1236166180

()

Source from the content-addressed store, hash-verified

1018
1019
1020def 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

Callers

nothing calls this directly

Calls 6

postMethod · 0.95
StarletteClass · 0.90
RouteClass · 0.90
TestClientClass · 0.90
replace_body_middlewareFunction · 0.85
LogRequestBodySizeClass · 0.85

Tested by

no test coverage detected