diff options
Diffstat (limited to 'tests/asgi')
| -rw-r--r-- | tests/asgi/tests.py | 177 | ||||
| -rw-r--r-- | tests/asgi/urls.py | 16 |
2 files changed, 100 insertions, 93 deletions
diff --git a/tests/asgi/tests.py b/tests/asgi/tests.py index 7eb35724df..f8d284e9a3 100644 --- a/tests/asgi/tests.py +++ b/tests/asgi/tests.py @@ -11,17 +11,23 @@ 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, + AsyncRequestFactory, + SimpleTestCase, + modify_settings, + override_settings, ) from django.utils.http import http_date from .urls import sync_waiter, test_filename -TEST_STATIC_ROOT = Path(__file__).parent / 'project' / 'static' +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') +@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() @@ -37,23 +43,23 @@ class ASGITest(SimpleTestCase): """ application = get_asgi_application() # Construct HTTP request. - scope = self.async_request_factory._base_scope(path='/') + 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.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(response_start["type"], "http.response.start") + self.assertEqual(response_start["status"], 200) self.assertEqual( - set(response_start['headers']), + set(response_start["headers"]), { - (b'Content-Length', b'12'), - (b'Content-Type', b'text/html; charset=utf-8'), + (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!') + self.assertEqual(response_body["type"], "http.response.body") + self.assertEqual(response_body["body"], b"Hello World!") async def test_file_response(self): """ @@ -61,22 +67,22 @@ class ASGITest(SimpleTestCase): """ application = get_asgi_application() # Construct HTTP request. - scope = self.async_request_factory._base_scope(path='/file/') + scope = self.async_request_factory._base_scope(path="/file/") communicator = ApplicationCommunicator(application, scope) - await communicator.send_input({'type': 'http.request'}) + await communicator.send_input({"type": "http.request"}) # Get the file content. - with open(test_filename, 'rb') as test_file: + 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) - headers = response_start['headers'] + 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"', + 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: @@ -84,52 +90,52 @@ class ASGITest(SimpleTestCase): except AssertionError: # Windows registry may not be configured with correct # mimetypes. - if sys.platform == 'win32' and key == b'Content-Type': - self.assertEqual(value, b'text/plain') + 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) + 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'}) + @modify_settings(INSTALLED_APPS={"append": "django.contrib.staticfiles"}) @override_settings( - STATIC_URL='static/', + STATIC_URL="static/", STATIC_ROOT=TEST_STATIC_ROOT, STATICFILES_DIRS=[TEST_STATIC_ROOT], STATICFILES_FINDERS=[ - 'django.contrib.staticfiles.finders.FileSystemFinder', + "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') + scope = self.async_request_factory._base_scope(path="/static/file.txt") communicator = ApplicationCommunicator(application, scope) - await communicator.send_input({'type': 'http.request'}) + 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: + 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(response_start["type"], "http.response.start") + self.assertEqual(response_start["status"], 200) self.assertEqual( - set(response_start['headers']), + 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')), + (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) + self.assertEqual(response_body["type"], "http.response.body") + self.assertEqual(response_body["body"], test_file_contents) # Allow response.close() to finish. await communicator.wait() @@ -138,79 +144,80 @@ class ASGITest(SimpleTestCase): communicator = ApplicationCommunicator( application, self.async_request_factory._base_scope( - path='/meta/', + 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'], + [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'}) + 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(response_start["type"], "http.response.start") + self.assertEqual(response_start["status"], 200) self.assertEqual( - set(response_start['headers']), + set(response_start["headers"]), { - (b'Content-Length', b'19'), - (b'Content-Type', b'text/plain; charset=utf-8'), + (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') + 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'): + for query_string in (b"name=Andrew", "name=Andrew"): with self.subTest(query_string=query_string): scope = self.async_request_factory._base_scope( - path='/', + path="/", query_string=query_string, ) communicator = ApplicationCommunicator(application, scope) - await communicator.send_input({'type': 'http.request'}) + 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(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!') + 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='/') + scope = self.async_request_factory._base_scope(path="/") communicator = ApplicationCommunicator(application, scope) - await communicator.send_input({'type': 'http.disconnect'}) + 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') + 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.' + 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') + scope = self.async_request_factory._base_scope(path="/", query_string=b"\xff") communicator = ApplicationCommunicator(application, scope) - await communicator.send_input({'type': 'http.request'}) + 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) + 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'') + 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): @@ -222,15 +229,15 @@ class ASGITest(SimpleTestCase): # Perform a basic request. application = get_asgi_application() - scope = self.async_request_factory._base_scope(path='/') + 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.request"}) response_start = await communicator.receive_output() - self.assertEqual(response_start['type'], 'http.response.start') - self.assertEqual(response_start['status'], 200) + 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!') + 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() @@ -245,22 +252,22 @@ class ASGITest(SimpleTestCase): # Send 2 requests concurrently application = get_asgi_application() - scope = self.async_request_factory._base_scope(path='/wait/') + 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'}) + await communicators[-1].send_input({"type": "http.request"}) # 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. for communicator in communicators: response_start = await communicator.receive_output() - self.assertEqual(response_start['type'], 'http.response.start') - self.assertEqual(response_start['status'], 200) + 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!') + 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() diff --git a/tests/asgi/urls.py b/tests/asgi/urls.py index 22d85604d1..e6c74ab488 100644 --- a/tests/asgi/urls.py +++ b/tests/asgi/urls.py @@ -5,14 +5,14 @@ from django.urls import path def hello(request): - name = request.GET.get('name') or 'World' - return HttpResponse('Hello %s!' % name) + name = request.GET.get("name") or "World" + return HttpResponse("Hello %s!" % name) def hello_meta(request): return HttpResponse( - 'From %s' % request.META.get('HTTP_REFERER') or '', - content_type=request.META.get('CONTENT_TYPE'), + "From %s" % request.META.get("HTTP_REFERER") or "", + content_type=request.META.get("CONTENT_TYPE"), ) @@ -32,8 +32,8 @@ test_filename = __file__ urlpatterns = [ - path('', hello), - path('file/', lambda x: FileResponse(open(test_filename, 'rb'))), - path('meta/', hello_meta), - path('wait/', sync_waiter), + path("", hello), + path("file/", lambda x: FileResponse(open(test_filename, "rb"))), + path("meta/", hello_meta), + path("wait/", sync_waiter), ] |