Upgrade djangorestframework to v3.1
* Upgrade edx-submissions * Upgrade edx-ora2 * Upgrade edx-val * Upgrade edx-proctoring * Update all edx-platform code that depends on DRF, including: - auth_exchange - cors_csrf - embargo - enrollment - util - commerce - course_structure - discussion_api - mobile_api - notifier_api - teams - credit - profile_images - user_api - lib/api (OAuth2 and pagination)
This commit is contained in:
Will Daly
@@ -1,10 +1,11 @@
|
||||
""" Common Authentication Handlers used across projects. """
|
||||
from rest_framework import authentication
|
||||
from rest_framework.authentication import SessionAuthentication
|
||||
from rest_framework_oauth.authentication import OAuth2Authentication
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
from rest_framework.compat import oauth2_provider, provider_now
|
||||
from rest_framework_oauth.compat import oauth2_provider, provider_now
|
||||
|
||||
|
||||
class SessionAuthenticationAllowInactiveUser(authentication.SessionAuthentication):
|
||||
class SessionAuthenticationAllowInactiveUser(SessionAuthentication):
|
||||
"""Ensure that the user is logged in, but do not require the account to be active.
|
||||
|
||||
We use this in the special case that a user has created an account,
|
||||
@@ -51,7 +52,7 @@ class SessionAuthenticationAllowInactiveUser(authentication.SessionAuthenticatio
|
||||
return (user, None)
|
||||
|
||||
|
||||
class OAuth2AuthenticationAllowInactiveUser(authentication.OAuth2Authentication):
|
||||
class OAuth2AuthenticationAllowInactiveUser(OAuth2Authentication):
|
||||
"""
|
||||
This is a temporary workaround while the is_active field on the user is coupled
|
||||
with whether or not the user has verified ownership of their claimed email address.
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
"""Fields useful for edX API implementations."""
|
||||
from django.core.exceptions import ValidationError
|
||||
|
||||
from rest_framework.serializers import CharField, Field
|
||||
from rest_framework.serializers import Field
|
||||
|
||||
|
||||
class ExpandableField(Field):
|
||||
@@ -18,25 +16,21 @@ class ExpandableField(Field):
|
||||
self.expanded = kwargs.pop('expanded_serializer')
|
||||
super(ExpandableField, self).__init__(**kwargs)
|
||||
|
||||
def field_to_native(self, obj, field_name):
|
||||
"""Converts obj to a native representation, using the expanded serializer if the context requires it."""
|
||||
if 'expand' in self.context and field_name in self.context['expand']:
|
||||
self.expanded.initialize(self, field_name)
|
||||
return self.expanded.field_to_native(obj, field_name)
|
||||
else:
|
||||
self.collapsed.initialize(self, field_name)
|
||||
return self.collapsed.field_to_native(obj, field_name)
|
||||
def to_representation(self, obj):
|
||||
"""
|
||||
Return a representation of the field that is either expanded or collapsed.
|
||||
"""
|
||||
should_expand = self.field_name in self.context.get("expand", [])
|
||||
field = self.expanded if should_expand else self.collapsed
|
||||
|
||||
# Avoid double-binding the field, otherwise we'll get
|
||||
# an error about the source kwarg being redundant.
|
||||
if field.source is None:
|
||||
field.bind(self.field_name, self)
|
||||
|
||||
class NonEmptyCharField(CharField):
|
||||
"""
|
||||
A field that enforces non-emptiness even for partial updates.
|
||||
# Exclude fields that should not be expanded in the nested field
|
||||
if should_expand:
|
||||
nested_expand_fields = set(field.context.get("expand", []))
|
||||
self.expanded.context["expand"] = list(nested_expand_fields - self.exclude_expand_fields)
|
||||
|
||||
This is necessary because prior to version 3, DRF skips validation for empty
|
||||
values. Thus, CharField's min_length and RegexField cannot be used to
|
||||
enforce this constraint.
|
||||
"""
|
||||
def validate(self, value):
|
||||
super(NonEmptyCharField, self).validate(value)
|
||||
if not value.strip():
|
||||
raise ValidationError(self.error_messages["required"])
|
||||
return field.to_representation(obj)
|
||||
|
||||
33
openedx/core/lib/api/mixins.py
Normal file
33
openedx/core/lib/api/mixins.py
Normal file
@@ -0,0 +1,33 @@
|
||||
"""
|
||||
Django Rest Framework view mixins.
|
||||
"""
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.http import Http404
|
||||
from rest_framework import status
|
||||
from rest_framework.mixins import CreateModelMixin
|
||||
from rest_framework.response import Response
|
||||
|
||||
|
||||
class PutAsCreateMixin(CreateModelMixin):
|
||||
"""
|
||||
Backwards compatibility with Django Rest Framework v2, which allowed
|
||||
creation of a new resource using PUT.
|
||||
"""
|
||||
|
||||
def update(self, request, *args, **kwargs):
|
||||
"""
|
||||
Create/update course modes for a course.
|
||||
"""
|
||||
# First, try to update the existing instance
|
||||
try:
|
||||
try:
|
||||
return super(PutAsCreateMixin, self).update(request, *args, **kwargs)
|
||||
except Http404:
|
||||
# If no instance exists yet, create it.
|
||||
# This is backwards-compatible with the behavior of DRF v2.
|
||||
return super(PutAsCreateMixin, self).create(request, *args, **kwargs)
|
||||
|
||||
# Backwards compatibility with DRF v2 behavior, which would catch model-level
|
||||
# validation errors and return a 400
|
||||
except ValidationError as err:
|
||||
return Response(err.messages, status=status.HTTP_400_BAD_REQUEST)
|
||||
@@ -3,6 +3,31 @@
|
||||
from django.http import Http404
|
||||
from django.core.paginator import Paginator, InvalidPage
|
||||
|
||||
from rest_framework.response import Response
|
||||
from rest_framework import pagination
|
||||
|
||||
|
||||
class DefaultPagination(pagination.PageNumberPagination):
|
||||
"""
|
||||
Default paginator for APIs in edx-platform.
|
||||
|
||||
This is configured in settings to be automatically used
|
||||
by any subclass of Django Rest Framework's generic API views.
|
||||
"""
|
||||
page_size_query_param = "page_size"
|
||||
|
||||
def get_paginated_response(self, data):
|
||||
"""
|
||||
Annotate the response with pagination information.
|
||||
"""
|
||||
return Response({
|
||||
'next': self.get_next_link(),
|
||||
'previous': self.get_previous_link(),
|
||||
'count': self.page.paginator.count,
|
||||
'num_pages': self.page.paginator.num_pages,
|
||||
'results': data
|
||||
})
|
||||
|
||||
|
||||
def paginate_search_results(object_class, search_results, page_size, page):
|
||||
"""
|
||||
|
||||
@@ -1,32 +1,8 @@
|
||||
from rest_framework import pagination, serializers
|
||||
"""
|
||||
Serializers to be used in APIs.
|
||||
"""
|
||||
|
||||
|
||||
class PaginationSerializer(pagination.PaginationSerializer):
|
||||
"""
|
||||
Custom PaginationSerializer for openedx.
|
||||
|
||||
Adds the following fields:
|
||||
- num_pages: total number of pages
|
||||
- current_page: the current page being returned
|
||||
- start: the index of the first page item within the overall collection
|
||||
"""
|
||||
start_page = 1 # django Paginator.page objects have 1-based indexes
|
||||
num_pages = serializers.Field(source='paginator.num_pages')
|
||||
current_page = serializers.SerializerMethodField('get_current_page')
|
||||
start = serializers.SerializerMethodField('get_start')
|
||||
sort_order = serializers.SerializerMethodField('get_sort_order')
|
||||
|
||||
def get_current_page(self, page):
|
||||
"""Get the current page"""
|
||||
return page.number
|
||||
|
||||
def get_start(self, page):
|
||||
"""Get the index of the first page item within the overall collection"""
|
||||
return (self.get_current_page(page) - self.start_page) * page.paginator.per_page
|
||||
|
||||
def get_sort_order(self, page): # pylint: disable=unused-argument
|
||||
"""Get the order by which this collection was sorted"""
|
||||
return self.context.get('sort_order')
|
||||
from rest_framework import serializers
|
||||
|
||||
|
||||
class CollapsedReferenceSerializer(serializers.HyperlinkedModelSerializer):
|
||||
@@ -54,9 +30,10 @@ class CollapsedReferenceSerializer(serializers.HyperlinkedModelSerializer):
|
||||
|
||||
super(CollapsedReferenceSerializer, self).__init__(*args, **kwargs)
|
||||
|
||||
self.fields[id_source] = serializers.CharField(read_only=True, source=id_source)
|
||||
self.fields[id_source] = serializers.CharField(read_only=True)
|
||||
self.fields['url'].view_name = view_name
|
||||
self.fields['url'].lookup_field = lookup_field
|
||||
self.fields['url'].lookup_url_kwarg = lookup_field
|
||||
|
||||
class Meta(object):
|
||||
"""Defines meta information for the ModelSerializer.
|
||||
|
||||
@@ -1,63 +1,235 @@
|
||||
"""Tests for util.authentication module."""
|
||||
"""
|
||||
Tests for OAuth2. This module is copied from django-rest-framework-oauth (tests/test_authentication.py)
|
||||
and updated to use our subclass of OAuth2Authentication.
|
||||
"""
|
||||
|
||||
from __future__ import unicode_literals
|
||||
import datetime
|
||||
|
||||
from django.conf.urls import patterns, url, include
|
||||
from django.contrib.auth.models import User
|
||||
from django.http import HttpResponse
|
||||
from django.test import TestCase
|
||||
from django.utils import unittest
|
||||
from django.utils.http import urlencode
|
||||
|
||||
from rest_framework import status
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework_oauth import permissions
|
||||
from rest_framework_oauth.compat import oauth2_provider, oauth2_provider_scope
|
||||
from rest_framework.test import APIRequestFactory, APIClient
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from mock import patch
|
||||
from django.conf import settings
|
||||
from rest_framework import permissions
|
||||
from rest_framework.compat import patterns, url
|
||||
from rest_framework.tests import test_authentication
|
||||
from provider import scope, constants
|
||||
from unittest import skipUnless
|
||||
|
||||
from ..authentication import OAuth2AuthenticationAllowInactiveUser
|
||||
|
||||
factory = APIRequestFactory() # pylint: disable=invalid-name
|
||||
|
||||
class OAuth2AuthAllowInactiveUserDebug(OAuth2AuthenticationAllowInactiveUser):
|
||||
"""
|
||||
A debug class analogous to the OAuth2AuthenticationDebug class that tests
|
||||
the OAuth2 flow with the access token sent in a query param."""
|
||||
|
||||
class MockView(APIView): # pylint: disable=missing-docstring
|
||||
permission_classes = (IsAuthenticated,)
|
||||
|
||||
def get(self, request): # pylint: disable=missing-docstring,unused-argument
|
||||
return HttpResponse({'a': 1, 'b': 2, 'c': 3})
|
||||
|
||||
def post(self, request): # pylint: disable=missing-docstring,unused-argument
|
||||
return HttpResponse({'a': 1, 'b': 2, 'c': 3})
|
||||
|
||||
def put(self, request): # pylint: disable=missing-docstring,unused-argument
|
||||
return HttpResponse({'a': 1, 'b': 2, 'c': 3})
|
||||
|
||||
|
||||
# This is the a change we've made from the django-rest-framework-oauth version
|
||||
# of these tests. We're subclassing our custom OAuth2AuthenticationAllowInactiveUser
|
||||
# instead of OAuth2Authentication.
|
||||
class OAuth2AuthenticationDebug(OAuth2AuthenticationAllowInactiveUser): # pylint: disable=missing-docstring
|
||||
allow_query_params_token = True
|
||||
|
||||
|
||||
# The following patch overrides the URL patterns for the MockView class used in
|
||||
# rest_framework.tests.test_authentication so that the corresponding AllowInactiveUser
|
||||
# classes are tested instead.
|
||||
@skipUnless(settings.FEATURES.get('ENABLE_OAUTH2_PROVIDER'), 'OAuth2 not enabled')
|
||||
@patch.object(
|
||||
test_authentication,
|
||||
'urlpatterns',
|
||||
patterns(
|
||||
'',
|
||||
url(
|
||||
r'^oauth2-test/$',
|
||||
test_authentication.MockView.as_view(authentication_classes=[OAuth2AuthenticationAllowInactiveUser])
|
||||
),
|
||||
url(
|
||||
r'^oauth2-test-debug/$',
|
||||
test_authentication.MockView.as_view(authentication_classes=[OAuth2AuthAllowInactiveUserDebug])
|
||||
),
|
||||
url(
|
||||
r'^oauth2-with-scope-test/$',
|
||||
test_authentication.MockView.as_view(
|
||||
authentication_classes=[OAuth2AuthenticationAllowInactiveUser],
|
||||
permission_classes=[permissions.TokenHasReadWriteScope]
|
||||
)
|
||||
urlpatterns = patterns(
|
||||
'',
|
||||
url(r'^oauth2/', include('provider.oauth2.urls', namespace='oauth2')),
|
||||
url(r'^oauth2-test/$', MockView.as_view(authentication_classes=[OAuth2AuthenticationAllowInactiveUser])),
|
||||
url(r'^oauth2-test-debug/$', MockView.as_view(authentication_classes=[OAuth2AuthenticationDebug])),
|
||||
url(
|
||||
r'^oauth2-with-scope-test/$',
|
||||
MockView.as_view(
|
||||
authentication_classes=[OAuth2AuthenticationAllowInactiveUser],
|
||||
permission_classes=[permissions.TokenHasReadWriteScope]
|
||||
)
|
||||
)
|
||||
),
|
||||
)
|
||||
class OAuth2AuthenticationAllowInactiveUserTestCase(test_authentication.OAuth2Tests):
|
||||
"""
|
||||
Tests the OAuth2AuthenticationAllowInactiveUser class by running all the existing tests in
|
||||
OAuth2Tests but with the is_active flag on the user set to False.
|
||||
"""
|
||||
def setUp(self):
|
||||
super(OAuth2AuthenticationAllowInactiveUserTestCase, self).setUp()
|
||||
|
||||
# set the user's is_active flag to False.
|
||||
|
||||
class OAuth2Tests(TestCase):
|
||||
"""OAuth 2.0 authentication"""
|
||||
urls = 'openedx.core.lib.api.tests.test_authentication'
|
||||
|
||||
def setUp(self):
|
||||
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)
|
||||
|
||||
self.CLIENT_ID = 'client_key' # pylint: disable=invalid-name
|
||||
self.CLIENT_SECRET = 'client_secret' # pylint: disable=invalid-name
|
||||
self.ACCESS_TOKEN = "access_token" # pylint: disable=invalid-name
|
||||
self.REFRESH_TOKEN = "refresh_token" # pylint: disable=invalid-name
|
||||
|
||||
self.oauth2_client = oauth2_provider.oauth2.models.Client.objects.create(
|
||||
client_id=self.CLIENT_ID,
|
||||
client_secret=self.CLIENT_SECRET,
|
||||
redirect_uri='',
|
||||
client_type=0,
|
||||
name='example',
|
||||
user=None,
|
||||
)
|
||||
|
||||
self.access_token = oauth2_provider.oauth2.models.AccessToken.objects.create(
|
||||
token=self.ACCESS_TOKEN,
|
||||
client=self.oauth2_client,
|
||||
user=self.user,
|
||||
)
|
||||
self.refresh_token = oauth2_provider.oauth2.models.RefreshToken.objects.create(
|
||||
user=self.user,
|
||||
access_token=self.access_token,
|
||||
client=self.oauth2_client
|
||||
)
|
||||
|
||||
# This is the a change we've made from the django-rest-framework-oauth version
|
||||
# of these tests.
|
||||
self.user.is_active = False
|
||||
self.user.save()
|
||||
|
||||
# This is the a change we've made from the django-rest-framework-oauth version
|
||||
# of these tests.
|
||||
# Override the SCOPE_NAME_DICT setting for tests for oauth2-with-scope-test. This is
|
||||
# needed to support READ and WRITE scopes as they currently aren't supported by the
|
||||
# edx-auth2-provider, and their scope values collide with other scopes defined in the
|
||||
# edx-auth2-provider.
|
||||
scope.SCOPE_NAME_DICT = {'read': constants.READ, 'write': constants.WRITE}
|
||||
|
||||
def _create_authorization_header(self, token=None): # pylint: disable=missing-docstring
|
||||
return "Bearer {0}".format(token or self.access_token.token)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_type_failing(self):
|
||||
"""Ensure that a wrong token type lead to the correct HTTP error status code"""
|
||||
auth = "Wrong token-type-obviously"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_format_failing(self):
|
||||
"""Ensure that a wrong token format lead to the correct HTTP error status code"""
|
||||
auth = "Bearer wrong token format"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_failing(self):
|
||||
"""Ensure that a wrong token lead to the correct HTTP error status code"""
|
||||
auth = "Bearer wrong-token"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_missing(self):
|
||||
"""Ensure that a missing token lead to the correct HTTP error status code"""
|
||||
auth = "Bearer"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_passing_auth(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials succeed"""
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_passing_auth_url_transport(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials in form data succeed"""
|
||||
response = self.csrf_client.post(
|
||||
'/oauth2-test/',
|
||||
data={'access_token': self.access_token.token}
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_passing_auth_url_transport(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials in query succeed when DEBUG is True"""
|
||||
query = urlencode({'access_token': self.access_token.token})
|
||||
response = self.csrf_client.get('/oauth2-test-debug/?%s' % query)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_failing_auth_url_transport(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials in query fails when DEBUG is False"""
|
||||
query = urlencode({'access_token': self.access_token.token})
|
||||
response = self.csrf_client.get('/oauth2-test/?%s' % query)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_passing_auth(self):
|
||||
"""Ensure POSTing form over OAuth with correct credentials passes and does not require CSRF"""
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_token_removed_failing_auth(self):
|
||||
"""Ensure POSTing when there is no OAuth access token in db fails"""
|
||||
self.access_token.delete()
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_refresh_token_failing_auth(self):
|
||||
"""Ensure POSTing with refresh token instead of access token fails"""
|
||||
auth = self._create_authorization_header(token=self.refresh_token.token)
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_expired_access_token_failing_auth(self):
|
||||
"""Ensure POSTing with expired access token fails with an 'Invalid token' error"""
|
||||
self.access_token.expires = datetime.datetime.now() - datetime.timedelta(seconds=10) # 10 seconds late
|
||||
self.access_token.save()
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
self.assertIn('Invalid token', response.content)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_invalid_scope_failing_auth(self):
|
||||
"""Ensure POSTing with a readonly scope instead of a write scope fails"""
|
||||
read_only_access_token = self.access_token
|
||||
read_only_access_token.scope = oauth2_provider_scope.SCOPE_NAME_DICT['read']
|
||||
read_only_access_token.save()
|
||||
auth = self._create_authorization_header(token=read_only_access_token.token)
|
||||
response = self.csrf_client.get('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
response = self.csrf_client.post('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_valid_scope_passing_auth(self):
|
||||
"""Ensure POSTing with a write scope succeed"""
|
||||
read_write_access_token = self.access_token
|
||||
read_write_access_token.scope = oauth2_provider_scope.SCOPE_NAME_DICT['write']
|
||||
read_write_access_token.save()
|
||||
auth = self._create_authorization_header(token=read_write_access_token.token)
|
||||
response = self.csrf_client.post('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@@ -9,6 +9,7 @@ from django.utils.translation import ugettext as _
|
||||
from rest_framework import status, response
|
||||
from rest_framework.exceptions import APIException
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.request import clone_request
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.mixins import RetrieveModelMixin, UpdateModelMixin
|
||||
from rest_framework.generics import GenericAPIView
|
||||
@@ -193,3 +194,23 @@ class RetrievePatchAPIView(RetrieveModelMixin, UpdateModelMixin, GenericAPIView)
|
||||
add_serializer_errors(serializer, patch, field_errors)
|
||||
|
||||
return field_errors
|
||||
|
||||
def get_object_or_none(self):
|
||||
"""
|
||||
Retrieve an object or return None if the object can't be found.
|
||||
|
||||
NOTE: This replaces functionality that was removed in Django Rest Framework v3.1.
|
||||
"""
|
||||
try:
|
||||
return self.get_object()
|
||||
except Http404:
|
||||
if self.request.method == 'PUT':
|
||||
# For PUT-as-create operation, we need to ensure that we have
|
||||
# relevant permissions, as if this was a POST request. This
|
||||
# will either raise a PermissionDenied exception, or simply
|
||||
# return None.
|
||||
self.check_permissions(clone_request(self.request, 'POST'))
|
||||
else:
|
||||
# PATCH requests where the object does not exist should still
|
||||
# return a 404 response.
|
||||
raise
|
||||
|
||||
Reference in New Issue
Block a user