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)