We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
The response.content_type is not add to headers in ASGI
Perhaps the response.content_type is add to headers here.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sanic/asgi.py`
Content:
```
1 import asyncio
2 import warnings
3 
4 from inspect import isawaitable
5 from typing import Any, Awaitable, Callable, MutableMapping, Union
6 from urllib.parse import quote
7 
8 from multidict import CIMultiDict
9 
10 from sanic.exceptions import InvalidUsage, ServerError
11 from sanic.log import logger
12 from sanic.request import Request
13 from sanic.response import HTTPResponse, StreamingHTTPResponse
14 from sanic.server import StreamBuffer
15 from sanic.websocket import WebSocketConnection
16 
17 
18 ASGIScope = MutableMapping[str, Any]
19 ASGIMessage = MutableMapping[str, Any]
20 ASGISend = Callable[[ASGIMessage], Awaitable[None]]
21 ASGIReceive = Callable[[], Awaitable[ASGIMessage]]
22 
23 
24 class MockProtocol:
25     def __init__(self, transport: "MockTransport", loop):
26         self.transport = transport
27         self._not_paused = asyncio.Event(loop=loop)
28         self._not_paused.set()
29         self._complete = asyncio.Event(loop=loop)
30 
31     def pause_writing(self) -> None:
32         self._not_paused.clear()
33 
34     def resume_writing(self) -> None:
35         self._not_paused.set()
36 
37     async def complete(self) -> None:
38         self._not_paused.set()
39         await self.transport.send(
40             {"type": "http.response.body", "body": b"", "more_body": False}
41         )
42 
43     @property
44     def is_complete(self) -> bool:
45         return self._complete.is_set()
46 
47     async def push_data(self, data: bytes) -> None:
48         if not self.is_complete:
49             await self.transport.send(
50                 {"type": "http.response.body", "body": data, "more_body": True}
51             )
52 
53     async def drain(self) -> None:
54         await self._not_paused.wait()
55 
56 
57 class MockTransport:
58     def __init__(
59         self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend
60     ) -> None:
61         self.scope = scope
62         self._receive = receive
63         self._send = send
64         self._protocol = None
65         self.loop = None
66 
67     def get_protocol(self) -> MockProtocol:
68         if not self._protocol:
69             self._protocol = MockProtocol(self, self.loop)
70         return self._protocol
71 
72     def get_extra_info(self, info: str) -> Union[str, bool]:
73         if info == "peername":
74             return self.scope.get("server")
75         elif info == "sslcontext":
76             return self.scope.get("scheme") in ["https", "wss"]
77 
78     def get_websocket_connection(self) -> WebSocketConnection:
79         try:
80             return self._websocket_connection
81         except AttributeError:
82             raise InvalidUsage("Improper websocket connection.")
83 
84     def create_websocket_connection(
85         self, send: ASGISend, receive: ASGIReceive
86     ) -> WebSocketConnection:
87         self._websocket_connection = WebSocketConnection(send, receive)
88         return self._websocket_connection
89 
90     def add_task(self) -> None:
91         raise NotImplementedError
92 
93     async def send(self, data) -> None:
94         # TODO:
95         # - Validation on data and that it is formatted properly and is valid
96         await self._send(data)
97 
98     async def receive(self) -> ASGIMessage:
99         return await self._receive()
100 
101 
102 class Lifespan:
103     def __init__(self, asgi_app: "ASGIApp") -> None:
104         self.asgi_app = asgi_app
105 
106         if "before_server_start" in self.asgi_app.sanic_app.listeners:
107             warnings.warn(
108                 'You have set a listener for "before_server_start" '
109                 "in ASGI mode. "
110                 "It will be executed as early as possible, but not before "
111                 "the ASGI server is started."
112             )
113         if "after_server_stop" in self.asgi_app.sanic_app.listeners:
114             warnings.warn(
115                 'You have set a listener for "after_server_stop" '
116                 "in ASGI mode. "
117                 "It will be executed as late as possible, but not after "
118                 "the ASGI server is stopped."
119             )
120 
121     async def startup(self) -> None:
122         """
123         Gather the listeners to fire on server start.
124         Because we are using a third-party server and not Sanic server, we do
125         not have access to fire anything BEFORE the server starts.
126         Therefore, we fire before_server_start and after_server_start
127         in sequence since the ASGI lifespan protocol only supports a single
128         startup event.
129         """
130         listeners = self.asgi_app.sanic_app.listeners.get(
131             "before_server_start", []
132         ) + self.asgi_app.sanic_app.listeners.get("after_server_start", [])
133 
134         for handler in listeners:
135             response = handler(
136                 self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop
137             )
138             if isawaitable(response):
139                 await response
140 
141     async def shutdown(self) -> None:
142         """
143         Gather the listeners to fire on server stop.
144         Because we are using a third-party server and not Sanic server, we do
145         not have access to fire anything AFTER the server stops.
146         Therefore, we fire before_server_stop and after_server_stop
147         in sequence since the ASGI lifespan protocol only supports a single
148         shutdown event.
149         """
150         listeners = self.asgi_app.sanic_app.listeners.get(
151             "before_server_stop", []
152         ) + self.asgi_app.sanic_app.listeners.get("after_server_stop", [])
153 
154         for handler in listeners:
155             response = handler(
156                 self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop
157             )
158             if isawaitable(response):
159                 await response
160 
161     async def __call__(
162         self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend
163     ) -> None:
164         message = await receive()
165         if message["type"] == "lifespan.startup":
166             await self.startup()
167             await send({"type": "lifespan.startup.complete"})
168 
169         message = await receive()
170         if message["type"] == "lifespan.shutdown":
171             await self.shutdown()
172             await send({"type": "lifespan.shutdown.complete"})
173 
174 
175 class ASGIApp:
176     def __init__(self) -> None:
177         self.ws = None
178 
179     @classmethod
180     async def create(
181         cls, sanic_app, scope: ASGIScope, receive: ASGIReceive, send: ASGISend
182     ) -> "ASGIApp":
183         instance = cls()
184         instance.sanic_app = sanic_app
185         instance.transport = MockTransport(scope, receive, send)
186         instance.transport.add_task = sanic_app.loop.create_task
187         instance.transport.loop = sanic_app.loop
188 
189         headers = CIMultiDict(
190             [
191                 (key.decode("latin-1"), value.decode("latin-1"))
192                 for key, value in scope.get("headers", [])
193             ]
194         )
195         instance.do_stream = (
196             True if headers.get("expect") == "100-continue" else False
197         )
198         instance.lifespan = Lifespan(instance)
199 
200         if scope["type"] == "lifespan":
201             await instance.lifespan(scope, receive, send)
202         else:
203             url_bytes = scope.get("root_path", "") + quote(scope["path"])
204             url_bytes = url_bytes.encode("latin-1")
205             url_bytes += b"?" + scope["query_string"]
206 
207             if scope["type"] == "http":
208                 version = scope["http_version"]
209                 method = scope["method"]
210             elif scope["type"] == "websocket":
211                 version = "1.1"
212                 method = "GET"
213 
214                 instance.ws = instance.transport.create_websocket_connection(
215                     send, receive
216                 )
217                 await instance.ws.accept()
218             else:
219                 pass
220                 # TODO:
221                 # - close connection
222 
223             request_class = sanic_app.request_class or Request
224             instance.request = request_class(
225                 url_bytes,
226                 headers,
227                 version,
228                 method,
229                 instance.transport,
230                 sanic_app,
231             )
232 
233             if sanic_app.is_request_stream:
234                 is_stream_handler = sanic_app.router.is_stream_handler(
235                     instance.request
236                 )
237                 if is_stream_handler:
238                     instance.request.stream = StreamBuffer(
239                         sanic_app.config.REQUEST_BUFFER_QUEUE_SIZE
240                     )
241                     instance.do_stream = True
242 
243         return instance
244 
245     async def read_body(self) -> bytes:
246         """
247         Read and return the entire body from an incoming ASGI message.
248         """
249         body = b""
250         more_body = True
251         while more_body:
252             message = await self.transport.receive()
253             body += message.get("body", b"")
254             more_body = message.get("more_body", False)
255 
256         return body
257 
258     async def stream_body(self) -> None:
259         """
260         Read and stream the body in chunks from an incoming ASGI message.
261         """
262         more_body = True
263 
264         while more_body:
265             message = await self.transport.receive()
266             chunk = message.get("body", b"")
267             await self.request.stream.put(chunk)
268 
269             more_body = message.get("more_body", False)
270 
271         await self.request.stream.put(None)
272 
273     async def __call__(self) -> None:
274         """
275         Handle the incoming request.
276         """
277         if not self.do_stream:
278             self.request.body = await self.read_body()
279         else:
280             self.sanic_app.loop.create_task(self.stream_body())
281 
282         handler = self.sanic_app.handle_request
283         callback = None if self.ws else self.stream_callback
284         await handler(self.request, None, callback)
285 
286     async def stream_callback(self, response: HTTPResponse) -> None:
287         """
288         Write the response.
289         """
290         headers = []
291         cookies = {}
292         try:
293             cookies = {
294                 v.key: v
295                 for _, v in list(
296                     filter(
297                         lambda item: item[0].lower() == "set-cookie",
298                         response.headers.items(),
299                     )
300                 )
301             }
302             headers += [
303                 (str(name).encode("latin-1"), str(value).encode("latin-1"))
304                 for name, value in response.headers.items()
305                 if name.lower() not in ["set-cookie"]
306             ]
307         except AttributeError:
308             logger.error(
309                 "Invalid response object for url %s, "
310                 "Expected Type: HTTPResponse, Actual Type: %s",
311                 self.request.url,
312                 type(response),
313             )
314             exception = ServerError("Invalid response type")
315             response = self.sanic_app.error_handler.response(
316                 self.request, exception
317             )
318             headers = [
319                 (str(name).encode("latin-1"), str(value).encode("latin-1"))
320                 for name, value in response.headers.items()
321                 if name not in (b"Set-Cookie",)
322             ]
323 
324         if "content-length" not in response.headers and not isinstance(
325             response, StreamingHTTPResponse
326         ):
327             headers += [
328                 (b"content-length", str(len(response.body)).encode("latin-1"))
329             ]
330 
331         if response.cookies:
332             cookies.update(
333                 {
334                     v.key: v
335                     for _, v in response.cookies.items()
336                     if v.key not in cookies.keys()
337                 }
338             )
339 
340         headers += [
341             (b"set-cookie", cookie.encode("utf-8"))
342             for k, cookie in cookies.items()
343         ]
344 
345         await self.transport.send(
346             {
347                 "type": "http.response.start",
348                 "status": response.status,
349                 "headers": headers,
350             }
351         )
352 
353         if isinstance(response, StreamingHTTPResponse):
354             response.protocol = self.transport.get_protocol()
355             await response.stream()
356             await response.protocol.complete()
357 
358         else:
359             await self.transport.send(
360                 {
361                     "type": "http.response.body",
362                     "body": response.body,
363                     "more_body": False,
364                 }
365             )
366
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sanic/asgi.py b/sanic/asgi.py
--- a/sanic/asgi.py
+++ b/sanic/asgi.py
@@ -328,6 +328,11 @@
                 (b"content-length", str(len(response.body)).encode("latin-1"))
             ]
 
+        if "content-type" not in response.headers:
+            headers += [
+                (b"content-type", str(response.content_type).encode("latin-1"))
+            ]
+
         if response.cookies:
             cookies.update(
                 {

{"golden_diff": "diff --git a/sanic/asgi.py b/sanic/asgi.py\n--- a/sanic/asgi.py\n+++ b/sanic/asgi.py\n@@ -328,6 +328,11 @@\n                 (b\"content-length\", str(len(response.body)).encode(\"latin-1\"))\n             ]\n \n+        if \"content-type\" not in response.headers:\n+            headers += [\n+                (b\"content-type\", str(response.content_type).encode(\"latin-1\"))\n+            ]\n+\n         if response.cookies:\n             cookies.update(\n                 {\n", "issue": "The response.content_type is not add to headers in ASGI\nPerhaps the response.content_type is add to headers here.\n", "before_files": [{"content": "import asyncio\nimport warnings\n\nfrom inspect import isawaitable\nfrom typing import Any, Awaitable, Callable, MutableMapping, Union\nfrom urllib.parse import quote\n\nfrom multidict import CIMultiDict\n\nfrom sanic.exceptions import InvalidUsage, ServerError\nfrom sanic.log import logger\nfrom sanic.request import Request\nfrom sanic.response import HTTPResponse, StreamingHTTPResponse\nfrom sanic.server import StreamBuffer\nfrom sanic.websocket import WebSocketConnection\n\n\nASGIScope = MutableMapping[str, Any]\nASGIMessage = MutableMapping[str, Any]\nASGISend = Callable[[ASGIMessage], Awaitable[None]]\nASGIReceive = Callable[[], Awaitable[ASGIMessage]]\n\n\nclass MockProtocol:\n    def __init__(self, transport: \"MockTransport\", loop):\n        self.transport = transport\n        self._not_paused = asyncio.Event(loop=loop)\n        self._not_paused.set()\n        self._complete = asyncio.Event(loop=loop)\n\n    def pause_writing(self) -> None:\n        self._not_paused.clear()\n\n    def resume_writing(self) -> None:\n        self._not_paused.set()\n\n    async def complete(self) -> None:\n        self._not_paused.set()\n        await self.transport.send(\n            {\"type\": \"http.response.body\", \"body\": b\"\", \"more_body\": False}\n        )\n\n    @property\n    def is_complete(self) -> bool:\n        return self._complete.is_set()\n\n    async def push_data(self, data: bytes) -> None:\n        if not self.is_complete:\n            await self.transport.send(\n                {\"type\": \"http.response.body\", \"body\": data, \"more_body\": True}\n            )\n\n    async def drain(self) -> None:\n        await self._not_paused.wait()\n\n\nclass MockTransport:\n    def __init__(\n        self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> None:\n        self.scope = scope\n        self._receive = receive\n        self._send = send\n        self._protocol = None\n        self.loop = None\n\n    def get_protocol(self) -> MockProtocol:\n        if not self._protocol:\n            self._protocol = MockProtocol(self, self.loop)\n        return self._protocol\n\n    def get_extra_info(self, info: str) -> Union[str, bool]:\n        if info == \"peername\":\n            return self.scope.get(\"server\")\n        elif info == \"sslcontext\":\n            return self.scope.get(\"scheme\") in [\"https\", \"wss\"]\n\n    def get_websocket_connection(self) -> WebSocketConnection:\n        try:\n            return self._websocket_connection\n        except AttributeError:\n            raise InvalidUsage(\"Improper websocket connection.\")\n\n    def create_websocket_connection(\n        self, send: ASGISend, receive: ASGIReceive\n    ) -> WebSocketConnection:\n        self._websocket_connection = WebSocketConnection(send, receive)\n        return self._websocket_connection\n\n    def add_task(self) -> None:\n        raise NotImplementedError\n\n    async def send(self, data) -> None:\n        # TODO:\n        # - Validation on data and that it is formatted properly and is valid\n        await self._send(data)\n\n    async def receive(self) -> ASGIMessage:\n        return await self._receive()\n\n\nclass Lifespan:\n    def __init__(self, asgi_app: \"ASGIApp\") -> None:\n        self.asgi_app = asgi_app\n\n        if \"before_server_start\" in self.asgi_app.sanic_app.listeners:\n            warnings.warn(\n                'You have set a listener for \"before_server_start\" '\n                \"in ASGI mode. \"\n                \"It will be executed as early as possible, but not before \"\n                \"the ASGI server is started.\"\n            )\n        if \"after_server_stop\" in self.asgi_app.sanic_app.listeners:\n            warnings.warn(\n                'You have set a listener for \"after_server_stop\" '\n                \"in ASGI mode. \"\n                \"It will be executed as late as possible, but not after \"\n                \"the ASGI server is stopped.\"\n            )\n\n    async def startup(self) -> None:\n        \"\"\"\n        Gather the listeners to fire on server start.\n        Because we are using a third-party server and not Sanic server, we do\n        not have access to fire anything BEFORE the server starts.\n        Therefore, we fire before_server_start and after_server_start\n        in sequence since the ASGI lifespan protocol only supports a single\n        startup event.\n        \"\"\"\n        listeners = self.asgi_app.sanic_app.listeners.get(\n            \"before_server_start\", []\n        ) + self.asgi_app.sanic_app.listeners.get(\"after_server_start\", [])\n\n        for handler in listeners:\n            response = handler(\n                self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop\n            )\n            if isawaitable(response):\n                await response\n\n    async def shutdown(self) -> None:\n        \"\"\"\n        Gather the listeners to fire on server stop.\n        Because we are using a third-party server and not Sanic server, we do\n        not have access to fire anything AFTER the server stops.\n        Therefore, we fire before_server_stop and after_server_stop\n        in sequence since the ASGI lifespan protocol only supports a single\n        shutdown event.\n        \"\"\"\n        listeners = self.asgi_app.sanic_app.listeners.get(\n            \"before_server_stop\", []\n        ) + self.asgi_app.sanic_app.listeners.get(\"after_server_stop\", [])\n\n        for handler in listeners:\n            response = handler(\n                self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop\n            )\n            if isawaitable(response):\n                await response\n\n    async def __call__(\n        self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> None:\n        message = await receive()\n        if message[\"type\"] == \"lifespan.startup\":\n            await self.startup()\n            await send({\"type\": \"lifespan.startup.complete\"})\n\n        message = await receive()\n        if message[\"type\"] == \"lifespan.shutdown\":\n            await self.shutdown()\n            await send({\"type\": \"lifespan.shutdown.complete\"})\n\n\nclass ASGIApp:\n    def __init__(self) -> None:\n        self.ws = None\n\n    @classmethod\n    async def create(\n        cls, sanic_app, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> \"ASGIApp\":\n        instance = cls()\n        instance.sanic_app = sanic_app\n        instance.transport = MockTransport(scope, receive, send)\n        instance.transport.add_task = sanic_app.loop.create_task\n        instance.transport.loop = sanic_app.loop\n\n        headers = CIMultiDict(\n            [\n                (key.decode(\"latin-1\"), value.decode(\"latin-1\"))\n                for key, value in scope.get(\"headers\", [])\n            ]\n        )\n        instance.do_stream = (\n            True if headers.get(\"expect\") == \"100-continue\" else False\n        )\n        instance.lifespan = Lifespan(instance)\n\n        if scope[\"type\"] == \"lifespan\":\n            await instance.lifespan(scope, receive, send)\n        else:\n            url_bytes = scope.get(\"root_path\", \"\") + quote(scope[\"path\"])\n            url_bytes = url_bytes.encode(\"latin-1\")\n            url_bytes += b\"?\" + scope[\"query_string\"]\n\n            if scope[\"type\"] == \"http\":\n                version = scope[\"http_version\"]\n                method = scope[\"method\"]\n            elif scope[\"type\"] == \"websocket\":\n                version = \"1.1\"\n                method = \"GET\"\n\n                instance.ws = instance.transport.create_websocket_connection(\n                    send, receive\n                )\n                await instance.ws.accept()\n            else:\n                pass\n                # TODO:\n                # - close connection\n\n            request_class = sanic_app.request_class or Request\n            instance.request = request_class(\n                url_bytes,\n                headers,\n                version,\n                method,\n                instance.transport,\n                sanic_app,\n            )\n\n            if sanic_app.is_request_stream:\n                is_stream_handler = sanic_app.router.is_stream_handler(\n                    instance.request\n                )\n                if is_stream_handler:\n                    instance.request.stream = StreamBuffer(\n                        sanic_app.config.REQUEST_BUFFER_QUEUE_SIZE\n                    )\n                    instance.do_stream = True\n\n        return instance\n\n    async def read_body(self) -> bytes:\n        \"\"\"\n        Read and return the entire body from an incoming ASGI message.\n        \"\"\"\n        body = b\"\"\n        more_body = True\n        while more_body:\n            message = await self.transport.receive()\n            body += message.get(\"body\", b\"\")\n            more_body = message.get(\"more_body\", False)\n\n        return body\n\n    async def stream_body(self) -> None:\n        \"\"\"\n        Read and stream the body in chunks from an incoming ASGI message.\n        \"\"\"\n        more_body = True\n\n        while more_body:\n            message = await self.transport.receive()\n            chunk = message.get(\"body\", b\"\")\n            await self.request.stream.put(chunk)\n\n            more_body = message.get(\"more_body\", False)\n\n        await self.request.stream.put(None)\n\n    async def __call__(self) -> None:\n        \"\"\"\n        Handle the incoming request.\n        \"\"\"\n        if not self.do_stream:\n            self.request.body = await self.read_body()\n        else:\n            self.sanic_app.loop.create_task(self.stream_body())\n\n        handler = self.sanic_app.handle_request\n        callback = None if self.ws else self.stream_callback\n        await handler(self.request, None, callback)\n\n    async def stream_callback(self, response: HTTPResponse) -> None:\n        \"\"\"\n        Write the response.\n        \"\"\"\n        headers = []\n        cookies = {}\n        try:\n            cookies = {\n                v.key: v\n                for _, v in list(\n                    filter(\n                        lambda item: item[0].lower() == \"set-cookie\",\n                        response.headers.items(),\n                    )\n                )\n            }\n            headers += [\n                (str(name).encode(\"latin-1\"), str(value).encode(\"latin-1\"))\n                for name, value in response.headers.items()\n                if name.lower() not in [\"set-cookie\"]\n            ]\n        except AttributeError:\n            logger.error(\n                \"Invalid response object for url %s, \"\n                \"Expected Type: HTTPResponse, Actual Type: %s\",\n                self.request.url,\n                type(response),\n            )\n            exception = ServerError(\"Invalid response type\")\n            response = self.sanic_app.error_handler.response(\n                self.request, exception\n            )\n            headers = [\n                (str(name).encode(\"latin-1\"), str(value).encode(\"latin-1\"))\n                for name, value in response.headers.items()\n                if name not in (b\"Set-Cookie\",)\n            ]\n\n        if \"content-length\" not in response.headers and not isinstance(\n            response, StreamingHTTPResponse\n        ):\n            headers += [\n                (b\"content-length\", str(len(response.body)).encode(\"latin-1\"))\n            ]\n\n        if response.cookies:\n            cookies.update(\n                {\n                    v.key: v\n                    for _, v in response.cookies.items()\n                    if v.key not in cookies.keys()\n                }\n            )\n\n        headers += [\n            (b\"set-cookie\", cookie.encode(\"utf-8\"))\n            for k, cookie in cookies.items()\n        ]\n\n        await self.transport.send(\n            {\n                \"type\": \"http.response.start\",\n                \"status\": response.status,\n                \"headers\": headers,\n            }\n        )\n\n        if isinstance(response, StreamingHTTPResponse):\n            response.protocol = self.transport.get_protocol()\n            await response.stream()\n            await response.protocol.complete()\n\n        else:\n            await self.transport.send(\n                {\n                    \"type\": \"http.response.body\",\n                    \"body\": response.body,\n                    \"more_body\": False,\n                }\n            )\n", "path": "sanic/asgi.py"}], "after_files": [{"content": "import asyncio\nimport warnings\n\nfrom inspect import isawaitable\nfrom typing import Any, Awaitable, Callable, MutableMapping, Union\nfrom urllib.parse import quote\n\nfrom multidict import CIMultiDict\n\nfrom sanic.exceptions import InvalidUsage, ServerError\nfrom sanic.log import logger\nfrom sanic.request import Request\nfrom sanic.response import HTTPResponse, StreamingHTTPResponse\nfrom sanic.server import StreamBuffer\nfrom sanic.websocket import WebSocketConnection\n\n\nASGIScope = MutableMapping[str, Any]\nASGIMessage = MutableMapping[str, Any]\nASGISend = Callable[[ASGIMessage], Awaitable[None]]\nASGIReceive = Callable[[], Awaitable[ASGIMessage]]\n\n\nclass MockProtocol:\n    def __init__(self, transport: \"MockTransport\", loop):\n        self.transport = transport\n        self._not_paused = asyncio.Event(loop=loop)\n        self._not_paused.set()\n        self._complete = asyncio.Event(loop=loop)\n\n    def pause_writing(self) -> None:\n        self._not_paused.clear()\n\n    def resume_writing(self) -> None:\n        self._not_paused.set()\n\n    async def complete(self) -> None:\n        self._not_paused.set()\n        await self.transport.send(\n            {\"type\": \"http.response.body\", \"body\": b\"\", \"more_body\": False}\n        )\n\n    @property\n    def is_complete(self) -> bool:\n        return self._complete.is_set()\n\n    async def push_data(self, data: bytes) -> None:\n        if not self.is_complete:\n            await self.transport.send(\n                {\"type\": \"http.response.body\", \"body\": data, \"more_body\": True}\n            )\n\n    async def drain(self) -> None:\n        await self._not_paused.wait()\n\n\nclass MockTransport:\n    def __init__(\n        self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> None:\n        self.scope = scope\n        self._receive = receive\n        self._send = send\n        self._protocol = None\n        self.loop = None\n\n    def get_protocol(self) -> MockProtocol:\n        if not self._protocol:\n            self._protocol = MockProtocol(self, self.loop)\n        return self._protocol\n\n    def get_extra_info(self, info: str) -> Union[str, bool]:\n        if info == \"peername\":\n            return self.scope.get(\"server\")\n        elif info == \"sslcontext\":\n            return self.scope.get(\"scheme\") in [\"https\", \"wss\"]\n\n    def get_websocket_connection(self) -> WebSocketConnection:\n        try:\n            return self._websocket_connection\n        except AttributeError:\n            raise InvalidUsage(\"Improper websocket connection.\")\n\n    def create_websocket_connection(\n        self, send: ASGISend, receive: ASGIReceive\n    ) -> WebSocketConnection:\n        self._websocket_connection = WebSocketConnection(send, receive)\n        return self._websocket_connection\n\n    def add_task(self) -> None:\n        raise NotImplementedError\n\n    async def send(self, data) -> None:\n        # TODO:\n        # - Validation on data and that it is formatted properly and is valid\n        await self._send(data)\n\n    async def receive(self) -> ASGIMessage:\n        return await self._receive()\n\n\nclass Lifespan:\n    def __init__(self, asgi_app: \"ASGIApp\") -> None:\n        self.asgi_app = asgi_app\n\n        if \"before_server_start\" in self.asgi_app.sanic_app.listeners:\n            warnings.warn(\n                'You have set a listener for \"before_server_start\" '\n                \"in ASGI mode. \"\n                \"It will be executed as early as possible, but not before \"\n                \"the ASGI server is started.\"\n            )\n        if \"after_server_stop\" in self.asgi_app.sanic_app.listeners:\n            warnings.warn(\n                'You have set a listener for \"after_server_stop\" '\n                \"in ASGI mode. \"\n                \"It will be executed as late as possible, but not after \"\n                \"the ASGI server is stopped.\"\n            )\n\n    async def startup(self) -> None:\n        \"\"\"\n        Gather the listeners to fire on server start.\n        Because we are using a third-party server and not Sanic server, we do\n        not have access to fire anything BEFORE the server starts.\n        Therefore, we fire before_server_start and after_server_start\n        in sequence since the ASGI lifespan protocol only supports a single\n        startup event.\n        \"\"\"\n        listeners = self.asgi_app.sanic_app.listeners.get(\n            \"before_server_start\", []\n        ) + self.asgi_app.sanic_app.listeners.get(\"after_server_start\", [])\n\n        for handler in listeners:\n            response = handler(\n                self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop\n            )\n            if isawaitable(response):\n                await response\n\n    async def shutdown(self) -> None:\n        \"\"\"\n        Gather the listeners to fire on server stop.\n        Because we are using a third-party server and not Sanic server, we do\n        not have access to fire anything AFTER the server stops.\n        Therefore, we fire before_server_stop and after_server_stop\n        in sequence since the ASGI lifespan protocol only supports a single\n        shutdown event.\n        \"\"\"\n        listeners = self.asgi_app.sanic_app.listeners.get(\n            \"before_server_stop\", []\n        ) + self.asgi_app.sanic_app.listeners.get(\"after_server_stop\", [])\n\n        for handler in listeners:\n            response = handler(\n                self.asgi_app.sanic_app, self.asgi_app.sanic_app.loop\n            )\n            if isawaitable(response):\n                await response\n\n    async def __call__(\n        self, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> None:\n        message = await receive()\n        if message[\"type\"] == \"lifespan.startup\":\n            await self.startup()\n            await send({\"type\": \"lifespan.startup.complete\"})\n\n        message = await receive()\n        if message[\"type\"] == \"lifespan.shutdown\":\n            await self.shutdown()\n            await send({\"type\": \"lifespan.shutdown.complete\"})\n\n\nclass ASGIApp:\n    def __init__(self) -> None:\n        self.ws = None\n\n    @classmethod\n    async def create(\n        cls, sanic_app, scope: ASGIScope, receive: ASGIReceive, send: ASGISend\n    ) -> \"ASGIApp\":\n        instance = cls()\n        instance.sanic_app = sanic_app\n        instance.transport = MockTransport(scope, receive, send)\n        instance.transport.add_task = sanic_app.loop.create_task\n        instance.transport.loop = sanic_app.loop\n\n        headers = CIMultiDict(\n            [\n                (key.decode(\"latin-1\"), value.decode(\"latin-1\"))\n                for key, value in scope.get(\"headers\", [])\n            ]\n        )\n        instance.do_stream = (\n            True if headers.get(\"expect\") == \"100-continue\" else False\n        )\n        instance.lifespan = Lifespan(instance)\n\n        if scope[\"type\"] == \"lifespan\":\n            await instance.lifespan(scope, receive, send)\n        else:\n            url_bytes = scope.get(\"root_path\", \"\") + quote(scope[\"path\"])\n            url_bytes = url_bytes.encode(\"latin-1\")\n            url_bytes += b\"?\" + scope[\"query_string\"]\n\n            if scope[\"type\"] == \"http\":\n                version = scope[\"http_version\"]\n                method = scope[\"method\"]\n            elif scope[\"type\"] == \"websocket\":\n                version = \"1.1\"\n                method = \"GET\"\n\n                instance.ws = instance.transport.create_websocket_connection(\n                    send, receive\n                )\n                await instance.ws.accept()\n            else:\n                pass\n                # TODO:\n                # - close connection\n\n            request_class = sanic_app.request_class or Request\n            instance.request = request_class(\n                url_bytes,\n                headers,\n                version,\n                method,\n                instance.transport,\n                sanic_app,\n            )\n\n            if sanic_app.is_request_stream:\n                is_stream_handler = sanic_app.router.is_stream_handler(\n                    instance.request\n                )\n                if is_stream_handler:\n                    instance.request.stream = StreamBuffer(\n                        sanic_app.config.REQUEST_BUFFER_QUEUE_SIZE\n                    )\n                    instance.do_stream = True\n\n        return instance\n\n    async def read_body(self) -> bytes:\n        \"\"\"\n        Read and return the entire body from an incoming ASGI message.\n        \"\"\"\n        body = b\"\"\n        more_body = True\n        while more_body:\n            message = await self.transport.receive()\n            body += message.get(\"body\", b\"\")\n            more_body = message.get(\"more_body\", False)\n\n        return body\n\n    async def stream_body(self) -> None:\n        \"\"\"\n        Read and stream the body in chunks from an incoming ASGI message.\n        \"\"\"\n        more_body = True\n\n        while more_body:\n            message = await self.transport.receive()\n            chunk = message.get(\"body\", b\"\")\n            await self.request.stream.put(chunk)\n\n            more_body = message.get(\"more_body\", False)\n\n        await self.request.stream.put(None)\n\n    async def __call__(self) -> None:\n        \"\"\"\n        Handle the incoming request.\n        \"\"\"\n        if not self.do_stream:\n            self.request.body = await self.read_body()\n        else:\n            self.sanic_app.loop.create_task(self.stream_body())\n\n        handler = self.sanic_app.handle_request\n        callback = None if self.ws else self.stream_callback\n        await handler(self.request, None, callback)\n\n    async def stream_callback(self, response: HTTPResponse) -> None:\n        \"\"\"\n        Write the response.\n        \"\"\"\n        headers = []\n        cookies = {}\n        try:\n            cookies = {\n                v.key: v\n                for _, v in list(\n                    filter(\n                        lambda item: item[0].lower() == \"set-cookie\",\n                        response.headers.items(),\n                    )\n                )\n            }\n            headers += [\n                (str(name).encode(\"latin-1\"), str(value).encode(\"latin-1\"))\n                for name, value in response.headers.items()\n                if name.lower() not in [\"set-cookie\"]\n            ]\n        except AttributeError:\n            logger.error(\n                \"Invalid response object for url %s, \"\n                \"Expected Type: HTTPResponse, Actual Type: %s\",\n                self.request.url,\n                type(response),\n            )\n            exception = ServerError(\"Invalid response type\")\n            response = self.sanic_app.error_handler.response(\n                self.request, exception\n            )\n            headers = [\n                (str(name).encode(\"latin-1\"), str(value).encode(\"latin-1\"))\n                for name, value in response.headers.items()\n                if name not in (b\"Set-Cookie\",)\n            ]\n\n        if \"content-length\" not in response.headers and not isinstance(\n            response, StreamingHTTPResponse\n        ):\n            headers += [\n                (b\"content-length\", str(len(response.body)).encode(\"latin-1\"))\n            ]\n\n        if \"content-type\" not in response.headers:\n            headers += [\n                (b\"content-type\", str(response.content_type).encode(\"latin-1\"))\n            ]\n\n        if response.cookies:\n            cookies.update(\n                {\n                    v.key: v\n                    for _, v in response.cookies.items()\n                    if v.key not in cookies.keys()\n                }\n            )\n\n        headers += [\n            (b\"set-cookie\", cookie.encode(\"utf-8\"))\n            for k, cookie in cookies.items()\n        ]\n\n        await self.transport.send(\n            {\n                \"type\": \"http.response.start\",\n                \"status\": response.status,\n                \"headers\": headers,\n            }\n        )\n\n        if isinstance(response, StreamingHTTPResponse):\n            response.protocol = self.transport.get_protocol()\n            await response.stream()\n            await response.protocol.complete()\n\n        else:\n            await self.transport.send(\n                {\n                    \"type\": \"http.response.body\",\n                    \"body\": response.body,\n                    \"more_body\": False,\n                }\n            )\n", "path": "sanic/asgi.py"}]}