django-rest-framework/tests/test_request.py

384 lines
13 KiB
Python
Raw Normal View History

"""
Tests for content parsing, and form-overloaded content parsing.
"""
import copy
import os.path
import sys
import tempfile
import pytest
2012-11-24 21:18:32 +04:00
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.middleware import AuthenticationMiddleware
2015-06-25 23:55:51 +03:00
from django.contrib.auth.models import User
2012-11-24 21:18:32 +04:00
from django.contrib.sessions.middleware import SessionMiddleware
from django.core.files.uploadedfile import SimpleUploadedFile
from django.http.request import RawPostDataException
from django.test import TestCase, override_settings
from django.urls import path
2015-06-25 23:55:51 +03:00
from rest_framework import status
from rest_framework.authentication import SessionAuthentication
2015-09-17 18:40:47 +03:00
from rest_framework.parsers import BaseParser, FormParser, MultiPartParser
from rest_framework.request import Request, WrappedAttributeError
from rest_framework.response import Response
2015-06-25 23:55:51 +03:00
from rest_framework.test import APIClient, APIRequestFactory
from rest_framework.views import APIView
2013-06-28 20:17:39 +04:00
factory = APIRequestFactory()
class TestInitializer(TestCase):
def test_request_type(self):
request = Request(factory.get('/'))
message = (
'The `request` argument must be an instance of '
'`django.http.HttpRequest`, not `rest_framework.request.Request`.'
)
with self.assertRaisesMessage(AssertionError, message):
Request(request)
2012-10-15 01:43:07 +04:00
class PlainTextParser(BaseParser):
media_type = 'text/plain'
def parse(self, stream, media_type=None, parser_context=None):
2012-10-15 01:43:07 +04:00
"""
Returns a 2-tuple of `(data, files)`.
`data` will simply be a string representing the body of the request.
`files` will always be `None`.
"""
return stream.read()
class TestContentParsing(TestCase):
def test_standard_behaviour_determines_no_content_GET(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns empty QueryDict for GET request.
"""
request = Request(factory.get('/'))
2017-01-12 19:41:27 +03:00
assert request.data == {}
def test_standard_behaviour_determines_no_content_HEAD(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns empty QueryDict for HEAD request.
"""
request = Request(factory.head('/'))
2017-01-12 19:41:27 +03:00
assert request.data == {}
2012-09-27 15:50:18 +04:00
def test_request_DATA_with_form_content(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns content for POST request with form content.
"""
data = {'qwerty': 'uiop'}
request = Request(factory.post('/', data))
2012-10-05 17:48:33 +04:00
request.parsers = (FormParser(), MultiPartParser())
2017-01-12 19:41:27 +03:00
assert list(request.data.items()) == list(data.items())
2012-09-27 15:50:18 +04:00
def test_request_DATA_with_text_content(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns content for POST request with
non-form content.
"""
content = b'qwerty'
content_type = 'text/plain'
request = Request(factory.post('/', content, content_type=content_type))
2012-10-05 17:48:33 +04:00
request.parsers = (PlainTextParser(),)
2017-01-12 19:41:27 +03:00
assert request.data == content
2012-09-27 15:50:18 +04:00
def test_request_POST_with_form_content(self):
"""
Ensure request.POST returns content for POST request with form content.
"""
data = {'qwerty': 'uiop'}
request = Request(factory.post('/', data))
2012-10-05 17:48:33 +04:00
request.parsers = (FormParser(), MultiPartParser())
2017-01-12 19:41:27 +03:00
assert list(request.POST.items()) == list(data.items())
2012-09-27 15:50:18 +04:00
def test_request_POST_with_files(self):
"""
Ensure request.POST returns no content for POST request with file content.
"""
upload = SimpleUploadedFile("file.txt", b"file_content")
request = Request(factory.post('/', {'upload': upload}))
request.parsers = (FormParser(), MultiPartParser())
assert list(request.POST) == []
assert list(request.FILES) == ['upload']
def test_standard_behaviour_determines_form_content_PUT(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns content for PUT request with form content.
"""
data = {'qwerty': 'uiop'}
2013-06-28 20:17:39 +04:00
request = Request(factory.put('/', data))
2012-10-05 17:48:33 +04:00
request.parsers = (FormParser(), MultiPartParser())
2017-01-12 19:41:27 +03:00
assert list(request.data.items()) == list(data.items())
def test_standard_behaviour_determines_non_form_content_PUT(self):
"""
2015-07-30 17:26:42 +03:00
Ensure request.data returns content for PUT request with
non-form content.
"""
content = b'qwerty'
content_type = 'text/plain'
request = Request(factory.put('/', content, content_type=content_type))
2012-10-05 17:48:33 +04:00
request.parsers = (PlainTextParser(), )
2017-01-12 19:41:27 +03:00
assert request.data == content
def test_calling_data_fails_when_attribute_error_is_raised(self):
"""
Ensure attribute errors raised when parsing are properly re-raised.
"""
expected_message = "Internal error"
class BrokenParser:
media_type = "application/json"
def parse(self, *args, **kwargs):
raise AttributeError(expected_message)
http_request = factory.post('/', data={}, format="json")
request = Request(http_request)
request.parsers = (BrokenParser,)
with self.assertRaisesMessage(WrappedAttributeError, expected_message):
request.data
class MockView(APIView):
authentication_classes = (SessionAuthentication,)
def post(self, request):
if request.POST.get('example') is not None:
return Response(status=status.HTTP_200_OK)
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
class EchoView(APIView):
def post(self, request):
response = Response(status=status.HTTP_200_OK, data=request.data)
response._request = request # test client sets `request` input
return response
class FileUploadView(APIView):
def post(self, request):
filenames = [file.temporary_file_path() for file in request.FILES.values()]
for filename in filenames:
assert os.path.exists(filename)
return Response(status=status.HTTP_200_OK, data=filenames)
urlpatterns = [
path('', MockView.as_view()),
path('echo/', EchoView.as_view()),
path('upload/', FileUploadView.as_view())
]
@override_settings(
ROOT_URLCONF='tests.test_request',
FILE_UPLOAD_HANDLERS=['django.core.files.uploadhandler.TemporaryFileUploadHandler'])
class FileUploadTests(TestCase):
def test_fileuploads_closed_at_request_end(self):
with tempfile.NamedTemporaryFile() as f:
response = self.client.post('/upload/', {'file': f})
# sanity check that file was processed
assert len(response.data) == 1
for file in response.data:
assert not os.path.exists(file)
@override_settings(ROOT_URLCONF='tests.test_request')
class TestContentParsingWithAuthentication(TestCase):
def setUp(self):
2013-06-28 20:50:30 +04:00
self.csrf_client = APIClient(enforce_csrf_checks=True)
self.username = 'john'
self.email = 'lennon@thebeatles.com'
self.password = 'password'
self.user = User.objects.create_user(self.username, self.email, self.password)
def test_user_logged_in_authentication_has_POST_when_not_logged_in(self):
"""
Ensures request.POST exists after SessionAuthentication when user
doesn't log in.
"""
content = {'example': 'example'}
response = self.client.post('/', content)
2017-01-12 19:41:27 +03:00
assert status.HTTP_200_OK == response.status_code
response = self.csrf_client.post('/', content)
2017-01-12 19:41:27 +03:00
assert status.HTTP_200_OK == response.status_code
2012-11-24 21:18:32 +04:00
class TestUserSetter(TestCase):
def setUp(self):
# Pass request object through session middleware so session is
# available to login and logout functions
2014-12-11 22:20:46 +03:00
self.wrapped_request = factory.get('/')
self.request = Request(self.wrapped_request)
def dummy_get_response(request): # pragma: no cover
return None
SessionMiddleware(dummy_get_response).process_request(self.wrapped_request)
AuthenticationMiddleware(dummy_get_response).process_request(self.wrapped_request)
2012-11-24 21:18:32 +04:00
User.objects.create_user('ringo', 'starr@thebeatles.com', 'yellow')
self.user = authenticate(username='ringo', password='yellow')
def test_user_can_be_set(self):
self.request.user = self.user
2017-01-12 19:41:27 +03:00
assert self.request.user == self.user
2012-11-24 21:18:32 +04:00
def test_user_can_login(self):
login(self.request, self.user)
2017-01-12 19:41:27 +03:00
assert self.request.user == self.user
2012-11-24 21:18:32 +04:00
def test_user_can_logout(self):
self.request.user = self.user
assert not self.request.user.is_anonymous
2012-11-24 21:18:32 +04:00
logout(self.request)
assert self.request.user.is_anonymous
2012-12-21 03:48:10 +04:00
2014-12-11 22:20:46 +03:00
def test_logged_in_user_is_set_on_wrapped_request(self):
login(self.request, self.user)
2017-01-12 19:41:27 +03:00
assert self.wrapped_request.user == self.user
2014-12-11 22:20:46 +03:00
2015-02-09 20:19:22 +03:00
def test_calling_user_fails_when_attribute_error_is_raised(self):
"""
This proves that when an AttributeError is raised inside of the request.user
property, that we can handle this and report the true, underlying error.
"""
class AuthRaisesAttributeError:
2015-02-09 20:19:22 +03:00
def authenticate(self, request):
self.MISSPELLED_NAME_THAT_DOESNT_EXIST
2015-02-09 20:19:22 +03:00
request = Request(self.wrapped_request, authenticators=(AuthRaisesAttributeError(),))
2015-02-09 20:19:22 +03:00
# The middleware processes the underlying Django request, sets anonymous user
assert self.wrapped_request.user.is_anonymous
# The DRF request object does not have a user and should run authenticators
expected = r"no attribute 'MISSPELLED_NAME_THAT_DOESNT_EXIST'"
with pytest.raises(WrappedAttributeError, match=expected):
request.user
with pytest.raises(WrappedAttributeError, match=expected):
hasattr(request, 'user')
with pytest.raises(WrappedAttributeError, match=expected):
login(request, self.user)
2012-12-21 03:48:10 +04:00
2015-02-09 20:22:13 +03:00
class TestAuthSetter(TestCase):
2012-12-21 03:48:10 +04:00
def test_auth_can_be_set(self):
request = Request(factory.get('/'))
request.auth = 'DUMMY'
2017-01-12 19:41:27 +03:00
assert request.auth == 'DUMMY'
class TestSecure(TestCase):
def test_default_secure_false(self):
request = Request(factory.get('/', secure=False))
2017-01-12 19:41:27 +03:00
assert request.scheme == 'http'
def test_default_secure_true(self):
request = Request(factory.get('/', secure=True))
2017-01-12 19:41:27 +03:00
assert request.scheme == 'https'
class TestHttpRequest(TestCase):
2020-03-29 14:01:14 +03:00
def test_repr(self):
http_request = factory.get('/path')
request = Request(http_request)
assert repr(request) == "<rest_framework.request.Request: GET '/path'>"
def test_attribute_access_proxy(self):
http_request = factory.get('/')
request = Request(http_request)
inner_sentinel = object()
http_request.inner_property = inner_sentinel
assert request.inner_property is inner_sentinel
outer_sentinel = object()
request.inner_property = outer_sentinel
assert request.inner_property is outer_sentinel
def test_exception_proxy(self):
# ensure the exception message is not for the underlying WSGIRequest
http_request = factory.get('/')
request = Request(http_request)
message = "'Request' object has no attribute 'inner_property'"
with self.assertRaisesMessage(AttributeError, message):
request.inner_property
@override_settings(ROOT_URLCONF='tests.test_request')
def test_duplicate_request_stream_parsing_exception(self):
"""
Check assumption that duplicate stream parsing will result in a
`RawPostDataException` being raised.
"""
response = APIClient().post('/echo/', data={'a': 'b'}, format='json')
request = response._request
# ensure that request stream was consumed by json parser
assert request.content_type.startswith('application/json')
assert response.data == {'a': 'b'}
# pass same HttpRequest to view, stream already consumed
with pytest.raises(RawPostDataException):
EchoView.as_view()(request._request)
@override_settings(ROOT_URLCONF='tests.test_request')
def test_duplicate_request_form_data_access(self):
"""
Form data is copied to the underlying django request for middleware
and file closing reasons. Duplicate processing of a request with form
data is 'safe' in so far as accessing `request.POST` does not trigger
the duplicate stream parse exception.
"""
response = APIClient().post('/echo/', data={'a': 'b'})
request = response._request
# ensure that request stream was consumed by form parser
assert request.content_type.startswith('multipart/form-data')
assert response.data == {'a': ['b']}
# pass same HttpRequest to view, form data set on underlying request
response = EchoView.as_view()(request._request)
request = response._request
# ensure that request stream was consumed by form parser
assert request.content_type.startswith('multipart/form-data')
assert response.data == {'a': ['b']}
class TestDeepcopy(TestCase):
def test_deepcopy_works(self):
request = Request(factory.get('/', secure=False))
copy.deepcopy(request)
class TestTyping(TestCase):
@pytest.mark.skipif(
sys.version_info < (3, 7),
reason="subscriptable classes requires Python 3.7 or higher",
)
def test_request_is_subscriptable(self):
assert Request is Request["foo"]