async def send(message: Message) -> None:
if message["type"] == "http.response.body":
- if not message.get("more_body", False):
+ if not message.get("more_body", False): # pragma: no branch
response_complete.set()
await app(scope, receive, send)
async def send(message: Message) -> None:
if message["type"] == "http.response.body":
- if not message.get("more_body", False):
+ if not message.get("more_body", False): # pragma: no branch
response_complete.set()
await app(scope, receive, send)
task_status.started()
while True:
message = await receive()
- if message["type"] == "http.disconnect":
+ if message["type"] == "http.disconnect": # pragma: no branch
task_group.cancel_scope.cancel()
break
async def test_read_request_stream_in_dispatch_wrapping_app_calls_body() -> None:
async def endpoint(scope: Scope, receive: Receive, send: Send) -> None:
request = Request(scope, receive)
- async for chunk in request.stream():
+ async for chunk in request.stream(): # pragma: no branch
assert chunk == b"2"
break
await Response()(scope, receive, send)
) -> Response:
expected = b"1"
response: Response | None = None
- async for chunk in request.stream():
+ async for chunk in request.stream(): # pragma: no branch
assert chunk == expected
if expected == b"1":
response = await call_next(request)
async def wrapped_app(scope: Scope, receive: Receive, send: Send) -> None:
async def wrapped_receive() -> Message:
msg = await receive()
- if msg["type"] == "http.request":
+ if msg["type"] == "http.request": # pragma: no branch
msg["body"] = msg["body"] * 2
return msg
id="staticmethod",
),
pytest.param(Endpoint(), "Endpoint", id="object"),
- pytest.param(lambda request: ..., "<lambda>", id="lambda"),
+ pytest.param(lambda request: ..., "<lambda>", id="lambda"), # pragma: no branch
],
)
def test_route_name(endpoint: typing.Callable[..., Response], expected_name: str) -> None: