import asyncio
+import http
import io
import json
import threading
self.session = session
+def _get_reason_phrase(status_code):
+ try:
+ return http.HTTPStatus(status_code).phrase
+ except ValueError:
+ return ""
+
+
class _ASGIAdapter(requests.adapters.HTTPAdapter):
def __init__(self, app: typing.Callable, raise_server_exceptions=True) -> None:
self.app = app
if message["type"] == "http.response.start":
raw_kwargs["version"] = 11
raw_kwargs["status"] = message["status"]
+ raw_kwargs["reason"] = _get_reason_phrase(message["status"])
raw_kwargs["headers"] = [
(key.decode(), value.decode()) for key, value in message["headers"]
]
raw_kwargs = {
"version": 11,
"status": 500,
+ "reason": "Internal Server Error",
"headers": [],
"preload_content": False,
"original_response": _MockOriginalResponse([]),
assert response.headers["x-header-2"] == "789"
+
+def test_response_phrase():
+ def app(scope):
+ return Response(b'', status_code=200)
+
+ client = TestClient(app)
+ response = client.get("/")
+ assert response.reason == "OK"
+
+ def app(scope):
+ return Response(b'', status_code=123)
+
+ client = TestClient(app)
+ response = client.get("/")
+ assert response.reason == ""
+
+
def test_file_response(tmpdir):
path = os.path.join(tmpdir, "xyz")
content = b"<file content>" * 1000