Implement EmptyResponse class

starlette/responses.py

@@ -129,14 +129,64 @@ def delete_cookie(self, key: str, path: str = "/", domain: str = None) -> None:
         self.set_cookie(key, expires=0, max_age=0, path=path, domain=domain)
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if self.status_code < 200:
+            # Response is terminated after the status line.  So no headers and no body.
+            # https://datatracker.ietf.org/doc/html/rfc7231#section-6.2
+            raw_headers = []
+            body = b""
+        elif self.status_code == 204:
+            # Response must not have a content-length header. See
+            # https://datatracker.ietf.org/doc/html/rfc7230#section-3.3.2
+            # http spec does not appear to say whether or not there can be
+            # a content-type header. Some clients
+            # will attempt to parse the message body if there is a content-type
+            # header, so we ensure that there isn't one in
+            # the response.
+            raw_headers = [
+                header
+                for header in self.raw_headers
+                if header[0] not in (b"content-length", b"content-type")
+            ]
+            body = b""
+        elif self.status_code == 205:
+            # Response must not include a body.
+            # Response can either have a content-length: 0 header or a
+            # transfer-encoding: chunked header.
+            # We check for a transfer-encoding header.  If not found,
+            # we ensure the presence of the content-length header.
+            # https://datatracker.ietf.org/doc/html/rfc7231#section-6.3.6
+            raw_headers = [
+                header
+                for header in self.raw_headers
+                if header[0] not in (b"content-length", b"content-type")
+            ]
+            for header in self.raw_headers:
+                if header[0] == b"transfer-encoding":
+                    break
+            else:
+                raw_headers.append((b"content-length", b"0"))
+            body = b""
+        elif self.status_code == 304:
+            # A 304 Not Modfied response may contain a transfer-encoding header,
+            # or content-length header whose value is the length of
+            # message that would have been sent in a 200 OK response.
+            # So we leave the headers as is.
+            # https://datatracker.ietf.org/doc/html/rfc7230#section-3.3.1
+            # https://datatracker.ietf.org/doc/html/rfc7230#section-3.3.2
+            raw_headers = self.raw_headers
+            body = b""
+        else:
+            raw_headers = self.raw_headers
+            body = self.body
+
         await send(
             {
                 "type": "http.response.start",
                 "status": self.status_code,
-                "headers": self.raw_headers,
+                "headers": raw_headers,
             }
         )
-        await send({"type": "http.response.body", "body": self.body})
+        await send({"type": "http.response.body", "body": body})
 
         if self.background is not None:
             await self.background()

tests/test_responses.py

@@ -1,3 +1,4 @@
+import itertools
 import os
 
 import anyio
@@ -309,3 +310,127 @@ def test_head_method(test_client_factory):
     client = test_client_factory(app)
     response = client.head("/")
     assert response.text == ""
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "response_cls, status_code",
+    list(itertools.product([Response, JSONResponse], [100, 101, 102])),
+)
+async def test_response_1xx(response_cls, status_code):
+    scope = {}
+
+    async def receive():
+        return {}  # pragma: no cover
+
+    async def send(message: dict):
+        if message["type"] == "http.response.start":
+            # also ensures that self.raw_headers is not None
+            assert len(message["headers"]) == 0
+        elif message["type"] == "http.response.body":
+            # per ASGI, if body key is missing, default is False
+            assert "body" not in message or message["body"] == b""
+            assert "more_body" not in message or message["more_body"] is False
+        else:
+            pass  # pragma: no cover
+
+    response = response_cls(status_code=status_code)
+    await response.__call__(scope, receive, send)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "response_cls, content",
+    itertools.product([Response, JSONResponse], [None, "test"]),
+)
+async def test_response_204(response_cls, content):
+    scope = {}
+
+    async def receive():
+        return {}  # pragma: no cover
+
+    async def send(message: dict):
+        if message["type"] == "http.response.start":
+            header_map = dict(message["headers"])
+            assert b"content-length" not in header_map
+            assert b"content-type" not in header_map
+        elif message["type"] == "http.response.body":
+            # per ASGI, if body key is missing, default is False
+            assert "body" not in message or message["body"] == b""
+            assert "more_body" not in message or message["more_body"] is False
+        else:
+            pass  # pragma: no cover
+
+    response = response_cls(status_code=204, content=content)
+    await response.__call__(scope, receive, send)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("response_cls", [Response, JSONResponse])
+async def test_response_205_with_te_header(response_cls):
+    scope = {}
+
+    async def receive():
+        return {}  # pragma: no cover
+
+    async def send(message: dict):
+        if message["type"] == "http.response.start":
+            header_map = dict(message["headers"])
+            assert header_map[b"transfer-encoding"] == b"chunked"
+            assert b"content-length" not in header_map
+            assert b"content-type" not in header_map
+        elif message["type"] == "http.response.body":
+            # per ASGI, if body key is missing, default is False
+            assert "body" not in message or message["body"] == b""
+            assert "more_body" not in message or message["more_body"] is False
+        else:
+            pass  # pragma: no cover
+
+    response = response_cls(status_code=205, headers={"transfer-encoding": "chunked"})
+    await response.__call__(scope, receive, send)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("response_cls", [Response, JSONResponse])
+async def test_response_205_with_cl_header(response_cls):
+    scope = {}
+
+    async def receive():
+        return {}  # pragma: no cover
+
+    async def send(message: dict):
+        if message["type"] == "http.response.start":
+            header_map = dict(message["headers"])
+            assert header_map[b"content-length"] == b"0"
+            assert b"content-type" not in header_map
+        elif message["type"] == "http.response.body":
+            # per ASGI, if body key is missing, default is False
+            assert "body" not in message or message["body"] == b""
+            assert "more_body" not in message or message["more_body"] is False
+        else:
+            pass  # pragma: no cover
+
+    response = response_cls(status_code=205)
+    await response.__call__(scope, receive, send)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("response_cls", [Response, JSONResponse])
+async def test_response_304(response_cls):
+    scope = {}
+
+    async def receive():
+        return {}  # pragma: no cover
+
+    async def send(message: dict):
+        if message["type"] == "http.response.start":
+            pass
+        elif message["type"] == "http.response.body":
+            # per ASGI, 'body', 'more_body' are optional.
+            assert "body" not in message or message["body"] == b""
+            assert "more_body" not in message or message["more_body"] is False
+        else:
+            pass  # pragma: no cover
+
+    response = response_cls(status_code=304)
+    await response.__call__(scope, receive, send)