feat: course staff permissions (#161)

Adds data model and permission checking for course staff users.

edx_exams/apps/api/permissions.py

@@ -2,20 +2,26 @@
 from rest_framework.permissions import BasePermission
 
 
-class StaffUserPermissions(BasePermission):
-    """ Permission class to check if user is staff """
+class CourseStaffUserPermissions(BasePermission):
+    """ Permission class to check if user has course staff permissions """
 
     def has_permission(self, request, view):
-        return request.user.is_staff
+        if not request.user.is_authenticated:
+            return False
 
+        if view.kwargs.get('course_id'):
+            return request.user.is_staff or request.user.has_course_staff_permission(view.kwargs['course_id'])
+        # right now we don't have any views that use this
+        return request.user.is_staff  # pragma: no cover
 
-class StaffUserOrReadOnlyPermissions(BasePermission):
+
+class CourseStaffOrReadOnlyPermissions(CourseStaffUserPermissions):
     """
-    Permission class granting write access to staff users and
-    read-only access to authenticated users
+    Permission class granting write access to course staff users
+    and read-only access to other authenticated users
     """
     def has_permission(self, request, view):
-        return request.user.is_staff or (
-            request.user.is_authenticated and
-            request.method == 'GET'
-        )
+        if request.user.is_authenticated and request.method == 'GET':
+            return True
+        else:
+            return super().has_permission(request, view)

edx_exams/apps/api/v1/tests/test_views.py

@@ -17,7 +17,7 @@
 from edx_exams.apps.api.test_utils import ExamsAPITestCase
 from edx_exams.apps.core.exam_types import get_exam_type
 from edx_exams.apps.core.exceptions import ExamAttemptOnPastDueExam, ExamIllegalStatusTransition
-from edx_exams.apps.core.models import CourseExamConfiguration, Exam, ExamAttempt, ProctoringProvider
+from edx_exams.apps.core.models import CourseExamConfiguration, CourseStaffRole, Exam, ExamAttempt, ProctoringProvider
 from edx_exams.apps.core.statuses import ExamAttemptStatus
 from edx_exams.apps.core.test_utils.factories import (
     AssessmentControlResultFactory,
@@ -100,6 +100,14 @@ def test_auth_failures(self):
         random_user = UserFactory()
         self.get_response(random_user, [], 403)
 
+    def test_course_staff_access(self):
+        """
+        Verify course staff can access endpoint
+        """
+        course_staff_user = UserFactory()
+        CourseStaffRole.objects.create(user=course_staff_user, course_id=self.course_id)
+        self.get_response(course_staff_user, [], 200)
+
     def test_exam_empty_exam_list(self):
         """
         Test that exams not included in request are marked as inactive
@@ -342,6 +350,17 @@ def test_patch_auth_failures(self):
         response = self.patch_api(random_user, {})
         self.assertEqual(403, response.status_code)
 
+    def test_course_staff_write_access(self):
+        """
+        Verify course staff have write access
+        """
+        course_staff_user = UserFactory()
+        CourseStaffRole.objects.create(user=course_staff_user, course_id=self.course_id)
+        response = self.patch_api(course_staff_user, {
+            'provider': None,
+        })
+        self.assertEqual(204, response.status_code)
+
     def test_patch_invalid_data(self):
         """
         Assert that endpoint returns 400 if provider is missing
@@ -1083,7 +1102,8 @@ def setUp(self):
         )
 
         self.non_staff_user = UserFactory()
-        self.staff_user = UserFactory(is_staff=True)
+        self.course_staff_user = UserFactory()
+        CourseStaffRole.objects.create(user=self.course_staff_user, course_id=self.exam.course_id)
 
     def delete_api(self, user, attempt_id):
         """
@@ -1181,13 +1201,14 @@ def test_put_staff_update_exam_attempt(self, action, expected_status, mock_updat
 
         mock_update_attempt_status.return_value = attempt.id
 
-        response = self.put_api(self.staff_user, attempt.id, {'action': action})
+        response = self.put_api(self.course_staff_user, attempt.id, {'action': action})
         self.assertEqual(response.status_code, 200)
         mock_update_attempt_status.assert_called_once_with(attempt.id, expected_status)
 
     def test_put_learner_verify(self):
         """
         Test that a learner account cannot verify an attempt
+        but course staff can
         """
         # create exam attempt for user
         attempt = ExamAttemptFactory(
@@ -1198,6 +1219,9 @@ def test_put_learner_verify(self):
         response = self.put_api(self.non_staff_user, attempt.id, {'action': 'verify'})
         self.assertEqual(response.status_code, 400)
 
+        response = self.put_api(self.course_staff_user, attempt.id, {'action': 'verify'})
+        self.assertEqual(response.status_code, 200)
+
     @patch('edx_exams.apps.api.v1.views.update_attempt_status')
     def test_put_exception_raised(self, mock_update_attempt_status):
         """
@@ -1309,7 +1333,7 @@ def test_staff_delete_attempt(self):
             exam=self.exam,
         )
 
-        response = self.delete_api(self.staff_user, attempt.id)
+        response = self.delete_api(self.course_staff_user, attempt.id)
         self.assertEqual(response.status_code, 204)
         with self.assertRaises(ExamAttempt.DoesNotExist):
             attempt.refresh_from_db()
@@ -1318,7 +1342,7 @@ def test_delete_attempt_with_bad_attempt_id(self):
         """
         Test that a bad attempt ID returns 400
         """
-        response = self.delete_api(self.staff_user, 9999999)
+        response = self.delete_api(self.course_staff_user, 9999999)
         self.assertEqual(response.status_code, 400)
 
 
@@ -1329,7 +1353,8 @@ class ExamAttemptListViewTests(ExamsAPITestCase):
     def setUp(self):
         super().setUp()
 
-        CourseExamConfigurationFactory.create()
+        config = CourseExamConfigurationFactory.create()
+        self.course_id = config.course_id
 
         self.exam_1 = ExamFactory.create()
         self.exam_2 = ExamFactory.create()
@@ -1341,19 +1366,41 @@ def get_api(self, exam_id, user=None, page_limit=20):
         user = user or self.user
         headers = self.build_jwt_headers(user)
         url = reverse(
-            'api:v1:instructor-attempts-list'
+            'api:v1:instructor-attempts-list',
+            kwargs={'course_id': self.course_id}
         )
 
         return self.client.get(f'{url}?exam_id={exam_id}&limit={page_limit}', **headers)
 
     def test_requires_staff_user(self):
         """
-        Test that only staff users can access this endpoint
+        Users that are neither staff nor course staff access this endpoint
         """
         non_staff_user = UserFactory.create()
         response = self.get_api(self.exam_1.id, user=non_staff_user)
         self.assertEqual(response.status_code, 403)
 
+    def test_course_staff_access(self):
+        """
+        Course staff can access this endpoint
+        """
+        course_staff_user = UserFactory.create()
+        CourseStaffRole.objects.create(user=course_staff_user, course_id=self.course_id)
+        response = self.get_api(self.exam_1.id, user=course_staff_user)
+        self.assertEqual(response.status_code, 200)
+
+    def test_exam_in_another_course(self):
+        """
+        Users cannot access exams that are not in their course
+        """
+        course_staff_user = UserFactory.create()
+        CourseStaffRole.objects.create(user=course_staff_user, course_id=self.course_id)
+        exam_in_another_course = ExamFactory.create(
+            course_id='course-v1:edx+another+course',
+        )
+        response = self.get_api(exam_in_another_course.id)
+        self.assertEqual(response.status_code, 404)
+
     def test_get_attempt_list_response_data(self):
         """
         Test that a list of attempts includes the expected fields
@@ -1426,17 +1473,24 @@ def test_get_attempt_list_response_pagination(self):
 
         response = self.get_api(self.exam_1.id, page_limit=5)
         next_url = response.data.get('next')
-        self.assertEqual(next_url, 'http://testserver/api/v1/instructor_view/attempts?exam_id=1&limit=5&offset=5')
+        self.assertEqual(
+            next_url,
+            f'http://testserver/api/v1/instructor_view/course_id/{self.course_id}/attempts?exam_id=1&limit=5&offset=5',
+        )
         self.assertEqual(response.data.get('count'), 12)
         self.assertEqual(len(response.data.get('results')), 5)
 
         headers = self.build_jwt_headers(self.user)
         response = self.client.get(next_url, **headers)
 
         next_url = response.data.get('next')
-        self.assertEqual(next_url, 'http://testserver/api/v1/instructor_view/attempts?exam_id=1&limit=5&offset=10')
         self.assertEqual(
-            response.data.get('previous'), 'http://testserver/api/v1/instructor_view/attempts?exam_id=1&limit=5'
+            next_url,
+            f'http://testserver/api/v1/instructor_view/course_id/{self.course_id}/attempts?exam_id=1&limit=5&offset=10',
+        )
+        self.assertEqual(
+            response.data.get('previous'),
+            f'http://testserver/api/v1/instructor_view/course_id/{self.course_id}/attempts?exam_id=1&limit=5',
         )
         self.assertEqual(response.data.get('count'), 12)
         self.assertEqual(len(response.data.get('results')), 5)

edx_exams/apps/api/v1/urls.py

@@ -53,8 +53,8 @@
         LatestExamAttemptView.as_view(),
         name='exams-attempt-latest',
     ),
-    path(
-        'instructor_view/attempts',
+    re_path(
+        fr'instructor_view/course_id/{COURSE_ID_PATTERN}/attempts',
         InstructorAttemptsListView.as_view(),
         name='instructor-attempts-list'
     ),

edx_exams/apps/api/v1/views.py

@@ -15,7 +15,7 @@
 from rest_framework.response import Response
 from token_utils.api import sign_token_for
 
-from edx_exams.apps.api.permissions import StaffUserOrReadOnlyPermissions, StaffUserPermissions
+from edx_exams.apps.api.permissions import CourseStaffOrReadOnlyPermissions, CourseStaffUserPermissions
 from edx_exams.apps.api.serializers import (
     ExamSerializer,
     InstructorViewAttemptSerializer,
@@ -33,6 +33,7 @@
     get_exam_attempt_time_remaining,
     get_exam_attempts,
     get_exam_by_content_id,
+    get_exam_by_id,
     get_provider_by_exam_id,
     is_exam_passed_due,
     update_attempt_status
@@ -82,7 +83,7 @@ class CourseExamsView(ExamsAPIView):
     """
 
     authentication_classes = (JwtAuthentication,)
-    permission_classes = (StaffUserPermissions,)
+    permission_classes = (CourseStaffUserPermissions,)
 
     @classmethod
     def update_exam(cls, exam_object, fields):
@@ -239,7 +240,7 @@ class CourseExamConfigurationsView(ExamsAPIView):
     """
 
     authentication_classes = (JwtAuthentication,)
-    permission_classes = (StaffUserOrReadOnlyPermissions,)
+    permission_classes = (CourseStaffOrReadOnlyPermissions,)
 
     def get(self, request, course_id):
         """
@@ -321,7 +322,7 @@ class ExamAccessTokensView(ExamsAPIView):
     """
 
     authentication_classes = (JwtAuthentication,)
-    permission_classes = (StaffUserOrReadOnlyPermissions,)
+    permission_classes = (CourseStaffOrReadOnlyPermissions,)
 
     @classmethod
     def get_expiration_window(cls, exam_attempt, default_exp_seconds):
@@ -552,7 +553,8 @@ def put(self, request, attempt_id):
             )
 
         action_mapping = {}
-        if request.user.is_staff:
+        course_id = attempt.exam.course_id
+        if request.user.is_staff or request.user.has_course_staff_permission(course_id):
             action_mapping = {
                 'verify': ExamAttemptStatus.verified,
                 'reject': ExamAttemptStatus.rejected,
@@ -569,7 +571,7 @@ def put(self, request, attempt_id):
         else:
             error_msg = (
                 f'user_id={attempt.user.id} attempted to update attempt_id={attempt.id} in '
-                f'course_id={attempt.exam.course_id} but does not have access to it. (action={action})'
+                f'course_id={course_id} but does not have access to it. (action={action})'
             )
             error = {'detail': error_msg}
             return Response(status=status.HTTP_403_FORBIDDEN, data=error)
@@ -612,8 +614,11 @@ def delete(self, request, attempt_id):
                 data={'detail': f'Attempt with attempt_id={attempt_id} does not exist.'}
             )
 
-        # TODO: this staff check will be updated once an instructor role is added
-        if not request.user.is_staff and exam_attempt.user.id != request.user.id:
+        if (
+            exam_attempt.user.id != request.user.id and
+            not request.user.is_staff and
+            not request.user.has_course_staff_permission(exam_attempt.exam.course_id)
+        ):
             error_msg = (
                 f'user_id={exam_attempt.user.id} attempted to delete attempt_id={exam_attempt.id} in '
                 f'course_id={exam_attempt.exam.course_id} but does not have access to it.'
@@ -637,9 +642,9 @@ class InstructorAttemptsListView(ExamsAPIView):
     """
 
     authentication_classes = (JwtAuthentication,)
-    permission_classes = (StaffUserPermissions,)
+    permission_classes = (CourseStaffUserPermissions,)
 
-    def get(self, request):
+    def get(self, request, course_id):
         """
         HTTP GET handler to fetch all exam attempt data for a given exam.
 
@@ -651,6 +656,14 @@ def get(self, request):
         """
         exam_id = request.query_params.get('exam_id', None)
 
+        # permissions are checked at the course level, the requested
+        # exam must be in the course the user has been authorized to access
+        if get_exam_by_id(exam_id).course_id != course_id:
+            return Response(
+                status=status.HTTP_404_NOT_FOUND,
+                data={'detail': 'Exam does not exist in course'}
+            )
+
         # instructor serializer will follow FK relationships to get user and
         # exam fields. This list is potentially large so use
         # select related to avoid n+1 issues.

edx_exams/apps/core/admin.py

@@ -6,7 +6,15 @@
 
 from edx_exams.apps.lti.utils import get_lti_root
 
-from .models import CourseExamConfiguration, Exam, ExamAttempt, ProctoringProvider, User
+from .models import (
+    AssessmentControlResult,
+    CourseExamConfiguration,
+    CourseStaffRole,
+    Exam,
+    ExamAttempt,
+    ProctoringProvider,
+    User
+)
 
 
 class CustomUserAdmin(UserAdmin):
@@ -63,8 +71,34 @@ class CourseExamConfigurationAdmin(admin.ModelAdmin):
     ordering = ('course_id',)
 
 
+class AssessmentControlResultAdmin(admin.ModelAdmin):
+    """ Admin configuration for the AssessmentControlResult model """
+    list_display = ('get_username', 'get_course_id', 'get_exam_name')
+    search_fields = ('user__username', 'course_id', 'exam_name')
+    ordering = ('-modified',)
+
+    def get_username(self, obj):
+        return obj.attempt.user.username
+
+    def get_course_id(self, obj):
+        return obj.attempt.exam.course_id
+
+    def get_exam_name(self, obj):
+        return obj.attempt.exam.exam_name
+
+
+class CourseStaffRoleAdmin(admin.ModelAdmin):
+    """ Admin configuration for the Course Staff Role model """
+    list_display = ('user', 'course_id')
+    list_filter = ('course_id',)
+    search_fields = ('user__username', 'course_id')
+    ordering = ('course_id',)
+
+
 admin.site.register(User, CustomUserAdmin)
 admin.site.register(ProctoringProvider, ProctoringProviderAdmin)
 admin.site.register(Exam, ExamAdmin)
 admin.site.register(ExamAttempt, ExamAttemptAdmin)
 admin.site.register(CourseExamConfiguration, CourseExamConfigurationAdmin)
+admin.site.register(AssessmentControlResult, AssessmentControlResultAdmin)
+admin.site.register(CourseStaffRole, CourseStaffRoleAdmin)