getasgiapplication returns a functioning asgi callable construct http request read the response allow response close to finish python s file api is not async compatible a thirdparty library such as https github comtincheaiofiles allows passing the file to fileresponse as an async iterator with a sync iterator streaminghttpresponse triggers a warning when iterating the file assertwarnsmessage is not async compatible so ignorewarnings for the test makes sure that fileresponse works over asgi construct http request get the file content read the response windows registry may not be configured with correct mimetypes warning ignored here allow response close to finish construct http request get the file content read the response allow response close to finish allow response close to finish allow response close to finish allow response close to finish force both the disconnect listener and the task that sends the response to finish at the same time fetch response headers this yields to asyncio and causes asghandler sendresponse to dump the body of the response in the queue fetch response body there s already some data queued up so this doesn t actually yield to the event loop it just succeeds instantly send disconnect at the same time that response finishes this just puts some info in a queue it doesn t have to yield to the event loop waiting for the communicator does yield to the event loop since asgihandler sendresponse is still waiting to do response close it so happens that there are enough remaining yield points in both tasks that they both finish while the loop is running track threads handler is dispatched on threads def callself kwargs self threads appendthreading currentthread signalhandler signalhandler requeststarted connectsignalhandler requestfinished connectsignalhandler perform a basic request application getasgiapplication scope self asyncrequestfactory basescopepath communicator applicationcommunicatorapplication scope await communicator sendinputtype http request responsestart await communicator receiveoutput self assertequalresponsestarttype http response start self assertequalresponsestartstatus 200 responsebody await communicator receiveoutput self assertequalresponsebodytype http response body self assertequalresponsebodybody bhello world give response close time to finish await communicator wait asynctosync should have executed the signals in the same thread requeststartedthread requestfinishedthread signalhandler threads self assertequalrequeststartedthread requestfinishedthread requeststarted disconnectsignalhandler requestfinished disconnectsignalhandler async def testconcurrentasyncusesmultiplethreadpoolsself syncwaiter activethreads clear send 2 requests concurrently application getasgiapplication scope self asyncrequestfactory basescopepathwait communicators for in range2 communicators appendapplicationcommunicatorapplication scope await communicators1 sendinputtype http request each request must complete with a status code of 200 if requests aren t scheduled concurrently the barrier in the syncwait view will time out resulting in a 500 status code for communicator in communicators responsestart await communicator receiveoutput self assertequalresponsestarttype http response start self assertequalresponsestartstatus 200 responsebody await communicator receiveoutput self assertequalresponsebodytype http response body self assertequalresponsebodybody bhello world give response close time to finish await communicator wait the requests should have scheduled on different threads note activethreads is a set a thread can only appear once therefore length is a sufficient check self assertequallensyncwaiter activethreads 2 syncwaiter activethreads clear async def testasynciocancelerrorself flag to check if the view was cancelled viewdidcancel false a view that will listen for the cancelled error async def viewrequest nonlocal viewdidcancel try await asyncio sleep0 2 return httpresponsehello world except asyncio cancellederror set the flag viewdidcancel true raise request class to use the view class testasgirequestasgirequest urlconf pathcancel view handler to use request class class testasgihandlerasgihandler requestclass testasgirequest request cycle should complete since no disconnect was sent application testasgihandler scope self asyncrequestfactory basescopepathcancel communicator applicationcommunicatorapplication scope await communicator sendinputtype http request responsestart await communicator receiveoutput self assertequalresponsestarttype http response start self assertequalresponsestartstatus 200 responsebody await communicator receiveoutput self assertequalresponsebodytype http response body self assertequalresponsebodybody bhello world give response close time to finish await communicator wait self assertisviewdidcancel false request cycle with a disconnect before the view can respond application testasgihandler scope self asyncrequestfactory basescopepathcancel communicator applicationcommunicatorapplication scope await communicator sendinputtype http request let the view actually start await asyncio sleep0 1 disconnect the client await communicator sendinputtype http disconnect the handler should not send a response with self assertraisesasyncio timeouterror await communicator receiveoutput await communicator wait self assertisviewdidcancel true async def testasynciostreamingcancelerrorself similar to testasynciocancelerror but during a streaming response viewdidcancel false async def streamingresponse nonlocal viewdidcancel try await asyncio sleep0 2 yield bhello world except asyncio cancellederror set the flag viewdidcancel true raise async def viewrequest return streaminghttpresponsestreamingresponse class testasgirequestasgirequest urlconf pathcancel view class testasgihandlerasgihandler requestclass testasgirequest with no disconnect the request cycle should complete in the same manner as the nonstreaming response application testasgihandler scope self asyncrequestfactory basescopepathcancel communicator applicationcommunicatorapplication scope await communicator sendinputtype http request responsestart await communicator receiveoutput self assertequalresponsestarttype http response start self assertequalresponsestartstatus 200 responsebody await communicator receiveoutput self assertequalresponsebodytype http response body self assertequalresponsebodybody bhello world await communicator wait self assertisviewdidcancel false request cycle with a disconnect application testasgihandler scope self asyncrequestfactory basescopepathcancel communicator applicationcommunicatorapplication scope await communicator sendinputtype http request responsestart await communicator receiveoutput fetch the start of response so streaming can begin self assertequalresponsestarttype http response start self assertequalresponsestartstatus 200 await asyncio sleep0 1 now disconnect the client await communicator sendinputtype http disconnect this time the handler should not send a response with self assertraisesasyncio timeouterror await communicator receiveoutput await communicator wait self assertisviewdidcancel true async def teststreamingself scope self asyncrequestfactory basescope pathstreaming querystringbsleep0 001 application getasgiapplication communicator applicationcommunicatorapplication scope await communicator sendinputtype http request fetch http response start await communicator receiveoutputtimeout1 fetch the first and last firstresponse await communicator receiveoutputtimeout1 self assertequalfirstresponsebody bfirstn secondresponse await communicator receiveoutputtimeout1 self assertequalsecondresponsebody blastn fetch the rest of the response so that coroutines are cleaned up await communicator receiveoutputtimeout1 with self assertraisesasyncio timeouterror await communicator receiveoutputtimeout1 async def teststreamingdisconnectself scope self asyncrequestfactory basescope pathstreaming querystringbsleep0 1 application getasgiapplication communicator applicationcommunicatorapplication scope await communicator sendinputtype http request await communicator receiveoutputtimeout1 firstresponse await communicator receiveoutputtimeout1 self assertequalfirstresponsebody bfirstn disconnect the client await communicator sendinputtype http disconnect lastn isn t sent with self assertraisesasyncio timeouterror await communicator receiveoutputtimeout0 2 get_asgi_application returns a functioning asgi callable construct http request read the response allow response close to finish python s file api is not async compatible a third party library such as https github com tinche aiofiles allows passing the file to fileresponse as an async iterator with a sync iterator streaminghttpresponse triggers a warning when iterating the file assertwarnsmessage is not async compatible so ignore_warnings for the test makes sure that fileresponse works over asgi construct http request get the file content read the response windows registry may not be configured with correct mimetypes warning ignored here allow response close to finish construct http request get the file content read the response allow response close to finish allow response close to finish allow response close to finish allow response close to finish force both the disconnect listener and the task that sends the response to finish at the same time fetch response headers this yields to asyncio and causes asghandler send_response to dump the body of the response in the queue fetch response body there s already some data queued up so this doesn t actually yield to the event loop it just succeeds instantly send disconnect at the same time that response finishes this just puts some info in a queue it doesn t have to yield to the event loop waiting for the communicator _does_ yield to the event loop since asgihandler send_response is still waiting to do response close it so happens that there are enough remaining yield points in both tasks that they both finish while the loop is running track threads handler is dispatched on perform a basic request give response close time to finish asynctosync should have executed the signals in the same thread send 2 requests concurrently each request must complete with a status code of 200 if requests aren t scheduled concurrently the barrier in the sync_wait view will time out resulting in a 500 status code give response close time to finish the requests should have scheduled on different threads note active_threads is a set a thread can only appear once therefore length is a sufficient check flag to check if the view was cancelled a view that will listen for the cancelled error set the flag request class to use the view handler to use request class request cycle should complete since no disconnect was sent give response close time to finish request cycle with a disconnect before the view can respond let the view actually start disconnect the client the handler should not send a response similar to test_asyncio_cancel_error but during a streaming response set the flag with no disconnect the request cycle should complete in the same manner as the non streaming response request cycle with a disconnect fetch the start of response so streaming can begin now disconnect the client this time the handler should not send a response fetch http response start fetch the first and last fetch the rest of the response so that coroutines are cleaned up disconnect the client last n isn t sent
import asyncio
import sys
import threading
from pathlib import Path

from asgiref.testing import ApplicationCommunicator

from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
from django.core.asgi import get_asgi_application
from django.core.handlers.asgi import ASGIHandler, ASGIRequest
from django.core.signals import request_finished, request_started
from django.db import close_old_connections
from django.http import HttpResponse, StreamingHttpResponse
from django.test import (
    AsyncRequestFactory,
    SimpleTestCase,
    ignore_warnings,
    modify_settings,
    override_settings,
)
from django.urls import path
from django.utils.http import http_date

from .urls import sync_waiter, test_filename

TEST_STATIC_ROOT = Path(__file__).parent / "project" / "static"


@override_settings(ROOT_URLCONF="asgi.urls")
class ASGITest(SimpleTestCase):
    async_request_factory = AsyncRequestFactory()

    def setUp(self):
        request_started.disconnect(close_old_connections)

    def tearDown(self):
        request_started.connect(close_old_connections)

    async def test_get_asgi_application(self):
        
        application = get_asgi_application()
        
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        self.assertEqual(
            set(response_start["headers"]),
            {
                (b"Content-Length", b"12"),
                (b"Content-Type", b"text/html; charset=utf-8"),
            },
        )
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"Hello World!")
        
        await communicator.wait()

    
    
    
    
    
    
    @ignore_warnings(module="django.http.response")
    async def test_file_response(self):
        
        application = get_asgi_application()
        
        scope = self.async_request_factory._base_scope(path="/file/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        
        with open(test_filename, "rb") as test_file:
            test_file_contents = test_file.read()
        
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        headers = response_start["headers"]
        self.assertEqual(len(headers), 3)
        expected_headers = {
            b"Content-Length": str(len(test_file_contents)).encode("ascii"),
            b"Content-Type": b"text/x-python",
            b"Content-Disposition": b'inline; filename="urls.py"',
        }
        for key, value in headers:
            try:
                self.assertEqual(value, expected_headers[key])
            except AssertionError:
                
                
                if sys.platform == "win32" and key == b"Content-Type":
                    self.assertEqual(value, b"text/plain")
                else:
                    raise

        
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], test_file_contents)
        
        await communicator.wait()

    @modify_settings(INSTALLED_APPS={"append": "django.contrib.staticfiles"})
    @override_settings(
        STATIC_URL="static/",
        STATIC_ROOT=TEST_STATIC_ROOT,
        STATICFILES_DIRS=[TEST_STATIC_ROOT],
        STATICFILES_FINDERS=[
            "django.contrib.staticfiles.finders.FileSystemFinder",
        ],
    )
    async def test_static_file_response(self):
        application = ASGIStaticFilesHandler(get_asgi_application())
        
        scope = self.async_request_factory._base_scope(path="/static/file.txt")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        
        file_path = TEST_STATIC_ROOT / "file.txt"
        with open(file_path, "rb") as test_file:
            test_file_contents = test_file.read()
        
        stat = file_path.stat()
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        self.assertEqual(
            set(response_start["headers"]),
            {
                (b"Content-Length", str(len(test_file_contents)).encode("ascii")),
                (b"Content-Type", b"text/plain"),
                (b"Content-Disposition", b'inline; filename="file.txt"'),
                (b"Last-Modified", http_date(stat.st_mtime).encode("ascii")),
            },
        )
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], test_file_contents)
        
        await communicator.wait()

    async def test_headers(self):
        application = get_asgi_application()
        communicator = ApplicationCommunicator(
            application,
            self.async_request_factory._base_scope(
                path="/meta/",
                headers=[
                    [b"content-type", b"text/plain; charset=utf-8"],
                    [b"content-length", b"77"],
                    [b"referer", b"Scotland"],
                    [b"referer", b"Wales"],
                ],
            ),
        )
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        self.assertEqual(
            set(response_start["headers"]),
            {
                (b"Content-Length", b"19"),
                (b"Content-Type", b"text/plain; charset=utf-8"),
            },
        )
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"From Scotland,Wales")
        
        await communicator.wait()

    async def test_post_body(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(
            method="POST",
            path="/post/",
            query_string="echo=1",
        )
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request", "body": b"Echo!"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"Echo!")

    async def test_untouched_request_body_gets_closed(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(method="POST", path="/post/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 204)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"")
        
        await communicator.wait()

    async def test_get_query_string(self):
        application = get_asgi_application()
        for query_string in (b"name=Andrew", "name=Andrew"):
            with self.subTest(query_string=query_string):
                scope = self.async_request_factory._base_scope(
                    path="/",
                    query_string=query_string,
                )
                communicator = ApplicationCommunicator(application, scope)
                await communicator.send_input({"type": "http.request"})
                response_start = await communicator.receive_output()
                self.assertEqual(response_start["type"], "http.response.start")
                self.assertEqual(response_start["status"], 200)
                response_body = await communicator.receive_output()
                self.assertEqual(response_body["type"], "http.response.body")
                self.assertEqual(response_body["body"], b"Hello Andrew!")
                
                await communicator.wait()

    async def test_disconnect(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.disconnect"})
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output()

    async def test_disconnect_both_return(self):
        
        
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request", "body": b"some body"})
        
        
        
        await communicator.receive_output()
        
        
        
        await communicator.receive_output()
        
        
        
        await communicator.send_input({"type": "http.disconnect"})
        
        
        
        
        await communicator.wait()

    async def test_disconnect_with_body(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request", "body": b"some body"})
        await communicator.send_input({"type": "http.disconnect"})
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output()

    async def test_assert_in_listen_for_disconnect(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        await communicator.send_input({"type": "http.not_a_real_message"})
        msg = "Invalid ASGI message after request body: http.not_a_real_message"
        with self.assertRaisesMessage(AssertionError, msg):
            await communicator.wait()

    async def test_delayed_disconnect_with_body(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/delayed_hello/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request", "body": b"some body"})
        await communicator.send_input({"type": "http.disconnect"})
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output()

    async def test_wrong_connection_type(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/", type="other")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        msg = "Django can only handle ASGI/HTTP connections, not other."
        with self.assertRaisesMessage(ValueError, msg):
            await communicator.receive_output()

    async def test_non_unicode_query_string(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/", query_string=b"\xff")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 400)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"")

    async def test_request_lifecycle_signals_dispatched_with_thread_sensitive(self):
        class SignalHandler:
            

            threads = []

            def __call__(self, **kwargs):
                self.threads.append(threading.current_thread())

        signal_handler = SignalHandler()
        request_started.connect(signal_handler)
        request_finished.connect(signal_handler)

        
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"Hello World!")
        
        await communicator.wait()

        
        request_started_thread, request_finished_thread = signal_handler.threads
        self.assertEqual(request_started_thread, request_finished_thread)
        request_started.disconnect(signal_handler)
        request_finished.disconnect(signal_handler)

    async def test_concurrent_async_uses_multiple_thread_pools(self):
        sync_waiter.active_threads.clear()

        
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/wait/")
        communicators = []
        for _ in range(2):
            communicators.append(ApplicationCommunicator(application, scope))
            await communicators[-1].send_input({"type": "http.request"})

        
        
        
        for communicator in communicators:
            response_start = await communicator.receive_output()
            self.assertEqual(response_start["type"], "http.response.start")
            self.assertEqual(response_start["status"], 200)
            response_body = await communicator.receive_output()
            self.assertEqual(response_body["type"], "http.response.body")
            self.assertEqual(response_body["body"], b"Hello World!")
            
            await communicator.wait()

        
        
        
        self.assertEqual(len(sync_waiter.active_threads), 2)

        sync_waiter.active_threads.clear()

    async def test_asyncio_cancel_error(self):
        
        view_did_cancel = False

        
        async def view(request):
            nonlocal view_did_cancel
            try:
                await asyncio.sleep(0.2)
                return HttpResponse("Hello World!")
            except asyncio.CancelledError:
                
                view_did_cancel = True
                raise

        
        class TestASGIRequest(ASGIRequest):
            urlconf = (path("cancel/", view),)

        
        class TestASGIHandler(ASGIHandler):
            request_class = TestASGIRequest

        
        application = TestASGIHandler()
        scope = self.async_request_factory._base_scope(path="/cancel/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"Hello World!")
        
        await communicator.wait()
        self.assertIs(view_did_cancel, False)

        
        application = TestASGIHandler()
        scope = self.async_request_factory._base_scope(path="/cancel/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        
        await asyncio.sleep(0.1)
        
        await communicator.send_input({"type": "http.disconnect"})
        
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output()
        await communicator.wait()
        self.assertIs(view_did_cancel, True)

    async def test_asyncio_streaming_cancel_error(self):
        
        
        view_did_cancel = False

        async def streaming_response():
            nonlocal view_did_cancel
            try:
                await asyncio.sleep(0.2)
                yield b"Hello World!"
            except asyncio.CancelledError:
                
                view_did_cancel = True
                raise

        async def view(request):
            return StreamingHttpResponse(streaming_response())

        class TestASGIRequest(ASGIRequest):
            urlconf = (path("cancel/", view),)

        class TestASGIHandler(ASGIHandler):
            request_class = TestASGIRequest

        
        
        application = TestASGIHandler()
        scope = self.async_request_factory._base_scope(path="/cancel/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        response_body = await communicator.receive_output()
        self.assertEqual(response_body["type"], "http.response.body")
        self.assertEqual(response_body["body"], b"Hello World!")
        await communicator.wait()
        self.assertIs(view_did_cancel, False)

        
        application = TestASGIHandler()
        scope = self.async_request_factory._base_scope(path="/cancel/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        
        self.assertEqual(response_start["type"], "http.response.start")
        self.assertEqual(response_start["status"], 200)
        await asyncio.sleep(0.1)
        
        await communicator.send_input({"type": "http.disconnect"})
        
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output()
        await communicator.wait()
        self.assertIs(view_did_cancel, True)

    async def test_streaming(self):
        scope = self.async_request_factory._base_scope(
            path="/streaming/", query_string=b"sleep=0.001"
        )
        application = get_asgi_application()
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        
        await communicator.receive_output(timeout=1)
        
        first_response = await communicator.receive_output(timeout=1)
        self.assertEqual(first_response["body"], b"first\n")
        second_response = await communicator.receive_output(timeout=1)
        self.assertEqual(second_response["body"], b"last\n")
        
        await communicator.receive_output(timeout=1)
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output(timeout=1)

    async def test_streaming_disconnect(self):
        scope = self.async_request_factory._base_scope(
            path="/streaming/", query_string=b"sleep=0.1"
        )
        application = get_asgi_application()
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        await communicator.receive_output(timeout=1)
        first_response = await communicator.receive_output(timeout=1)
        self.assertEqual(first_response["body"], b"first\n")
        
        await communicator.send_input({"type": "http.disconnect"})
        
        with self.assertRaises(asyncio.TimeoutError):
            await communicator.receive_output(timeout=0.2)