if isinstance(event, h11.InformationalResponse):
event = await self._receive_event()
assert isinstance(event, h11.Response)
+ reason = event.reason.decode('latin1')
status_code = event.status_code
headers = event.headers
body = self._body_iter()
return Response(
- status_code=status_code, headers=headers, body=body, on_close=self._release
+ status_code=status_code, reason=reason, headers=headers, body=body, on_close=self._release
)
async def _body_iter(self) -> typing.AsyncIterator[bytes]:
+import http
import typing
from urllib.parse import urlsplit
self,
status_code: int,
*,
+ reason: typing.Optional[str] = None,
headers: typing.Sequence[typing.Tuple[bytes, bytes]] = (),
body: typing.Union[bytes, typing.AsyncIterator[bytes]] = b"",
on_close: typing.Callable = None,
):
self.status_code = status_code
+ if not reason:
+ try:
+ self.reason = http.HTTPStatus(status_code).phrase
+ except ValueError as exc:
+ self.reason = ""
+ else:
+ self.reason = reason
self.headers = list(headers)
self.on_close = on_close
self.is_closed = False
async def test_request():
response = await http.request("GET", "http://example.com")
assert response.status_code == 200
+ assert response.reason == "OK"
assert response.body == b"Hello, world!"
assert response.is_closed
with pytest.raises(httpcore.ResponseClosed):
await response.read()
+
+
+def test_unknown_status_code():
+ response = httpcore.Response(600)
+ assert response.status_code == 600
+ assert response.reason == ""