Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adjustments to library user access API endpoints required by frontend needs. (#24919)

Browse Source
This commit is contained in:
Fox Piacenti
2020-09-25 08:33:32 -05:00
committed by GitHub
parent a96a23ef94
commit 7bc10e0464
6 changed files with 157 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
openedx/core/djangoapps/content_libraries/api.py
View File

@@ -114,6 +114,10 @@ class InvalidNameError(ValueError):
""" The specified name/identifier is not valid """
class LibraryPermissionIntegrityError(IntegrityError):
""" Thrown when an operation would cause insane permissions. """
# Models:
@attr.s
@@ -412,6 +416,22 @@ def get_library_team(library_key):
]
def get_library_user_permissions(library_key, user):
"""
Fetch the specified user's access information. Will return None if no
permissions have been granted.
"""
ref = ContentLibrary.objects.get_by_key(library_key)
grant = ref.permission_grants.filter(user=user).first()
if grant is None:
return None
return ContentLibraryPermissionEntry(
user=grant.user,
group=grant.group,
access_level=grant.access_level,
)
def set_library_user_permissions(library_key, user, access_level):
"""
Change the specified user's level of access to this library.
@@ -419,6 +439,10 @@ def set_library_user_permissions(library_key, user, access_level):
access_level should be one of the AccessLevel values defined above.
"""
ref = ContentLibrary.objects.get_by_key(library_key)
current_grant = get_library_user_permissions(library_key, user)
if current_grant and current_grant.access_level == AccessLevel.ADMIN_LEVEL:
if not ref.permission_grants.filter(access_level=AccessLevel.ADMIN_LEVEL).exclude(user_id=user.id).exists():
raise LibraryPermissionIntegrityError(_('Cannot change or remove the access level for the only admin.'))
if access_level is None:
ref.permission_grants.filter(user=user).delete()
else:
@@ -528,7 +552,7 @@ def get_library_blocks(library_key, text_search=None):
for item in LibraryBlockIndexer.get_items(filter_terms=filter_terms, text_search=text_search)
if item is not None
]
except (ConnectionError) as e:
except (ElasticConnectionError) as e:
log.exception(e)
# If indexing is disabled, or connection to elastic failed

11
openedx/core/djangoapps/content_libraries/serializers.py
View File

@@ -57,12 +57,21 @@ class ContentLibraryPermissionLevelSerializer(serializers.Serializer):
access_level = serializers.ChoiceField(choices=ContentLibraryPermission.ACCESS_LEVEL_CHOICES)
class ContentLibraryAddPermissionByEmailSerializer(serializers.Serializer):
"""
Serializer for adding a new user and granting their access level via their email address.
"""
access_level = serializers.ChoiceField(choices=ContentLibraryPermission.ACCESS_LEVEL_CHOICES)
email = serializers.EmailField()
class ContentLibraryPermissionSerializer(ContentLibraryPermissionLevelSerializer):
"""
Serializer for a ContentLibraryPermission object, which grants either a user
or a group permission to view a content library.
"""
user_id = serializers.IntegerField(source="user.id", allow_null=True)
email = serializers.EmailField(source="user.email", read_only=True, default=None)
username = serializers.CharField(source="user.username", read_only=True, default=None)
group_name = serializers.CharField(source="group.name", allow_null=True, allow_blank=False, default=None)

26
openedx/core/djangoapps/content_libraries/tests/base.py
View File

@@ -30,7 +30,7 @@ URL_LIB_LINKS = URL_LIB_DETAIL + 'links/' # Get the list of links in this libra
URL_LIB_COMMIT = URL_LIB_DETAIL + 'commit/' # Commit (POST) or revert (DELETE) all pending changes to this library
URL_LIB_BLOCKS = URL_LIB_DETAIL + 'blocks/' # Get the list of XBlocks in this library, or add a new one
URL_LIB_TEAM = URL_LIB_DETAIL + 'team/' # Get the list of users/groups authorized to use this library
URL_LIB_TEAM_USER = URL_LIB_TEAM + 'user/{user_id}/' # Add/edit/remove a user's permission to use this library
URL_LIB_TEAM_USER = URL_LIB_TEAM + 'user/{username}/' # Add/edit/remove a user's permission to use this library
URL_LIB_TEAM_GROUP = URL_LIB_TEAM + 'group/{group_name}/' # Add/edit/remove a group's permission to use this library
URL_LIB_BLOCK = URL_PREFIX + 'blocks/{block_key}/' # Get data about a block, or delete it
URL_LIB_BLOCK_OLX = URL_LIB_BLOCK + 'olx/' # Get or set the OLX of the specified XBlock
@@ -225,13 +225,25 @@ class ContentLibrariesRestApiTest(APITestCase):
""" Get the list of users/groups authorized to use this library """
return self._api('get', URL_LIB_TEAM.format(lib_key=lib_key), None, expect_response)
def _set_user_access_level(self, lib_key, user_id, access_level, expect_response=200):
def _get_user_access_level(self, lib_key, username, expect_response=200):
""" Fetch a user's access level """
url = URL_LIB_TEAM_USER.format(lib_key=lib_key, username=username)
return self._api('get', url, None, expect_response)
def _add_user_by_email(self, lib_key, email, access_level, expect_response=200):
""" Add a user of a specified permission level by their email address. """
url = URL_LIB_TEAM.format(lib_key=lib_key)
return self._api('post', url, {"access_level": access_level, "email": email}, expect_response)
def _set_user_access_level(self, lib_key, username, access_level, expect_response=200):
""" Change the specified user's access level """
url = URL_LIB_TEAM_USER.format(lib_key=lib_key, user_id=user_id)
if access_level is None:
return self._api('delete', url, None, expect_response)
else:
return self._api('put', url, {"access_level": access_level}, expect_response)
url = URL_LIB_TEAM_USER.format(lib_key=lib_key, username=username)
return self._api('put', url, {"access_level": access_level}, expect_response)
def _remove_user_access(self, lib_key, username, expect_response=200):
""" Should effectively be the same as the above with access_level=None, but using the delete HTTP verb. """
url = URL_LIB_TEAM_USER.format(lib_key=lib_key, username=username)
return self._api('delete', url, None, expect_response)
def _set_group_access_level(self, lib_key, group_name, access_level, expect_response=200):
""" Change the specified group's access level """

56
openedx/core/djangoapps/content_libraries/tests/test_content_libraries.py
View File

@@ -2,7 +2,6 @@
"""
Tests for Blockstore-based Content Libraries
"""
import unittest
from uuid import UUID
import ddt
@@ -13,7 +12,6 @@ from mock import patch
from organizations.models import Organization
from openedx.core.djangoapps.content_libraries.tests.base import ContentLibrariesRestApiTest, elasticsearch_test
from openedx.core.djangoapps.content_libraries.api import BlockLimitReachedError
from student.tests.factories import UserFactory
@@ -355,6 +353,7 @@ class ContentLibrariesTest(ContentLibrariesRestApiTest):
author_group_member = UserFactory.create(username="GroupMember", email="groupmember@example.com")
author_group_member.groups.add(group)
random_user = UserFactory.create(username="Random", email="random@example.com")
never_added = UserFactory.create(username="Never", email="never@example.com")
# Library CRUD #########################################################
@@ -369,22 +368,32 @@ class ContentLibrariesTest(ContentLibrariesRestApiTest):
# By default, the creator of a new library is the only admin
data = self._get_library_team(lib_id)
self.assertEqual(len(data), 1)
self.assertDictContainsEntries(data[0], {"user_id": admin.pk, "group_name": None, "access_level": "admin"})
self.assertDictContainsEntries(data[0], {
"username": admin.username, "group_name": None, "access_level": "admin",
})
# Add the other users to the content library:
self._set_user_access_level(lib_id, author.pk, access_level="author")
self._set_user_access_level(lib_id, reader.pk, access_level="read")
self._set_user_access_level(lib_id, author.username, access_level="author")
# Delete it, add it again.
self._remove_user_access(lib_id, author.username)
self._set_user_access_level(lib_id, author.username, access_level="author")
# Add one of them via the email-based creation endpoint.
self._add_user_by_email(lib_id, reader.email, access_level="read")
self._set_group_access_level(lib_id, group.name, access_level="author")
team_response = self._get_library_team(lib_id)
self.assertEqual(len(team_response), 4)
# We'll use this one later.
reader_grant = {"username": reader.username, "group_name": None, "access_level": "read"}
# The response should also always be sorted in a specific order (by username and group name):
expected_response = [
{"user_id": None, "group_name": "group1", "access_level": "author"},
{"user_id": admin.pk, "group_name": None, "access_level": "admin"},
{"user_id": author.pk, "group_name": None, "access_level": "author"},
{"user_id": reader.pk, "group_name": None, "access_level": "read"},
{"username": None, "group_name": "group1", "access_level": "author"},
{"username": admin.username, "group_name": None, "access_level": "admin"},
{"username": author.username, "group_name": None, "access_level": "author"},
reader_grant,
]
from pprint import pprint
pprint(team_response)
for entry, expected in zip(team_response, expected_response):
self.assertDictContainsEntries(entry, expected)
@@ -392,6 +401,7 @@ class ContentLibrariesTest(ContentLibrariesRestApiTest):
with self.as_user(random_user):
self._get_library(lib_id, expect_response=403)
self._get_library_team(lib_id, expect_response=403)
self._add_user_by_email(lib_id, never_added.email, access_level="read", expect_response=403)
# But every authorized user can:
for user in [admin, author, author_group_member]:
@@ -399,19 +409,25 @@ class ContentLibrariesTest(ContentLibrariesRestApiTest):
self._get_library(lib_id)
data = self._get_library_team(lib_id)
self.assertEqual(data, team_response)
data = self._get_user_access_level(lib_id, reader.username)
self.assertEqual(data, {**reader_grant, 'username': 'Reader', 'email': 'reader@example.com'})
# A user with only read permission can get data about the library but not the team:
with self.as_user(reader):
self._get_library(lib_id)
self._get_library_team(lib_id, expect_response=403)
self._get_user_access_level(lib_id, author.username, expect_response=403)
self._add_user_by_email(lib_id, never_added.email, access_level="read", expect_response=403)
# Users without admin access cannot delete the library nor change its team:
for user in [author, reader, author_group_member, random_user]:
with self.as_user(user):
self._delete_library(lib_id, expect_response=403)
self._set_user_access_level(lib_id, author.pk, access_level="admin", expect_response=403)
self._set_user_access_level(lib_id, admin.pk, access_level=None, expect_response=403)
self._set_user_access_level(lib_id, random_user.pk, access_level="read", expect_response=403)
self._set_user_access_level(lib_id, author.username, access_level="admin", expect_response=403)
self._set_user_access_level(lib_id, admin.username, access_level=None, expect_response=403)
self._set_user_access_level(lib_id, random_user.username, access_level="read", expect_response=403)
self._remove_user_access(lib_id, admin.username, expect_response=403)
self._add_user_by_email(lib_id, never_added.email, access_level="read", expect_response=403)
# Users with author access (or higher) can edit the library's properties:
with self.as_user(author):
@@ -480,6 +496,22 @@ class ContentLibrariesTest(ContentLibrariesRestApiTest):
self._commit_library_changes(lib_id)
self._revert_library_changes(lib_id) # This is a no-op after the commit, but should still have 200 response
def test_no_lockout(self):
"""
Test that administrators cannot be removed if they are the only administrator granted access.
"""
admin = UserFactory.create(username="Admin", email="admin@example.com")
successor = UserFactory.create(username="Successor", email="successor@example.com")
with self.as_user(admin):
lib = self._create_library(slug="permtest", title="Permission Test Library", description="Testing")
# Fail to downgrade permissions.
self._remove_user_access(lib_key=lib['id'], username=admin.username, expect_response=400)
# Promote another user.
self._set_user_access_level(
lib_key=lib['id'], username=successor.username, access_level="admin",
)
self._remove_user_access(lib_key=lib['id'], username=admin.username)
def test_library_blocks_with_links(self):
"""
Test that libraries can link to XBlocks in other content libraries

2
openedx/core/djangoapps/content_libraries/urls.py
View File

@@ -31,7 +31,7 @@ urlpatterns = [
# Get the list of users/groups who have permission to view/edit/administer this library:
url(r'^team/$', views.LibraryTeamView.as_view()),
# Add/Edit (PUT) or remove (DELETE) a user's permission to use this library
url(r'^team/user/(?P<user_id>\d+)/$', views.LibraryTeamUserView.as_view()),
url(r'^team/user/(?P<username>[^/]+)/$', views.LibraryTeamUserView.as_view()),
# Add/Edit (PUT) or remove (DELETE) a group's permission to use this library
url(r'^team/group/(?P<group_name>[^/]+)/$', views.LibraryTeamGroupView.as_view()),
])),

67
openedx/core/djangoapps/content_libraries/views.py
View File

@@ -7,6 +7,7 @@ import logging
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.shortcuts import get_object_or_404
from django.utils.translation import ugettext as _
import edx_api_doc_tools as apidocs
from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
from organizations.models import Organization
@@ -31,6 +32,7 @@ from openedx.core.djangoapps.content_libraries.serializers import (
LibraryXBlockOlxSerializer,
LibraryXBlockStaticFileSerializer,
LibraryXBlockStaticFilesSerializer,
ContentLibraryAddPermissionByEmailSerializer,
)
from openedx.core.lib.api.view_utils import view_auth_classes
@@ -211,6 +213,33 @@ class LibraryTeamView(APIView):
Note also the 'allow_public_' settings which can be edited by PATCHing the
library itself (LibraryDetailsView.patch).
"""
@convert_exceptions
def post(self, request, lib_key_str):
"""
Add a user to this content library via email, with permissions specified in the
request body.
"""
key = LibraryLocatorV2.from_string(lib_key_str)
api.require_permission_for_library_key(key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM)
serializer = ContentLibraryAddPermissionByEmailSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
user = User.objects.get(email=serializer.validated_data.get('email'))
except User.DoesNotExist:
raise ValidationError({'email': _('We could not find a user with that email address.')})
grant = api.get_library_user_permissions(key, user)
if grant:
return Response(
{'email': [_('This user already has access to this library.')]},
status=status.HTTP_400_BAD_REQUEST,
)
try:
api.set_library_user_permissions(key, user, access_level=serializer.validated_data["access_level"])
except api.LibraryPermissionIntegrityError as err:
raise ValidationError(detail=str(err))
grant = api.get_library_user_permissions(key, user)
return Response(ContentLibraryPermissionSerializer(grant).data)
@convert_exceptions
def get(self, request, lib_key_str):
"""
@@ -230,7 +259,7 @@ class LibraryTeamUserView(APIView):
library.
"""
@convert_exceptions
def put(self, request, lib_key_str, user_id):
def put(self, request, lib_key_str, username):
"""
Add a user to this content library, with permissions specified in the
request body.
@@ -239,20 +268,40 @@ class LibraryTeamUserView(APIView):
api.require_permission_for_library_key(key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM)
serializer = ContentLibraryPermissionLevelSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = get_object_or_404(User, pk=int(user_id))
api.set_library_user_permissions(key, user, access_level=serializer.validated_data["access_level"])
return Response({})
user = get_object_or_404(User, username=username)
try:
api.set_library_user_permissions(key, user, access_level=serializer.validated_data["access_level"])
except api.LibraryPermissionIntegrityError as err:
raise ValidationError(detail=str(err))
grant = api.get_library_user_permissions(key, user)
return Response(ContentLibraryPermissionSerializer(grant).data)
@convert_exceptions
def delete(self, request, lib_key_str, user_id):
def get(self, request, lib_key_str, username):
"""
Gets the current permissions settings for a particular user.
"""
key = LibraryLocatorV2.from_string(lib_key_str)
api.require_permission_for_library_key(key, request.user, permissions.CAN_VIEW_THIS_CONTENT_LIBRARY_TEAM)
user = get_object_or_404(User, username=username)
grant = api.get_library_user_permissions(key, user)
if not grant:
raise NotFound
return Response(ContentLibraryPermissionSerializer(grant).data)
@convert_exceptions
def delete(self, request, lib_key_str, username):
"""
Remove the specified user's permission to access or edit this content
library.
"""
key = LibraryLocatorV2.from_string(lib_key_str)
api.require_permission_for_library_key(key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM)
user = get_object_or_404(User, pk=int(user_id))
api.set_library_user_permissions(key, user, access_level=None)
user = get_object_or_404(User, username=username)
try:
api.set_library_user_permissions(key, user, access_level=None)
except api.LibraryPermissionIntegrityError as err:
raise ValidationError(detail=str(err))
return Response({})
@@ -276,14 +325,14 @@ class LibraryTeamGroupView(APIView):
return Response({})
@convert_exceptions
def delete(self, request, lib_key_str, user_id):
def delete(self, request, lib_key_str, username):
"""
Remove the specified user's permission to access or edit this content
library.
"""
key = LibraryLocatorV2.from_string(lib_key_str)
api.require_permission_for_library_key(key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM)
group = get_object_or_404(Group, pk=int(user_id))
group = get_object_or_404(Group, username=username)
api.set_library_group_permissions(key, group, access_level=None)
return Response({})