mirror of https://github.com/django/django.git
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
|
import sys
|
||
|
|
|
||
|
|
from asgiref.sync import async_to_sync
|
||
|
|
from asgiref.testing import ApplicationCommunicator
|
||
|
|
|
||
|
|
from django.core.asgi import get_asgi_application
|
||
|
|
from django.core.signals import request_started
|
||
|
|
from django.db import close_old_connections
|
||
|
|
from django.test import SimpleTestCase, override_settings
|
||
|
|
|
||
|
|
from .urls import test_filename
|
||
|
|
|
||
|
|
|
||
|
|
@override_settings(ROOT_URLCONF='asgi.urls')
|
||
|
|
class ASGITest(SimpleTestCase):
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
request_started.disconnect(close_old_connections)
|
||
|
|
|
||
|
|
def _get_scope(self, **kwargs):
|
||
|
|
return {
|
||
|
|
'type': 'http',
|
||
|
|
'asgi': {'version': '3.0', 'spec_version': '2.1'},
|
||
|
|
'http_version': '1.1',
|
||
|
|
'method': 'GET',
|
||
|
|
'query_string': b'',
|
||
|
|
'server': ('testserver', 80),
|
||
|
|
**kwargs,
|
||
|
|
}
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
request_started.connect(close_old_connections)
|
||
|
|
|
||
|
|
@async_to_sync
|
||
|
|
async def test_get_asgi_application(self):
|
||
|
|
"""
|
||
|
|
get_asgi_application() returns a functioning ASGI callable.
|
||
|
|
"""
|
||
|
|
application = get_asgi_application()
|
||
|
|
# Construct HTTP request.
|
||
|
|
communicator = ApplicationCommunicator(application, self._get_scope(path='/'))
|
||
|
|
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_to_sync
|
||
|
|
async def test_file_response(self):
|
||
|
|
"""
|
||
|
|
Makes sure that FileResponse works over ASGI.
|
||
|
|
"""
|
||
|
|
application = get_asgi_application()
|
||
|
|
# Construct HTTP request.
|
||
|
|
communicator = ApplicationCommunicator(application, self._get_scope(path='/file/'))
|
||
|
|
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.startswith('win') 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)
|