Class ASGITest

tests/asgi/tests.py:51–805

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

    def setUp(self):
        request_started.disconnect(close_old_connections)
        self.addCleanup(request_started.connect, close_old_connections)

    async def test_get_asgi_application(self):
        """
        get_asgi_application() returns a functioning ASGI callable.
        """
        application = get_asgi_application()
        # Construct HTTP request.
        scope = self.async_request_factory._base_scope(path="/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        # Read the response.
        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!")
        # Allow response.close() to finish.
        await communicator.wait()

    async def test_asgi_cookies(self):
        application = get_asgi_application()
        scope = self.async_request_factory._base_scope(path="/cookie/")
        communicator = ApplicationCommunicator(application, scope)
        await communicator.send_input({"type": "http.request"})
        response_start = await communicator.receive_output()
        self.assertIn((b"Set-Cookie", b"key=value; Path=/"), response_start["headers"])
        # Allow response.close() to finish.
        await communicator.wait()

    # 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.
    @ignore_warnings(module="django.core.handlers.asgi")
    async def test_file_response(self):
        """
        Makes sure that FileResponse works over ASGI.
        """
        application = get_asgi_application()
        # Construct HTTP request.
        scope = self.async_request_factory._base_scope(path="/file/")
        communicator = ApplicationCommunicator(application, scope)