mattdrayer/api-moderator-role-valiation: Confirm add/remove of role

Author: Matt Drayer

lms/djangoapps/api_manager/courses/tests.py

@@ -11,6 +11,7 @@
 
 from django.contrib.auth.models import Group
 from django.core.cache import cache
+from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase, Client
 from django.test.utils import override_settings
 
@@ -1476,7 +1477,6 @@ def test_coursemodulecompletions_filters(self):
 
     def test_coursemodulecompletions_get_invalid_course(self):
         completion_uri = '{}/{}/completions/'.format(self.base_courses_uri, self.test_bogus_course_id)
-        print completion_uri
         response = self.do_get(completion_uri)
         self.assertEqual(response.status_code, 404)
 
@@ -1796,11 +1796,23 @@ def test_courses_roles_list_get(self):
 
         # filter roleset by user
         user_id = {'user_id': '{}'.format(self.users[0].id)}
-        test_uri = '{}?{}'.format(test_uri, urlencode(user_id))
-        response = self.do_get(test_uri)
+        user_filter_uri = '{}?{}'.format(test_uri, urlencode(user_id))
+        response = self.do_get(user_filter_uri)
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 1)
 
+        # filter roleset by role
+        role = {'role': 'instructor'}
+        role_filter_uri = '{}?{}'.format(test_uri, urlencode(role))
+        response = self.do_get(role_filter_uri)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        role = {'role': 'invalid_role'}
+        role_filter_uri = '{}?{}'.format(test_uri, urlencode(role))
+        response = self.do_get(role_filter_uri)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 0)
+
     def test_courses_roles_list_get_invalid_course(self):
         test_uri = '/api/courses/{}/roles/'.format(self.test_bogus_course_id)
         response = self.do_get(test_uri)
@@ -1819,6 +1831,11 @@ def test_courses_roles_list_post(self):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 1)
 
+        # Confirm this user also has forum moderation permissions
+        role = Role.objects.get(course_id=self.course.id, name=FORUM_ROLE_MODERATOR)
+        has_role = role.users.get(id=self.users[0].id)
+        self.assertTrue(has_role)
+
     def test_courses_roles_list_post_invalid_course(self):
         test_uri = '/api/courses/{}/roles/'.format(self.test_bogus_course_id)
         data = {'user_id': self.users[0].id, 'role': 'instructor'}
@@ -1844,17 +1861,23 @@ def test_courses_roles_users_detail_delete(self):
         self.assertEqual(response.status_code, 201)
 
         response = self.do_get(test_uri)
-        print response.data
         self.assertEqual(len(response.data), 1)
 
         delete_uri = '{}instructor/users/{}'.format(test_uri, self.users[0].id)
-        print delete_uri
         response = self.do_delete(delete_uri)
         self.assertEqual(response.status_code, 204)
 
         response = self.do_get(test_uri)
         self.assertEqual(len(response.data), 0)
 
+        # Confirm this user no longer has forum moderation permissions
+        role = Role.objects.get(course_id=self.course.id, name=FORUM_ROLE_MODERATOR)
+        try:
+            has_role = role.users.get(id=self.users[0].id)
+            self.assertTrue(False)
+        except ObjectDoesNotExist:
+            pass
+
     def test_courses_roles_users_detail_delete_invalid_course(self):
         test_uri = '/api/courses/{}/roles/'.format(self.test_bogus_course_id)
         delete_uri = '{}instructor/users/{}'.format(test_uri, self.users[0].id)
@@ -1870,6 +1893,5 @@ def test_courses_roles_users_detail_delete_invalid_user(self):
     def test_courses_roles_users_detail_delete_invalid_role(self):
         test_uri = '/api/courses/{}/roles/'.format(unicode(self.course.id))
         delete_uri = '{}invalid_role/users/{}'.format(test_uri, self.users[0].id)
-        print delete_uri
         response = self.do_delete(delete_uri)
         self.assertEqual(response.status_code, 404)

lms/djangoapps/api_manager/courses/views.py

@@ -6,7 +6,6 @@
 from lxml import etree
 from StringIO import StringIO
 
-
 from django.conf import settings
 from django.contrib.auth.models import Group, User
 from django.core.exceptions import ObjectDoesNotExist
@@ -22,9 +21,9 @@
 from courseware.models import StudentModule
 from courseware.views import get_static_tab_contents
 from django_comment_common.models import FORUM_ROLE_MODERATOR
-from instructor.access import allow_access, revoke_access, update_forum_role
+from instructor.access import revoke_access, update_forum_role
 from student.models import CourseEnrollment, CourseEnrollmentAllowed
-from student.roles import CourseInstructorRole, CourseStaffRole, CourseObserverRole, UserBasedRole
+from student.roles import CourseAccessRole, CourseInstructorRole, CourseStaffRole, CourseObserverRole, UserBasedRole
 
 from xmodule.modulestore.django import modulestore
 
@@ -279,24 +278,30 @@ def _manage_role(course_descriptor, user, role, action):
     """
     Helper method for managing course/forum roles
     """
+    supported_roles = ('instructor', 'staff', 'observer')
     forum_moderator_roles = ('instructor', 'staff')
+    if role not in supported_roles:
+        raise ValueError
     if action is 'allow':
-        allow_access(course_descriptor, user, role)
+        existing_role = CourseAccessRole.objects.filter(user=user, role=role, course_id=course_descriptor.id, org=course_descriptor.org)
+        if not existing_role:
+            new_role = CourseAccessRole(user=user, role=role, course_id=course_descriptor.id, org=course_descriptor.org)
+            new_role.save()
         if role in forum_moderator_roles:
             update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'allow')
     elif action is 'revoke':
+        revoke_access(course_descriptor, user, role)
         if role in forum_moderator_roles:
             # There's a possibilty that the user may play more than one role in a course
             # And that more than one of these roles allow for forum moderation
-            # So we need to confirm the current role is the only one for this user for this course
+            # So we need to confirm the removed role was the only role for this user for this course
             # Before we can safely remove the corresponding forum moderator role
             user_instructor_courses = UserBasedRole(user, CourseInstructorRole.ROLE).courses_with_role()
             user_staff_courses = UserBasedRole(user, CourseStaffRole.ROLE).courses_with_role()
             queryset = user_instructor_courses | user_staff_courses
             queryset = queryset.filter(course_id=course_descriptor.id)
-            if len(queryset) <= 1:
-                update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'allow')
-        revoke_access(course_descriptor, user, role)
+            if len(queryset) == 0:
+                update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'revoke')
 
 
 class CourseContentList(SecureAPIView):
@@ -1329,7 +1334,6 @@ def get_queryset(self):
         stage = self.request.QUERY_PARAMS.get('stage', None)
         course_id = self.kwargs['course_id']
         course_descriptor, course_key, course_content = get_course(self.request, self.request.user, course_id)  # pylint: disable=W0612
-        print course_descriptor
         if not course_descriptor:
             raise Http404
         queryset = CourseModuleCompletion.objects.filter(course_id=course_key)
@@ -1712,6 +1716,10 @@ def get(self, request, course_id):  # pylint: disable=W0613
         if user_id:
             response_data = list([item for item in response_data if int(item['id']) == int(user_id)])
 
+        role = self.request.QUERY_PARAMS.get('role', None)
+        if role:
+            response_data = list([item for item in response_data if item['role'] == role])
+
         return Response(response_data, status=status.HTTP_200_OK)
 
     def post(self, request, course_id):

lms/djangoapps/api_manager/groups/tests.py

@@ -894,7 +894,6 @@ def test_group_courses_detail_get(self):
         response = self.do_post(test_uri, data)
         self.assertEqual(response.status_code, 201)
         test_uri = '{}/{}/courses/{}'.format(self.base_groups_uri, group_id, self.test_course_id)
-        print test_uri
         response = self.do_get(test_uri)
         self.assertEqual(response.status_code, 200)
         confirm_uri = '{}{}/{}/courses/{}'.format(

lms/djangoapps/api_manager/users/tests.py

@@ -12,10 +12,11 @@
 from urllib import urlencode
 from mock import patch
 
-from django.utils.translation import ugettext as _
 from django.core.cache import cache
+from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase, Client
 from django.test.utils import override_settings
+from django.utils.translation import ugettext as _
 
 from capa.tests.response_xml_factory import StringResponseXMLFactory
 from courseware.tests.factories import StudentModuleFactory
@@ -1387,11 +1388,23 @@ def test_users_roles_list_get(self):
 
         # filter roleset by course
         course_id = {'course_id': '{}'.format(unicode(course3.id))}
-        test_uri = '{}?{}'.format(test_uri, urlencode(course_id))
-        response = self.do_get(test_uri)
+        course_filter_uri = '{}?{}'.format(test_uri, urlencode(course_id))
+        response = self.do_get(course_filter_uri)
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data['count'], 1)
 
+        # filter roleset by role
+        role = {'role': 'instructor'}
+        role_filter_uri = '{}?{}'.format(test_uri, urlencode(role))
+        response = self.do_get(role_filter_uri)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data['count'], 1)
+        role = {'role': 'invalid_role'}
+        role_filter_uri = '{}?{}'.format(test_uri, urlencode(role))
+        response = self.do_get(role_filter_uri)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data['count'], 0)
+
     def test_users_roles_list_get_invalid_user(self):
         test_uri = '/api/users/23423/roles/'
         response = self.do_get(test_uri)
@@ -1417,6 +1430,11 @@ def test_users_roles_list_post(self):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data['count'], 1)
 
+        # Confirm this user also has forum moderation permissions
+        role = Role.objects.get(course_id=self.course.id, name=FORUM_ROLE_MODERATOR)
+        has_role = role.users.get(id=self.user.id)
+        self.assertTrue(has_role)
+
     def test_users_roles_list_post_invalid_user(self):
         test_uri = '/api/users/2131/roles/'
         data = {'course_id': unicode(self.course.id), 'role': 'instructor'}
@@ -1527,6 +1545,14 @@ def test_users_roles_courses_detail_delete(self):
         response = self.do_get(test_uri)
         self.assertEqual(response.data['count'], 0)
 
+        # Confirm this user no longer has forum moderation permissions
+        role = Role.objects.get(course_id=self.course.id, name=FORUM_ROLE_MODERATOR)
+        try:
+            has_role = role.users.get(id=self.user.id)
+            self.assertTrue(False)
+        except ObjectDoesNotExist:
+            pass
+
     def test_users_roles_courses_detail_delete_invalid_course(self):
         test_uri = '/api/users/{}/roles/'.format(self.user.id)
         delete_uri = '{}instructor/courses/{}'.format(test_uri, self.test_bogus_course_id)

lms/djangoapps/api_manager/users/views.py

@@ -20,12 +20,12 @@
 from courseware.models import StudentModule
 from courseware.views import get_module_for_descriptor, save_child_position, get_current_child
 from django_comment_common.models import FORUM_ROLE_MODERATOR
-from instructor.access import allow_access, revoke_access, update_forum_role
+from instructor.access import revoke_access, update_forum_role
 from lang_pref import LANGUAGE_KEY
 from lms.lib.comment_client.user import User as CommentUser
 from lms.lib.comment_client.utils import CommentClientRequestError
 from student.models import CourseEnrollment, PasswordHistory, UserProfile
-from student.roles import CourseInstructorRole, CourseObserverRole, CourseStaffRole, UserBasedRole
+from student.roles import CourseAccessRole, CourseInstructorRole, CourseObserverRole, CourseStaffRole, UserBasedRole
 from user_api.models import UserPreference
 from util.bad_request_rate_limiter import BadRequestRateLimiter
 from util.password_policy_validators import (
@@ -98,24 +98,30 @@ def _manage_role(course_descriptor, user, role, action):
     """
     Helper method for managing course/forum roles
     """
+    supported_roles = ('instructor', 'staff', 'observer')
     forum_moderator_roles = ('instructor', 'staff')
+    if role not in supported_roles:
+        raise ValueError
     if action is 'allow':
-        allow_access(course_descriptor, user, role)
+        existing_role = CourseAccessRole.objects.filter(user=user, role=role, course_id=course_descriptor.id, org=course_descriptor.org)
+        if not existing_role:
+            new_role = CourseAccessRole(user=user, role=role, course_id=course_descriptor.id, org=course_descriptor.org)
+            new_role.save()
         if role in forum_moderator_roles:
             update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'allow')
     elif action is 'revoke':
+        revoke_access(course_descriptor, user, role)
         if role in forum_moderator_roles:
             # There's a possibilty that the user may play more than one role in a course
             # And that more than one of these roles allow for forum moderation
-            # So we need to confirm the current role is the only one for this user for this course
+            # So we need to confirm the removed role was the only role for this user for this course
             # Before we can safely remove the corresponding forum moderator role
             user_instructor_courses = UserBasedRole(user, CourseInstructorRole.ROLE).courses_with_role()
             user_staff_courses = UserBasedRole(user, CourseStaffRole.ROLE).courses_with_role()
             queryset = user_instructor_courses | user_staff_courses
             queryset = queryset.filter(course_id=course_descriptor.id)
-            if len(queryset) == 1:
-                update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'allow')
-        revoke_access(course_descriptor, user, role)
+            if len(queryset) == 0:
+                update_forum_role(course_descriptor.id, user, FORUM_ROLE_MODERATOR, 'revoke')
 
 
 class UsersList(SecureListAPIView):
@@ -1140,6 +1146,10 @@ def get_queryset(self):
                 raise Http404
             queryset = queryset.filter(course_id=course_key)
 
+        role = self.request.QUERY_PARAMS.get('role', None)
+        if role:
+            queryset = queryset.filter(role=role)
+
         return queryset
 
     def post(self, request, user_id):
@@ -1217,9 +1227,7 @@ def delete(self, request, user_id, role, course_id):  # pylint: disable=W0613
             return Response({}, status=status.HTTP_404_NOT_FOUND)
 
         try:
-            revoke_access(course_descriptor, user, role)
-            if role in ('instructor', 'staff'):
-                update_forum_role(course_key, user, FORUM_ROLE_MODERATOR, 'revoke')
+            _manage_role(course_descriptor, user, role, 'revoke')
         except ValueError:
             return Response({}, status=status.HTTP_404_NOT_FOUND)
 

lms/djangoapps/projects/tests/test_workgroups.py

@@ -172,7 +172,7 @@ def test_workgroups_list_post(self):
             response.data['id'],
             self.test_workgroup_name
         )
-        cohort = get_cohort_by_name(self.test_project.course_id, cohort_name, CourseUserGroup.WORKGROUP)
+        cohort = get_cohort_by_name(self.test_course.id, cohort_name, CourseUserGroup.WORKGROUP)
         self.assertIsNotNone(cohort)
 
     def test_workgroups_detail_get(self):
@@ -266,7 +266,7 @@ def test_workgroups_users_post(self):
             response.data['id'],
             self.test_workgroup_name
         )
-        cohort = get_cohort_by_name(self.test_project.course_id, cohort_name, CourseUserGroup.WORKGROUP)
+        cohort = get_cohort_by_name(self.test_course.id, cohort_name, CourseUserGroup.WORKGROUP)
         self.assertIsNotNone(cohort)
         self.assertTrue(is_user_in_cohort(cohort, self.test_user.id, CourseUserGroup.WORKGROUP))
 
@@ -296,15 +296,15 @@ def test_workgroups_users_post_with_cohort_backfill(self):
         )
 
         # now let's remove existing cohort users
-        cohort = get_cohort_by_name(self.test_project.course_id, cohort_name, CourseUserGroup.WORKGROUP)
+        cohort = get_cohort_by_name(self.test_course.id, cohort_name, CourseUserGroup.WORKGROUP)
         self.assertTrue(is_user_in_cohort(cohort, self.test_user.id, CourseUserGroup.WORKGROUP))
 
         remove_user_from_cohort(cohort, self.test_user.username, CourseUserGroup.WORKGROUP)
         self.assertFalse(is_user_in_cohort(cohort, self.test_user.id, CourseUserGroup.WORKGROUP))
 
         # delete cohort
-        delete_empty_cohort(self.test_project.course_id, cohort_name, CourseUserGroup.WORKGROUP)
-        self.assertEqual(0, len(get_course_cohort_names(self.test_project.course_id, CourseUserGroup.WORKGROUP)))
+        delete_empty_cohort(self.test_course.id, cohort_name, CourseUserGroup.WORKGROUP)
+        self.assertEqual(0, len(get_course_cohort_names(self.test_course.id, CourseUserGroup.WORKGROUP)))
 
         # add a 2nd user and make sure a discussion cohort was created and users were backfilled
         test_uri = '{}{}/'.format(self.test_workgroups_uri, str(response.data['id']))
@@ -314,7 +314,7 @@ def test_workgroups_users_post_with_cohort_backfill(self):
         self.assertEqual(response.status_code, 201)
 
         # now inspect cohort and assert that things are as we anticipate (i.e. both users are in there)
-        cohort = get_cohort_by_name(self.test_project.course_id, cohort_name, CourseUserGroup.WORKGROUP)
+        cohort = get_cohort_by_name(self.test_course.id, cohort_name, CourseUserGroup.WORKGROUP)
         self.assertIsNotNone(cohort)
         self.assertTrue(is_user_in_cohort(cohort, self.test_user.id, CourseUserGroup.WORKGROUP))
         self.assertTrue(is_user_in_cohort(cohort, self.test_user2.id, CourseUserGroup.WORKGROUP))

lms/djangoapps/projects/views.py

@@ -177,28 +177,20 @@ def users(self, request, pk):
                 return Response({"detail": message}, status.HTTP_400_BAD_REQUEST)
             workgroup = self.get_object()
             workgroup.users.add(user)
-
-            # add user to the workgroup cohort
-            course_descriptor, course_key, course_content = _get_course(self.request, user, workgroup.project.course_id)  # pylint: disable=W0612
-            cohort = get_cohort_by_name(course_key,
-                                        workgroup.cohort_name, CourseUserGroup.WORKGROUP)
-            add_user_to_cohort(cohort, user.username)
-
             workgroup.save()
 
             # add user to the workgroup cohort, create it if it doesn't exist (for cases where there is a legacy
             # workgroup)
+            course_descriptor, course_key, course_content = _get_course(self.request, user, workgroup.project.course_id)  # pylint: disable=W0612
             try:
-                cohort = get_cohort_by_name(workgroup.project.course_id,
-                                            workgroup.cohort_name, CourseUserGroup.WORKGROUP)
+                cohort = get_cohort_by_name(course_key, workgroup.cohort_name, CourseUserGroup.WORKGROUP)
                 add_user_to_cohort(cohort, user.username)
             except ObjectDoesNotExist:
                 # This use case handles cases where a workgroup might have been created before
                 # the notion of a cohorted discussion. So we need to backfill in the data
-                cohort = add_cohort(workgroup.project.course_id, workgroup.cohort_name, CourseUserGroup.WORKGROUP)
+                cohort = add_cohort(course_key, workgroup.cohort_name, CourseUserGroup.WORKGROUP)
                 for workgroup_user in workgroup.users.all():
                     add_user_to_cohort(cohort, workgroup_user.username)
-
             return Response({}, status=status.HTTP_201_CREATED)
         else:
             user_id = request.DATA.get('id')