1
import asyncio
2
import logging
3
 
4
import asgi
5
import js
6
from workers import Request, WorkerEntrypoint
7
 
8
from pyodide.ffi import to_js
9
 
10
 
11
def check_encoding(byte_str, encoding="utf-8"):
12
    try:
13
        byte_str.decode(encoding)
14
    except UnicodeDecodeError:
15
        return False
16
    return True
17
 
18
 
19
class Server:
20
    def __init__(self):
21
        pass
22
 
23
    async def __call__(self, scope, receive, send) -> None:
24
        scope["app"] = self
25
 
26
        assert scope["type"] in ("http", "websocket", "lifespan")
27
 
28
        if scope["type"] == "lifespan":
29
            message = await receive()
30
            if message["type"] == "lifespan.startup":
31
                await send({"type": "lifespan.startup.complete"})
32
            return
33
 
34
        elif scope["type"] == "http":
35
            headers = scope["headers"]
36
            for header in headers:
37
                assert isinstance(header[0], bytes) and isinstance(header[1], bytes)
38
                assert check_encoding(header[0]) and check_encoding(header[1])
39
 
40
            await receive()
41
            # Send response and return
42
            await send(
43
                {
44
                    "type": "http.response.start",
45
                    "status": 200,
46
                    "headers": headers,
47
                }
48
            )
49
 
50
            await send(
51
                {
52
                    "type": "http.response.body",
53
                    "body": b"Hello, World",
54
                }
55
            )
56
 
57
 
58
example_hdr = {"Header1": "Value1", "Header2": "Value2"}
59
 
60
 
61
class Default(WorkerEntrypoint):
62
    async def fetch(self, request):
63
        # Verify that `asgi` can handle JS-style headers and Python-style headers:
64
        js_request = js.Request.new("http://example.com/", headers=to_js(example_hdr))
65
        py_request = Request("http://example.com/", headers=example_hdr)
66
        js_scope = asgi.request_to_scope(js_request, self.env)
67
        py_scope = asgi.request_to_scope(py_request, self.env)
68
        assert (
69
            js_scope["headers"]
70
            == py_scope["headers"]
71
            == [(k.lower().encode(), v.encode()) for k, v in example_hdr.items()]
72
        )
73
 
74
        # Standard asgi.fetch test path.
75
        return await asgi.fetch(app, request, self.env, self.ctx)
76
 
77
    async def test(self, ctrl):
78
        await header_test(self.env)
79
        await test_error_after_response_is_logged(self.env)
80
        await test_background_task_error_is_logged()
81
        await test_app_exception_before_response_is_logged()
82
 
83
 
84
app = Server()
85
 
86
 
87
async def header_test(env):
88
    response = await env.SELF.fetch("http://example.com/", headers=to_js(example_hdr))
89
    for header in response.headers.items():
90
        assert isinstance(header[0], str) and isinstance(header[1], str)
91
        expected_hdr = {k.lower(): v.lower() for k, v in example_hdr.items()}
92
        assert header[0] in expected_hdr.keys()
93
        assert expected_hdr[header[0]] == header[1].lower()
94
 
95
 
96
# ---------------------------------------------------------------------------
97
# Logging tests
98
# ---------------------------------------------------------------------------
99
 
100
 
101
class _ListHandler(logging.Handler):
102
    """A logging handler that captures records into a list for assertions."""
103
 
104
    def __init__(self):
105
        super().__init__()
106
        self.records: list[logging.LogRecord] = []
107
 
108
    def emit(self, record):
109
        self.records.append(record)
110
 
111
    def clear(self):
112
        self.records.clear()
113
 
114
 
115
def _install_handler():
116
    """Install a ListHandler on the 'asgi' logger and return it."""
117
    handler = _ListHandler()
118
    logger = logging.getLogger("asgi")
119
    logger.addHandler(handler)
120
    # Ensure the logger level is low enough to capture everything.
121
    logger.setLevel(logging.DEBUG)
122
    return handler
123
 
124
 
125
def _remove_handler(handler):
126
    logging.getLogger("asgi").removeHandler(handler)
127
 
128
 
129
class _ErrorAfterResponseApp:
130
    """ASGI app that sends a valid response, then raises an exception."""
131
 
132
    async def __call__(self, scope, receive, send):
133
        if scope["type"] == "lifespan":
134
            message = await receive()
135
            if message["type"] == "lifespan.startup":
136
                await send({"type": "lifespan.startup.complete"})
137
            return
138
 
139
        if scope["type"] == "http":
140
            await receive()
141
            await send(
142
                {
143
                    "type": "http.response.start",
144
                    "status": 200,
145
                    "headers": [(b"content-type", b"text/plain")],
146
                }
147
            )
148
            await send(
149
                {
150
                    "type": "http.response.body",
151
                    "body": b"ok",
152
                }
153
            )
154
            # Response is already sent — now raise an error.
155
            raise RuntimeError("post-response error for testing")
156
 
157
 
158
async def test_error_after_response_is_logged(env):
159
    handler = _install_handler()
160
    try:
161
        req = js.Request.new("http://example.com/log-test")
162
        # The response should still succeed — the error happens after it's sent.
163
        response = await asgi.fetch(_ErrorAfterResponseApp(), req, env)
164
        assert response.status == 200, f"Expected 200, got {response.status}"
165
 
166
        # Let the event loop run
167
        await asyncio.sleep(20)
168
 
169
        # The error should have been logged, not swallowed.
170
        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
171
        assert len(error_records) > 0, (
172
            "Expected at least one ERROR log record for post-response exception, "
173
            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
174
        )
175
        matched = any(
176
            "post-response error for testing" in str(r.exc_info[1])
177
            if r.exc_info
178
            else False
179
            for r in error_records
180
        )
181
        assert matched, (
182
            "Expected log message containing 'post-response error for testing', "
183
            f"got: {[r.getMessage() for r in error_records]}"
184
        )
185
    finally:
186
        _remove_handler(handler)
187
 
188
 
189
async def test_background_task_error_is_logged():
190
    handler = _install_handler()
191
    try:
192
 
193
        async def failing_task():
194
            raise ValueError("background task failure for testing")
195
 
196
        asgi.run_in_background(failing_task())
197
 
198
        # Let the event loop run
199
        await asyncio.sleep(20)
200
 
201
        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
202
        assert len(error_records) > 0, (
203
            "Expected at least one ERROR log for background task failure, "
204
            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
205
        )
206
        matched = any(
207
            "background task failure for testing" in str(r.exc_info[1])
208
            if r.exc_info
209
            else False
210
            for r in error_records
211
        )
212
        assert matched, (
213
            "Expected log message containing 'background task failure for testing', "
214
            f"got: {[r.getMessage() for r in error_records]}"
215
        )
216
    finally:
217
        _remove_handler(handler)
218
 
219
 
220
class _ErrorBeforeResponseApp:
221
    """ASGI app that raises before sending any response."""
222
 
223
    async def __call__(self, scope, receive, send):
224
        if scope["type"] == "lifespan":
225
            message = await receive()
226
            if message["type"] == "lifespan.startup":
227
                await send({"type": "lifespan.startup.complete"})
228
            return
229
 
230
        if scope["type"] == "http":
231
            await receive()
232
            raise RuntimeError("app crash before response for testing")
233
 
234
 
235
async def test_app_exception_before_response_is_logged():
236
    handler = _install_handler()
237
    try:
238
        req = js.Request.new("http://example.com/crash-test")
239
        threw = False
240
        try:
241
            await asgi.fetch(_ErrorBeforeResponseApp(), req, {})
242
        except RuntimeError as e:
243
            threw = True
244
            assert "app crash before response for testing" in str(e), (
245
                f"Expected original exception message, got: {e}"
246
            )
247
 
248
        assert threw, "Expected RuntimeError to be raised from asgi.fetch"
249
 
250
        # fetch() should have logged the error before re-raising.
251
        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
252
        assert len(error_records) > 0, (
253
            "Expected at least one ERROR log for request failure, "
254
            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
255
        )
256
        matched = any("ASGI request failed" in r.getMessage() for r in error_records)
257
        assert matched, (
258
            "Expected log message containing 'ASGI request failed', "
259
            f"got: {[r.getMessage() for r in error_records]}"
260
        )
261
    finally:
262
        _remove_handler(handler)