From a7829ec457edf589b2dc99af20633eecc96824bf Mon Sep 17 00:00:00 2001
From: Gyeongjae Choi <gyeongjae@cloudflare.com>
Date: Tue, 7 Jul 2026 16:29:48 +0900
Subject: [PATCH 1/4] fix: improve lifespan handling in asgi.py

---
 .../cli/tests/workerd-test/asgi/worker.py     | 228 ++++++++++++++++++
 packages/runtime-sdk/src/asgi.py              | 106 +++++---
 2 files changed, 307 insertions(+), 27 deletions(-)

diff --git a/packages/cli/tests/workerd-test/asgi/worker.py b/packages/cli/tests/workerd-test/asgi/worker.py
index 0532bfa..daeaa5f 100644
--- a/packages/cli/tests/workerd-test/asgi/worker.py
+++ b/packages/cli/tests/workerd-test/asgi/worker.py
@@ -204,6 +204,13 @@ async def test(self, ctrl):
         await test_error_after_response_is_logged(self.env)
         await test_background_task_error_is_logged()
         await test_app_exception_before_response_is_logged()
+        await test_lifespan_unsupported_app_does_not_hang(self.env)
+        await test_lifespan_full_cycle(self.env)
+        await test_lifespan_startup_failed_propagates(self.env)
+        await test_lifespan_shutdown_failed_propagates(self.env)
+        await test_null_body_statuses(self.env)
+        await test_response_start_without_headers(self.env)
+        await test_response_body_without_body_key(self.env)
 
 
 # ---------------------------------------------------------------------------
@@ -476,3 +483,224 @@ async def test_app_exception_before_response_is_logged():
         )
     finally:
         _remove_handler(handler)
+
+
+# ---------------------------------------------------------------------------
+# Test: lifespan protocol tolerance (app that rejects the lifespan scope)
+# ---------------------------------------------------------------------------
+
+
+class _NoLifespanApp:
+    """Mimics Django's ASGIHandler, which raises on any non-http scope."""
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            raise ValueError(f"can only handle http, not {scope['type']}")
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"served"})
+
+
+async def test_lifespan_unsupported_app_does_not_hang(env):
+    req = js.Request.new("http://example.com/no-lifespan")
+    # The previous adaptor blocked forever waiting for a startup ack the app
+    # never sends; wait_for turns that hang into an observable test failure.
+    response = await asyncio.wait_for(asgi.fetch(_NoLifespanApp(), req, env), timeout=5)
+    assert response.status == 200
+    assert await response.text() == "served"
+
+
+# ---------------------------------------------------------------------------
+# Test: null-body statuses must not carry a response body
+# ---------------------------------------------------------------------------
+
+
+class _NullBodyApp:
+    def __init__(self, status):
+        self._status = status
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": self._status,
+                "headers": [(b"x-null-body", b"1")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"body-should-be-dropped"})
+
+
+async def test_null_body_statuses(env):
+    for status in (204, 205, 304):
+        req = js.Request.new(f"http://example.com/null-body/{status}")
+        response = await asgi.fetch(_NullBodyApp(status), req, env)
+        assert response.status == status, f"expected {status}, got {response.status}"
+        assert response.headers["x-null-body"] == "1"
+        assert await response.text() == ""
+
+
+# ---------------------------------------------------------------------------
+# Test: defensive handling of ASGI messages missing optional keys
+# ---------------------------------------------------------------------------
+
+
+class _MissingHeadersApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send({"type": "http.response.start", "status": 200})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+async def test_response_start_without_headers(env):
+    req = js.Request.new("http://example.com/missing-headers")
+    response = await asgi.fetch(_MissingHeadersApp(), req, env)
+    assert response.status == 200
+    assert await response.text() == "ok"
+
+
+class _MissingBodyKeyApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body"})
+
+
+async def test_response_body_without_body_key(env):
+    req = js.Request.new("http://example.com/missing-body")
+    response = await asgi.fetch(_MissingBodyKeyApp(), req, env)
+    assert response.status == 200
+    assert await response.text() == ""
+
+
+# ---------------------------------------------------------------------------
+# Test: full lifespan cycle (startup AND shutdown are both delivered)
+# ---------------------------------------------------------------------------
+
+
+class _LifespanCycleApp:
+    def __init__(self):
+        self.events = []
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    self.events.append("startup")
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    self.events.append("shutdown")
+                    await send({"type": "lifespan.shutdown.complete"})
+                    return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+async def test_lifespan_full_cycle(env):
+    app = _LifespanCycleApp()
+    req = js.Request.new("http://example.com/lifespan-cycle")
+    response = await asgi.fetch(app, req, env)
+    assert response.status == 200
+    assert app.events == ["startup", "shutdown"]
+
+
+# ---------------------------------------------------------------------------
+# Test: lifespan startup/shutdown failures propagate out of asgi.fetch
+# ---------------------------------------------------------------------------
+
+
+class _StartupFailApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            await receive()
+            await send({"type": "lifespan.startup.failed", "message": "boom-startup"})
+            return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+async def test_lifespan_startup_failed_propagates(env):
+    req = js.Request.new("http://example.com/startup-fail")
+    raised = None
+    try:
+        await asyncio.wait_for(asgi.fetch(_StartupFailApp(), req, env), timeout=5)
+    except Exception as e:
+        raised = e
+    assert isinstance(raised, RuntimeError), f"expected RuntimeError, got {raised!r}"
+    assert "boom-startup" in str(raised)
+
+
+class _ShutdownFailApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    await send(
+                        {"type": "lifespan.shutdown.failed", "message": "boom-shutdown"}
+                    )
+                    return
+        if scope["type"] != "http":
+            return
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+async def test_lifespan_shutdown_failed_propagates(env):
+    req = js.Request.new("http://example.com/shutdown-fail")
+    raised = None
+    try:
+        await asyncio.wait_for(asgi.fetch(_ShutdownFailApp(), req, env), timeout=5)
+    except Exception as e:
+        raised = e
+    assert isinstance(raised, RuntimeError), f"expected RuntimeError, got {raised!r}"
+    assert "boom-shutdown" in str(raised)
diff --git a/packages/runtime-sdk/src/asgi.py b/packages/runtime-sdk/src/asgi.py
index 705615f..2b9d054 100644
--- a/packages/runtime-sdk/src/asgi.py
+++ b/packages/runtime-sdk/src/asgi.py
@@ -1,8 +1,7 @@
 import logging
-from asyncio import Event, Future, Queue, create_task, ensure_future, sleep
+from asyncio import Event, Future, Queue, create_task, ensure_future
 from collections.abc import Awaitable
 from contextlib import contextmanager
-from inspect import isawaitable
 from typing import Any
 
 import js
@@ -10,6 +9,7 @@
 from workers import Context, Request
 
 ASGI = {"spec_version": "2.0", "version": "3.0"}
+NULL_BODY_STATUSES = frozenset({101, 103, 204, 205, 304})
 logger = logging.getLogger("asgi")
 background_tasks = set()
 
@@ -76,42 +76,86 @@ def request_to_scope(req, env, ws=False):
 
 
 async def start_application(app):
-    shutdown_future = Future()
+    # Drives one ASGI lifespan startup/shutdown cycle before/after serving the
+    # request. The lifespan protocol is optional, so we must tolerate apps that
+    # don't implement it and fall back to serving requests without lifespan.
+    # https://asgi.readthedocs.io/en/latest/specs/lifespan.html
+    receive_queue = Queue()
+    await receive_queue.put({"type": "lifespan.startup"})
+
+    # `startup` resolves True on `lifespan.startup.complete`, False when the app
+    # has no lifespan support, and raises on `lifespan.startup.failed`.
+    # `shutdown_complete` mirrors the shutdown phase.
+    startup = Future()
+    shutdown_complete = Future()
 
     async def shutdown():
-        shutdown_future.set_result(None)
-        await sleep(0)
+        # Nothing to shut down if the app never completed a lifespan startup.
+        if startup.done() and not startup.result():
+            return
+        await receive_queue.put({"type": "lifespan.shutdown"})
+        await shutdown_complete
 
-    it = iter([{"type": "lifespan.startup"}, Future()])
+    async def no_lifespan_shutdown():
+        return
 
     async def receive():
-        res = next(it)
-        if isawaitable(res):
-            await res
-        return res
-
-    ready = Future()
+        return await receive_queue.get()
 
     async def send(got):
         if got["type"] == "lifespan.startup.complete":
-            ready.set_result(None)
+            if not startup.done():
+                startup.set_result(True)
+            return
+        if got["type"] == "lifespan.startup.failed":
+            message = got.get("message", "ASGI lifespan startup failed")
+            if not startup.done():
+                startup.set_exception(RuntimeError(message))
             return
         if got["type"] == "lifespan.shutdown.complete":
+            if not shutdown_complete.done():
+                shutdown_complete.set_result(None)
+            return
+        if got["type"] == "lifespan.shutdown.failed":
+            message = got.get("message", "ASGI lifespan shutdown failed")
+            if not shutdown_complete.done():
+                shutdown_complete.set_exception(RuntimeError(message))
             return
         raise RuntimeError(f"Unexpected lifespan event {got['type']}")
 
-    run_in_background(
-        app(
-            {
-                "asgi": ASGI,
-                "state": {},
-                "type": "lifespan",
-            },
-            receive,
-            send,
-        )
-    )
-    await ready
+    async def run_lifespan():
+        try:
+            await app(
+                {
+                    "asgi": ASGI,
+                    "state": {},
+                    "type": "lifespan",
+                },
+                receive,
+                send,
+            )
+            # App returned without acking: a missing startup ack means it has no
+            # lifespan handler, so mark startup unsupported rather than hanging.
+            if not startup.done():
+                startup.set_result(False)
+            elif not shutdown_complete.done():
+                shutdown_complete.set_result(None)
+        except Exception as exc:
+            # Spec: an exception raised before startup is acked signals that the
+            # app doesn't support lifespan; swallow it and serve requests anyway.
+            if not startup.done():
+                startup.set_result(False)
+                return
+            # After a successful startup, a shutdown-phase error can't affect the
+            # already-served request, so log it and let shutdown complete.
+            logger.exception("Exception in ASGI lifespan application", exc_info=exc)
+            if not shutdown_complete.done():
+                shutdown_complete.set_result(None)
+
+    run_in_background(run_lifespan())
+    supported = await startup
+    if not supported:
+        return no_lifespan_shutdown
     return shutdown
 
 
@@ -163,10 +207,10 @@ async def send(got):
         if got["type"] == "http.response.start":
             status = got["status"]
             # Like above, we need to convert byte-pairs into string explicitly.
-            headers = [(k.decode(), v.decode()) for k, v in got["headers"]]
+            headers = [(k.decode(), v.decode()) for k, v in got.get("headers", [])]
 
         elif got["type"] == "http.response.body":
-            body = got["body"]
+            body = got.get("body", b"")
             more_body = got.get("more_body", False)
 
             if writer is not None:
@@ -189,6 +233,14 @@ async def send(got):
                 result.set_result(resp)
                 with acquire_js_buffer(body) as jsbytes:
                     await writer.write(jsbytes.slice())
+            elif status in NULL_BODY_STATUSES:
+                # 101/103/204/205/304 must not carry a body per the Fetch spec.
+                # https://fetch.spec.whatwg.org/#null-body-status
+                resp = Response.new(
+                    None, headers=Object.fromEntries(headers), status=status
+                )
+                result.set_result(resp)
+                finished_response.set()
             else:
                 # Complete body in a single chunk
                 px = create_proxy(body)

From 61f68e8fb78e0f794171691c964553eb7dec4316 Mon Sep 17 00:00:00 2001
From: Gyeongjae Choi <gyeongjae@cloudflare.com>
Date: Wed, 8 Jul 2026 16:38:35 +0900
Subject: [PATCH 2/4] chore: use pytest

---
 .../tests/workerd-test/asgi/pyproject.toml    |   2 +-
 .../workerd-test/asgi/tests/test_asgi.py      | 384 ++++++++++++
 .../cli/tests/workerd-test/asgi/worker.py     | 547 +-----------------
 3 files changed, 409 insertions(+), 524 deletions(-)
 create mode 100644 packages/cli/tests/workerd-test/asgi/tests/test_asgi.py

diff --git a/packages/cli/tests/workerd-test/asgi/pyproject.toml b/packages/cli/tests/workerd-test/asgi/pyproject.toml
index c37f006..f7d3318 100644
--- a/packages/cli/tests/workerd-test/asgi/pyproject.toml
+++ b/packages/cli/tests/workerd-test/asgi/pyproject.toml
@@ -2,4 +2,4 @@
 name = "test"
 version = "0.0.0"
 requires-python = ">=3.12"
-dependencies = ["pytest"]
+dependencies = ["pytest", "pytest-asyncio<1.2.0"]
diff --git a/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py b/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py
new file mode 100644
index 0000000..7eac24f
--- /dev/null
+++ b/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py
@@ -0,0 +1,384 @@
+import asyncio
+import logging
+
+import asgi
+import js
+import pytest
+from pyodide.ffi import to_js
+from worker import STREAMING_CHUNK_SIZE, STREAMING_NUM_CHUNKS, example_hdr
+from workers import Request, env
+
+
+def test_request_to_scope_matches_js_and_py():
+    # Verify that `asgi` can handle JS-style headers and Python-style headers:
+    js_request = js.Request.new("http://example.com/", headers=to_js(example_hdr))
+    py_request = Request("http://example.com/", headers=example_hdr)
+    js_scope = asgi.request_to_scope(js_request, env)
+    py_scope = asgi.request_to_scope(py_request, env)
+    expected = [(k.lower().encode(), v.encode()) for k, v in example_hdr.items()]
+    assert js_scope["headers"] == py_scope["headers"] == expected
+
+
+@pytest.mark.asyncio
+async def test_headers():
+    response = await env.SELF.fetch("http://example.com/", headers=to_js(example_hdr))
+    expected_hdr = {k.lower(): v.lower() for k, v in example_hdr.items()}
+    for header in response.headers.items():
+        assert isinstance(header[0], str) and isinstance(header[1], str)
+        assert header[0] in expected_hdr
+        assert expected_hdr[header[0]] == header[1].lower()
+
+
+@pytest.mark.asyncio
+async def test_sse():
+    response = await env.SELF.fetch("http://example.com/sse")
+
+    assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
+    assert response.headers["cache-control"] == "no-store"
+
+    reader = response.body.getReader()
+    decoder = js.TextDecoder.new()
+    content = ""
+    while True:
+        result = await reader.read()
+        if result.done:
+            break
+        content += decoder.decode(result.value, {"stream": True})
+    # Final flush
+    content += decoder.decode()
+
+    # Verify the expected events are in the response
+    assert "event: endpoint" in content
+    assert "data: /messages/?session_id=test123" in content
+    assert ": ping - message 1" in content
+    assert ": ping - message 2" in content
+    assert ": ping - message 3" in content
+
+
+@pytest.mark.asyncio
+async def test_streaming():
+    response = await env.SELF.fetch("http://example.com/stream")
+
+    assert response.status == 200
+    assert response.headers["content-type"] == "application/octet-stream"
+
+    # Read the full response body via ReadableStream (same pattern as SSE test)
+    reader = response.body.getReader()
+    body_bytes = b""
+    while True:
+        result = await reader.read()
+        if result.done:
+            break
+        body_bytes += result.value.to_bytes()
+
+    expected_size = STREAMING_CHUNK_SIZE * STREAMING_NUM_CHUNKS  # 5120 bytes
+    # This is the key assertion: with bug #10, only the first chunk (1024 bytes)
+    # is returned. The fix should deliver all 5120 bytes.
+    assert len(body_bytes) == expected_size, (
+        f"Expected {expected_size} bytes, got {len(body_bytes)}."
+    )
+    # Verify each chunk has the correct content
+    for i in range(STREAMING_NUM_CHUNKS):
+        start = i * STREAMING_CHUNK_SIZE
+        chunk = body_bytes[start : start + STREAMING_CHUNK_SIZE]
+        assert all(b == i % 256 for b in chunk)
+
+
+class _ListHandler(logging.Handler):
+    """A logging handler that captures records into a list for assertions."""
+
+    def __init__(self):
+        super().__init__()
+        self.records: list[logging.LogRecord] = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+def _install_handler():
+    """Install a ListHandler on the 'asgi' logger and return it."""
+    handler = _ListHandler()
+    logger = logging.getLogger("asgi")
+    logger.addHandler(handler)
+    # Ensure the logger level is low enough to capture everything.
+    logger.setLevel(logging.DEBUG)
+    return handler
+
+
+def _remove_handler(handler):
+    logging.getLogger("asgi").removeHandler(handler)
+
+
+class _ErrorAfterResponseApp:
+    """ASGI app that sends a valid response, then raises an exception."""
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"ok"})
+        # Response is already sent — now raise an error.
+        raise RuntimeError("post-response error for testing")
+
+
+@pytest.mark.asyncio
+async def test_error_after_response_is_logged():
+    handler = _install_handler()
+    try:
+        req = js.Request.new("http://example.com/log-test")
+        # The response should still succeed — the error happens after it's sent.
+        response = await asgi.fetch(_ErrorAfterResponseApp(), req, env)
+        assert response.status == 200
+        # consume the body
+        await response.arrayBuffer()
+        # Let the event loop run
+        await asyncio.sleep(0.5)
+
+        # The error should have been logged, not swallowed.
+        errors = [r for r in handler.records if r.levelno >= logging.ERROR]
+        assert any(
+            r.exc_info and "post-response error for testing" in str(r.exc_info[1])
+            for r in errors
+        )
+    finally:
+        _remove_handler(handler)
+
+
+@pytest.mark.asyncio
+async def test_background_task_error_is_logged():
+    handler = _install_handler()
+    try:
+
+        async def failing_task():
+            raise ValueError("background task failure for testing")
+
+        asgi.run_in_background(failing_task())
+        # Let the event loop run
+        await asyncio.sleep(0.5)
+
+        errors = [r for r in handler.records if r.levelno >= logging.ERROR]
+        assert any(
+            r.exc_info and "background task failure for testing" in str(r.exc_info[1])
+            for r in errors
+        )
+    finally:
+        _remove_handler(handler)
+
+
+class _ErrorBeforeResponseApp:
+    """ASGI app that raises before sending any response."""
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        await receive()
+        raise RuntimeError("app crash before response for testing")
+
+
+@pytest.mark.asyncio
+async def test_app_exception_before_response_is_logged():
+    handler = _install_handler()
+    try:
+        req = js.Request.new("http://example.com/crash-test")
+        with pytest.raises(RuntimeError, match="app crash before response for testing"):
+            await asgi.fetch(_ErrorBeforeResponseApp(), req, {})
+
+        # fetch() should have logged the error before re-raising.
+        errors = [r for r in handler.records if r.levelno >= logging.ERROR]
+        assert any("ASGI request failed" in r.getMessage() for r in errors)
+    finally:
+        _remove_handler(handler)
+
+
+class _NoLifespanApp:
+    """Mimics Django's ASGIHandler, which raises on any non-http scope."""
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            raise ValueError(f"can only handle http, not {scope['type']}")
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"served"})
+
+
+@pytest.mark.asyncio
+async def test_lifespan_unsupported_app_does_not_hang():
+    req = js.Request.new("http://example.com/no-lifespan")
+    # The previous adaptor blocked forever waiting for a startup ack the app
+    # never sends; wait_for turns that hang into an observable test failure.
+    response = await asyncio.wait_for(asgi.fetch(_NoLifespanApp(), req, env), timeout=5)
+    assert response.status == 200
+    assert await response.text() == "served"
+
+
+class _NullBodyApp:
+    def __init__(self, status):
+        self._status = status
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": self._status,
+                "headers": [(b"x-null-body", b"1")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"body-should-be-dropped"})
+
+
+@pytest.mark.asyncio
+async def test_null_body_statuses():
+    for status in (204, 205, 304):
+        req = js.Request.new(f"http://example.com/null-body/{status}")
+        response = await asgi.fetch(_NullBodyApp(status), req, env)
+        assert response.status == status
+        assert response.headers["x-null-body"] == "1"
+        assert await response.text() == ""
+
+
+class _MissingHeadersApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        await receive()
+        await send({"type": "http.response.start", "status": 200})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+@pytest.mark.asyncio
+async def test_response_start_without_headers():
+    req = js.Request.new("http://example.com/missing-headers")
+    response = await asgi.fetch(_MissingHeadersApp(), req, env)
+    assert response.status == 200
+    assert await response.text() == "ok"
+
+
+class _MissingBodyKeyApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body"})
+
+
+@pytest.mark.asyncio
+async def test_response_body_without_body_key():
+    req = js.Request.new("http://example.com/missing-body")
+    response = await asgi.fetch(_MissingBodyKeyApp(), req, env)
+    assert response.status == 200
+    assert await response.text() == ""
+
+
+class _LifespanCycleApp:
+    def __init__(self):
+        self.events = []
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    self.events.append("startup")
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    self.events.append("shutdown")
+                    await send({"type": "lifespan.shutdown.complete"})
+                    return
+        await receive()
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/plain")],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+@pytest.mark.asyncio
+async def test_lifespan_full_cycle():
+    lifespan_app = _LifespanCycleApp()
+    req = js.Request.new("http://example.com/lifespan-cycle")
+    response = await asgi.fetch(lifespan_app, req, env)
+    assert response.status == 200
+    assert lifespan_app.events == ["startup", "shutdown"]
+
+
+class _StartupFailApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            await receive()
+            await send({"type": "lifespan.startup.failed", "message": "boom-startup"})
+            return
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+@pytest.mark.asyncio
+async def test_lifespan_startup_failed_propagates():
+    req = js.Request.new("http://example.com/startup-fail")
+    with pytest.raises(RuntimeError, match="boom-startup"):
+        await asyncio.wait_for(asgi.fetch(_StartupFailApp(), req, env), timeout=5)
+
+
+class _ShutdownFailApp:
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message["type"] == "lifespan.startup":
+                    await send({"type": "lifespan.startup.complete"})
+                elif message["type"] == "lifespan.shutdown":
+                    await send(
+                        {"type": "lifespan.shutdown.failed", "message": "boom-shutdown"}
+                    )
+                    return
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"ok"})
+
+
+@pytest.mark.asyncio
+async def test_lifespan_shutdown_failed_propagates():
+    req = js.Request.new("http://example.com/shutdown-fail")
+    with pytest.raises(RuntimeError, match="boom-shutdown"):
+        await asyncio.wait_for(asgi.fetch(_ShutdownFailApp(), req, env), timeout=5)
diff --git a/packages/cli/tests/workerd-test/asgi/worker.py b/packages/cli/tests/workerd-test/asgi/worker.py
index daeaa5f..7bd14fa 100644
--- a/packages/cli/tests/workerd-test/asgi/worker.py
+++ b/packages/cli/tests/workerd-test/asgi/worker.py
@@ -1,10 +1,26 @@
 import asyncio
-import logging
+import os
+import sys
 
 import asgi
-import js
-from pyodide.ffi import to_js
-from workers import Request, WorkerEntrypoint
+import pytest
+from pyodide.webloop import WebLoop
+from workers import WorkerEntrypoint
+
+
+async def noop(*args):
+    pass
+
+
+# pytest-asyncio relies on these but in Pyodide < 0.29 WebLoop does not implement them
+WebLoop.shutdown_asyncgens = noop
+WebLoop.shutdown_default_executor = noop
+
+# Pyodide 0.26.0a2's _cancel_all_tasks calls task.exception() on pending tasks,
+# which raises InvalidStateError under Pyodide's WebLoop.
+if sys.version_info < (3, 13):
+    asyncio.runners._cancel_all_tasks = lambda loop: None  # type: ignore[attr-defined]
+
 
 # ---------------------------------------------------------------------------
 # ASGI apps
@@ -183,524 +199,9 @@ async def fetch(self, request):
         elif path == "/stream":
             return await asgi.fetch(streaming_app, request, self.env, self.ctx)
 
-        # Verify that `asgi` can handle JS-style headers and Python-style headers:
-        js_request = js.Request.new("http://example.com/", headers=to_js(example_hdr))
-        py_request = Request("http://example.com/", headers=example_hdr)
-        js_scope = asgi.request_to_scope(js_request, self.env)
-        py_scope = asgi.request_to_scope(py_request, self.env)
-        assert (
-            js_scope["headers"]
-            == py_scope["headers"]
-            == [(k.lower().encode(), v.encode()) for k, v in example_hdr.items()]
-        )
-
-        # Standard asgi.fetch test path.
         return await asgi.fetch(app, request, self.env, self.ctx)
 
-    async def test(self, ctrl):
-        await test_headers(self.env)
-        await test_sse(self.env)
-        await test_streaming(self.env)
-        await test_error_after_response_is_logged(self.env)
-        await test_background_task_error_is_logged()
-        await test_app_exception_before_response_is_logged()
-        await test_lifespan_unsupported_app_does_not_hang(self.env)
-        await test_lifespan_full_cycle(self.env)
-        await test_lifespan_startup_failed_propagates(self.env)
-        await test_lifespan_shutdown_failed_propagates(self.env)
-        await test_null_body_statuses(self.env)
-        await test_response_start_without_headers(self.env)
-        await test_response_body_without_body_key(self.env)
-
-
-# ---------------------------------------------------------------------------
-# Test: headers
-# ---------------------------------------------------------------------------
-
-
-async def test_headers(env):
-    response = await env.SELF.fetch("http://example.com/", headers=to_js(example_hdr))
-    for header in response.headers.items():
-        assert isinstance(header[0], str) and isinstance(header[1], str)
-        expected_hdr = {k.lower(): v.lower() for k, v in example_hdr.items()}
-        assert header[0] in expected_hdr.keys()
-        assert expected_hdr[header[0]] == header[1].lower()
-
-
-# ---------------------------------------------------------------------------
-# Test: SSE (Server-Sent Events)
-# ---------------------------------------------------------------------------
-
-
-async def test_sse(env):
-    response = await env.SELF.fetch("http://example.com/sse")
-
-    assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
-    assert response.headers["cache-control"] == "no-store"
-
-    from js import TextDecoder
-
-    reader = response.body.getReader()
-    content = ""
-    decoder = TextDecoder.new()
-
-    while True:
-        result = await reader.read()
-        if result.done:
-            break
-        chunk_text = decoder.decode(result.value, {"stream": True})
-        content += chunk_text
-
-    # Final flush
-    content += decoder.decode()
-    # Verify the expected events are in the response
-    assert "event: endpoint" in content
-    assert "data: /messages/?session_id=test123" in content
-    assert ": ping - message 1" in content
-    assert ": ping - message 2" in content
-    assert ": ping - message 3" in content
-
-
-# ---------------------------------------------------------------------------
-# Test: non-SSE multi-chunk streaming
-# Reproduction test for bug https://github.com/cloudflare/workers-py/issues/67
-# ---------------------------------------------------------------------------
-
-
-async def test_streaming(env):
-    response = await env.SELF.fetch("http://example.com/stream")
-
-    assert response.status == 200
-    assert response.headers["content-type"] == "application/octet-stream"
-
-    # Read the full response body via ReadableStream (same pattern as SSE test)
-    reader = response.body.getReader()
-    body_bytes = b""
-
-    while True:
-        result = await reader.read()
-        if result.done:
-            break
-        body_bytes += result.value.to_bytes()
-
-    expected_size = STREAMING_CHUNK_SIZE * STREAMING_NUM_CHUNKS  # 5120 bytes
-
-    # This is the key assertion: with bug #10, only the first chunk (1024 bytes)
-    # is returned. The fix should deliver all 5120 bytes.
-    assert len(body_bytes) == expected_size, (
-        f"Expected {expected_size} bytes, got {len(body_bytes)}. "
-        f"Only {len(body_bytes) // STREAMING_CHUNK_SIZE} of {STREAMING_NUM_CHUNKS} chunks delivered."
-    )
-
-    # Verify each chunk has the correct content
-    for i in range(STREAMING_NUM_CHUNKS):
-        start = i * STREAMING_CHUNK_SIZE
-        end = start + STREAMING_CHUNK_SIZE
-        expected_byte = i % 256
-        chunk = body_bytes[start:end]
-        assert all(b == expected_byte for b in chunk), (
-            f"Chunk {i}: expected all bytes to be {expected_byte}, "
-            f"but got different content"
-        )
-
-
-# ---------------------------------------------------------------------------
-# Logging tests
-# ---------------------------------------------------------------------------
-
-
-class _ListHandler(logging.Handler):
-    """A logging handler that captures records into a list for assertions."""
-
-    def __init__(self):
-        super().__init__()
-        self.records: list[logging.LogRecord] = []
-
-    def emit(self, record):
-        self.records.append(record)
-
-    def clear(self):
-        self.records.clear()
-
-
-def _install_handler():
-    """Install a ListHandler on the 'asgi' logger and return it."""
-    handler = _ListHandler()
-    logger = logging.getLogger("asgi")
-    logger.addHandler(handler)
-    # Ensure the logger level is low enough to capture everything.
-    logger.setLevel(logging.DEBUG)
-    return handler
-
-
-def _remove_handler(handler):
-    logging.getLogger("asgi").removeHandler(handler)
-
-
-# -- Test: error after response already sent is logged (not swallowed) --------
-
-
-class _ErrorAfterResponseApp:
-    """ASGI app that sends a valid response, then raises an exception."""
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-            return
-
-        if scope["type"] == "http":
-            await receive()
-            await send(
-                {
-                    "type": "http.response.start",
-                    "status": 200,
-                    "headers": [(b"content-type", b"text/plain")],
-                }
-            )
-            await send(
-                {
-                    "type": "http.response.body",
-                    "body": b"ok",
-                }
-            )
-            # Response is already sent — now raise an error.
-            raise RuntimeError("post-response error for testing")
-
-
-async def test_error_after_response_is_logged(env):
-    handler = _install_handler()
-    try:
-        req = js.Request.new("http://example.com/log-test")
-        # The response should still succeed — the error happens after it's sent.
-        response = await asgi.fetch(_ErrorAfterResponseApp(), req, env)
-        assert response.status == 200, f"Expected 200, got {response.status}"
-
-        # consume the body
-        await response.arrayBuffer()
-
-        # Let the event loop run
-        await asyncio.sleep(0.5)
-
-        # The error should have been logged, not swallowed.
-        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
-        assert len(error_records) > 0, (
-            "Expected at least one ERROR log record for post-response exception, "
-            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
-        )
-        matched = any(
-            "post-response error for testing" in str(r.exc_info[1])
-            if r.exc_info
-            else False
-            for r in error_records
-        )
-        assert matched, (
-            "Expected log message containing 'post-response error for testing', "
-            f"got: {[r.getMessage() for r in error_records]}"
-        )
-    finally:
-        _remove_handler(handler)
-
-
-# -- Test: background task error is logged ------------------------------------
-
-
-async def test_background_task_error_is_logged():
-    handler = _install_handler()
-    try:
-
-        async def failing_task():
-            raise ValueError("background task failure for testing")
-
-        asgi.run_in_background(failing_task())
-
-        # Let the event loop run
-        await asyncio.sleep(0.5)
-
-        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
-        assert len(error_records) > 0, (
-            "Expected at least one ERROR log for background task failure, "
-            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
-        )
-        matched = any(
-            "background task failure for testing" in str(r.exc_info[1])
-            if r.exc_info
-            else False
-            for r in error_records
-        )
-        assert matched, (
-            "Expected log message containing 'background task failure for testing', "
-            f"got: {[r.getMessage() for r in error_records]}"
-        )
-    finally:
-        _remove_handler(handler)
-
-
-# -- Test: app exception before response is logged and re-raised -------------
-
-
-class _ErrorBeforeResponseApp:
-    """ASGI app that raises before sending any response."""
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-            return
-
-        if scope["type"] == "http":
-            await receive()
-            raise RuntimeError("app crash before response for testing")
-
-
-async def test_app_exception_before_response_is_logged():
-    handler = _install_handler()
-    try:
-        req = js.Request.new("http://example.com/crash-test")
-        threw = False
-        try:
-            await asgi.fetch(_ErrorBeforeResponseApp(), req, {})
-        except RuntimeError as e:
-            threw = True
-            assert "app crash before response for testing" in str(e), (
-                f"Expected original exception message, got: {e}"
-            )
-
-        assert threw, "Expected RuntimeError to be raised from asgi.fetch"
-
-        # fetch() should have logged the error before re-raising.
-        error_records = [r for r in handler.records if r.levelno >= logging.ERROR]
-        assert len(error_records) > 0, (
-            "Expected at least one ERROR log for request failure, "
-            f"got {len(error_records)}. All records: {[r.getMessage() for r in handler.records]}"
-        )
-        matched = any("ASGI request failed" in r.getMessage() for r in error_records)
-        assert matched, (
-            "Expected log message containing 'ASGI request failed', "
-            f"got: {[r.getMessage() for r in error_records]}"
-        )
-    finally:
-        _remove_handler(handler)
-
-
-# ---------------------------------------------------------------------------
-# Test: lifespan protocol tolerance (app that rejects the lifespan scope)
-# ---------------------------------------------------------------------------
-
-
-class _NoLifespanApp:
-    """Mimics Django's ASGIHandler, which raises on any non-http scope."""
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] != "http":
-            raise ValueError(f"can only handle http, not {scope['type']}")
-        await receive()
-        await send(
-            {
-                "type": "http.response.start",
-                "status": 200,
-                "headers": [(b"content-type", b"text/plain")],
-            }
-        )
-        await send({"type": "http.response.body", "body": b"served"})
-
-
-async def test_lifespan_unsupported_app_does_not_hang(env):
-    req = js.Request.new("http://example.com/no-lifespan")
-    # The previous adaptor blocked forever waiting for a startup ack the app
-    # never sends; wait_for turns that hang into an observable test failure.
-    response = await asyncio.wait_for(asgi.fetch(_NoLifespanApp(), req, env), timeout=5)
-    assert response.status == 200
-    assert await response.text() == "served"
-
-
-# ---------------------------------------------------------------------------
-# Test: null-body statuses must not carry a response body
-# ---------------------------------------------------------------------------
-
-
-class _NullBodyApp:
-    def __init__(self, status):
-        self._status = status
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-            return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send(
-            {
-                "type": "http.response.start",
-                "status": self._status,
-                "headers": [(b"x-null-body", b"1")],
-            }
-        )
-        await send({"type": "http.response.body", "body": b"body-should-be-dropped"})
-
-
-async def test_null_body_statuses(env):
-    for status in (204, 205, 304):
-        req = js.Request.new(f"http://example.com/null-body/{status}")
-        response = await asgi.fetch(_NullBodyApp(status), req, env)
-        assert response.status == status, f"expected {status}, got {response.status}"
-        assert response.headers["x-null-body"] == "1"
-        assert await response.text() == ""
-
-
-# ---------------------------------------------------------------------------
-# Test: defensive handling of ASGI messages missing optional keys
-# ---------------------------------------------------------------------------
-
-
-class _MissingHeadersApp:
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-            return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send({"type": "http.response.start", "status": 200})
-        await send({"type": "http.response.body", "body": b"ok"})
-
-
-async def test_response_start_without_headers(env):
-    req = js.Request.new("http://example.com/missing-headers")
-    response = await asgi.fetch(_MissingHeadersApp(), req, env)
-    assert response.status == 200
-    assert await response.text() == "ok"
-
-
-class _MissingBodyKeyApp:
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-            return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send(
-            {
-                "type": "http.response.start",
-                "status": 200,
-                "headers": [(b"content-type", b"text/plain")],
-            }
-        )
-        await send({"type": "http.response.body"})
-
-
-async def test_response_body_without_body_key(env):
-    req = js.Request.new("http://example.com/missing-body")
-    response = await asgi.fetch(_MissingBodyKeyApp(), req, env)
-    assert response.status == 200
-    assert await response.text() == ""
-
-
-# ---------------------------------------------------------------------------
-# Test: full lifespan cycle (startup AND shutdown are both delivered)
-# ---------------------------------------------------------------------------
-
-
-class _LifespanCycleApp:
-    def __init__(self):
-        self.events = []
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            while True:
-                message = await receive()
-                if message["type"] == "lifespan.startup":
-                    self.events.append("startup")
-                    await send({"type": "lifespan.startup.complete"})
-                elif message["type"] == "lifespan.shutdown":
-                    self.events.append("shutdown")
-                    await send({"type": "lifespan.shutdown.complete"})
-                    return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send(
-            {
-                "type": "http.response.start",
-                "status": 200,
-                "headers": [(b"content-type", b"text/plain")],
-            }
-        )
-        await send({"type": "http.response.body", "body": b"ok"})
-
-
-async def test_lifespan_full_cycle(env):
-    app = _LifespanCycleApp()
-    req = js.Request.new("http://example.com/lifespan-cycle")
-    response = await asgi.fetch(app, req, env)
-    assert response.status == 200
-    assert app.events == ["startup", "shutdown"]
-
-
-# ---------------------------------------------------------------------------
-# Test: lifespan startup/shutdown failures propagate out of asgi.fetch
-# ---------------------------------------------------------------------------
-
-
-class _StartupFailApp:
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            await receive()
-            await send({"type": "lifespan.startup.failed", "message": "boom-startup"})
-            return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send({"type": "http.response.start", "status": 200, "headers": []})
-        await send({"type": "http.response.body", "body": b"ok"})
-
-
-async def test_lifespan_startup_failed_propagates(env):
-    req = js.Request.new("http://example.com/startup-fail")
-    raised = None
-    try:
-        await asyncio.wait_for(asgi.fetch(_StartupFailApp(), req, env), timeout=5)
-    except Exception as e:
-        raised = e
-    assert isinstance(raised, RuntimeError), f"expected RuntimeError, got {raised!r}"
-    assert "boom-startup" in str(raised)
-
-
-class _ShutdownFailApp:
-    async def __call__(self, scope, receive, send):
-        if scope["type"] == "lifespan":
-            while True:
-                message = await receive()
-                if message["type"] == "lifespan.startup":
-                    await send({"type": "lifespan.startup.complete"})
-                elif message["type"] == "lifespan.shutdown":
-                    await send(
-                        {"type": "lifespan.shutdown.failed", "message": "boom-shutdown"}
-                    )
-                    return
-        if scope["type"] != "http":
-            return
-        await receive()
-        await send({"type": "http.response.start", "status": 200, "headers": []})
-        await send({"type": "http.response.body", "body": b"ok"})
-
-
-async def test_lifespan_shutdown_failed_propagates(env):
-    req = js.Request.new("http://example.com/shutdown-fail")
-    raised = None
-    try:
-        await asyncio.wait_for(asgi.fetch(_ShutdownFailApp(), req, env), timeout=5)
-    except Exception as e:
-        raised = e
-    assert isinstance(raised, RuntimeError), f"expected RuntimeError, got {raised!r}"
-    assert "boom-shutdown" in str(raised)
+    async def test(self):
+        os.chdir("/session/metadata/tests")
+        args = [".", "-vv"]
+        assert pytest.main(args) == 0

From 2c0e273ff8ed003a0b5df426736fb33382d5ff20 Mon Sep 17 00:00:00 2001
From: Gyeongjae Choi <gyeongjae@cloudflare.com>
Date: Wed, 8 Jul 2026 18:38:10 +0900
Subject: [PATCH 3/4] chore: exclude asgi from 3.12 + linux

---
 packages/cli/tests/test_in_workerd.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/cli/tests/test_in_workerd.py b/packages/cli/tests/test_in_workerd.py
index f47f5fa..c48b07c 100644
--- a/packages/cli/tests/test_in_workerd.py
+++ b/packages/cli/tests/test_in_workerd.py
@@ -61,7 +61,7 @@ def test_in_workerd(  # noqa: PLR0913  (too-many-arguments)
     # This is reproducible only in the unittest environment, and doesn't happen
     # when running the same worker manually.
     if (
-        test_dir.name in ("sdk", "entropy-patches")
+        test_dir.name in ("sdk", "entropy-patches", "asgi")
         and compat_date < "2025-09-29"
         and sys.platform == "linux"
     ):

From 9778365005c370a9c8f07d18da7ea7adc79c8a04 Mon Sep 17 00:00:00 2001
From: Gyeongjae Choi <gyeongjae@cloudflare.com>
Date: Thu, 9 Jul 2026 12:04:38 +0900
Subject: [PATCH 4/4] chore: address comments

---
 packages/cli/tests/workerd-test/asgi/tests/test_asgi.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py b/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py
index 7eac24f..0ec8c49 100644
--- a/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py
+++ b/packages/cli/tests/workerd-test/asgi/tests/test_asgi.py
@@ -72,11 +72,7 @@ async def test_streaming():
         body_bytes += result.value.to_bytes()
 
     expected_size = STREAMING_CHUNK_SIZE * STREAMING_NUM_CHUNKS  # 5120 bytes
-    # This is the key assertion: with bug #10, only the first chunk (1024 bytes)
-    # is returned. The fix should deliver all 5120 bytes.
-    assert len(body_bytes) == expected_size, (
-        f"Expected {expected_size} bytes, got {len(body_bytes)}."
-    )
+    assert len(body_bytes) == expected_size
     # Verify each chunk has the correct content
     for i in range(STREAMING_NUM_CHUNKS):
         start = i * STREAMING_CHUNK_SIZE