text: import asyncio
  import sys
  import threading
  from pathlib import Path
  from unittest import skipIf
  
  from asgiref.sync import SyncToAsync
  from asgiref.testing import ApplicationCommunicator
  
  from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
  from django.core.asgi import get_asgi_application
  from django.core.signals import request_finished, request_started
  from django.db import close_old_connections
  from django.test import (
      AsyncRequestFactory, SimpleTestCase, modify_settings, override_settings,
  )
  from django.utils.http import http_date
  
  from .urls import test_filename
  
  TEST_STATIC_ROOT = Path(__file__).parent / 'project' / 'static'
  
  
  @skipIf(sys.platform == 'win32' and (3, 8, 0) < sys.version_info < (3, 8, 1), 'https://bugs.python.org/issue38563')
  @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):
          """
          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!')
  
      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)
          await communicator.send_input({'type': 'http.request'})
          # Get the file content.
          with open(test_filename, 'rb') as test_file:
              test_file_contents = test_file.read()
          # 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', str(len(test_file_contents)).encode('ascii')),
                  (b'Content-Type', b'text/plain' if sys.platform == 'win32' else b'text/x-python'),
                  (b'Content-Disposition', b'inline; filename="urls.py"'),
              },
          )
          response_body = await communicator.receive_output()
          self.assertEqual(response_body['type'], 'http.response.body')
          self.assertEqual(response_body['body'], test_file_contents)
          # Allow response.close() to finish.
          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())
          # Construct HTTP request.
          scope = self.async_request_factory._base_scope(path='/static/file.txt')
          communicator = ApplicationCommunicator(application, scope)
          await communicator.send_input({'type': 'http.request'})
          # Get the file content.
          file_path = TEST_STATIC_ROOT / 'file.txt'
          with open(file_path, 'rb') as test_file:
              test_file_contents = test_file.read()
          # Read the response.
          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)
          # Allow response.close() to finish.
          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')
  
      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!')
  
      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_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:
              """Track threads handler is dispatched on."""
              threads = []
  
              def __call__(self, **kwargs):
                  self.threads.append(threading.current_thread())
  
          signal_handler = SignalHandler()
          request_started.connect(signal_handler)
          request_finished.connect(signal_handler)
  
          # Perform a basic request.
          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!')
          # Give response.close() time to finish.
          await communicator.wait()
  
          # At this point, AsyncToSync does not have a current executor. Thus
          # SyncToAsync falls-back to .single_thread_executor.
          target_thread = next(iter(SyncToAsync.single_thread_executor._threads))
          request_started_thread, request_finished_thread = signal_handler.threads
          self.assertEqual(request_started_thread, target_thread)
          self.assertEqual(request_finished_thread, target_thread)
          request_started.disconnect(signal_handler)
          request_finished.disconnect(signal_handler)
  