feat: CourseEnrollmentAllowed API

openedx/core/djangoapps/enrollments/serializers.py

@@ -8,7 +8,8 @@
 from rest_framework import serializers
 
 from common.djangoapps.course_modes.models import CourseMode
-from common.djangoapps.student.models import CourseEnrollment
+from common.djangoapps.student.models import (CourseEnrollment,
+                                              CourseEnrollmentAllowed)
 
 log = logging.getLogger(__name__)
 
@@ -127,3 +128,16 @@ class ModeSerializer(serializers.Serializer):  # pylint: disable=abstract-method
     description = serializers.CharField()
     sku = serializers.CharField()
     bulk_sku = serializers.CharField()
+
+
+class CourseEnrollmentAllowedSerializer(serializers.ModelSerializer):
+    """
+    Serializes CourseEnrollmentAllowed model
+
+    Aggregates all data from the CourseEnrollmentAllowed table, and pulls in the serialization
+    to give a complete representation of course enrollment allowed.
+    """
+    class Meta:
+        model = CourseEnrollmentAllowed
+        exclude = ['id']
+        lookup_field = 'user'

openedx/core/djangoapps/enrollments/tests/test_views.py

@@ -1828,3 +1828,105 @@ def test_response_valid_queries(self, args):
         results = content['results']
 
         self.assertCountEqual(results, expected_results)
+
+
+@ddt.ddt
+@skip_unless_lms
+class EnrollmentAllowedViewTest(APITestCase):
+    """
+    Test the view that allows the retrieval and creation of enrollment
+    allowed for a given user email and course id.
+    """
+
+    def setUp(self):
+        self.url = reverse('courseenrollmentallowed')
+        self.staff_user = AdminFactory(
+            username='staff',
+            email='staff@example.com',
+            password='edx'
+        )
+        self.student1 = UserFactory(
+            username='student1',
+            email='student1@example.com',
+            password='edx'
+        )
+        self.data = {
+            'email': 'new-student@example.com',
+            'course_id': 'course-v1:edX+DemoX+Demo_Course'
+        }
+        self.staff_token = create_jwt_for_user(self.staff_user)
+        self.student_token = create_jwt_for_user(self.student1)
+        self.client.credentials(HTTP_AUTHORIZATION='JWT ' + self.staff_token)
+        return super().setUp()
+
+    @ddt.data(
+        [{'email': 'new-student@example.com', 'course_id': 'course-v1:edX+DemoX+Demo_Course'}, status.HTTP_201_CREATED],
+        [{'course_id': 'course-v1:edX+DemoX+Demo_Course'}, status.HTTP_400_BAD_REQUEST],
+        [{'email': 'new-student@example.com'}, status.HTTP_400_BAD_REQUEST],
+    )
+    @ddt.unpack
+    def test_post_enrollment_allowed(self, data, expected_result):
+        """
+        Expected results:
+        - 201: If the request has email and course_id.
+        - 400: If the request has not.
+        """
+        response = self.client.post(self.url, data)
+        assert response.status_code == expected_result
+
+    def test_post_enrollment_allowed_without_staff(self):
+        """
+        Expected result:
+        - 403: Get when I am not staff.
+        """
+        self.client.credentials(HTTP_AUTHORIZATION='JWT ' + self.student_token)
+        response = self.client.post(self.url, self.data)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_get_enrollment_allowed_empty(self):
+        """
+        Expected result:
+        - Get the enrollment allowed from the request.user.
+        """
+        response = self.client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_get_enrollment_allowed(self):
+        """
+        Expected result:
+        - Get the course enrollment allows.
+        """
+        response = self.client.post(path=self.url, data=self.data)
+        response = self.client.get(self.url, {"email": "new-student@example.com"})
+        self.assertContains(response, 'new-student@example.com', status_code=status.HTTP_200_OK)
+
+    def test_get_enrollment_allowed_without_staff(self):
+        """
+        Expected result:
+        - 403: Get when I am not staff.
+        """
+        self.client.credentials(HTTP_AUTHORIZATION='JWT ' + self.student_token)
+        response = self.client.get(self.url, {"email": "new-student@example.com"})
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    @ddt.data(
+        [{'email': 'new-student@example.com',
+          'course_id': 'course-v1:edX+DemoX+Demo_Course'},
+         status.HTTP_204_NO_CONTENT],
+        [{'email': 'other-student@example.com',
+          'course_id': 'course-v1:edX+DemoX+Demo_Course'},
+         status.HTTP_404_NOT_FOUND],
+        [{'course_id': 'course-v1:edX+DemoX+Demo_Course'},
+         status.HTTP_400_BAD_REQUEST],
+    )
+    @ddt.unpack
+    def test_delete_enrollment_allowed(self, delete_data, expected_result):
+        """
+        Expected results:
+        - 204: Enrollment allowed deleted.
+        - 404: Not found, the course enrollment allowed doesn't exists.
+        - 400: Bad request, missing data.
+        """
+        self.client.post(self.url, self.data)
+        response = self.client.delete(self.url, delete_data)
+        assert response.status_code == expected_result

openedx/core/djangoapps/enrollments/urls.py

@@ -9,6 +9,7 @@
 
 from .views import (
     CourseEnrollmentsApiListView,
+    EnrollmentAllowedView,
     EnrollmentCourseDetailView,
     EnrollmentListView,
     EnrollmentUserRolesView,
@@ -29,4 +30,5 @@
             EnrollmentCourseDetailView.as_view(), name='courseenrollmentdetails'),
     path('unenroll/', UnenrollmentView.as_view(), name='unenrollment'),
     path('roles/', EnrollmentUserRolesView.as_view(), name='roles'),
+    path('enrollment_allowed/', EnrollmentAllowedView.as_view(), name='courseenrollmentallowed'),
 ]

openedx/core/djangoapps/enrollments/views.py

@@ -11,6 +11,7 @@
     ObjectDoesNotExist,
     ValidationError
 )
+from django.db import IntegrityError    # lint-amnesty, pylint: disable=wrong-import-order
 from django.utils.decorators import method_decorator  # lint-amnesty, pylint: disable=wrong-import-order
 from edx_rest_framework_extensions.auth.jwt.authentication import \
     JwtAuthentication  # lint-amnesty, pylint: disable=wrong-import-order
@@ -26,7 +27,7 @@
 
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.auth import user_has_role
-from common.djangoapps.student.models import CourseEnrollment, User
+from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed, User
 from common.djangoapps.student.roles import CourseStaffRole, GlobalStaff
 from common.djangoapps.util.disable_rate_limit import can_disable_rate_limit
 from openedx.core.djangoapps.cors_csrf.authentication import SessionAuthenticationCrossDomainCsrf
@@ -41,7 +42,10 @@
 )
 from openedx.core.djangoapps.enrollments.forms import CourseEnrollmentsApiListForm
 from openedx.core.djangoapps.enrollments.paginators import CourseEnrollmentsApiListPagination
-from openedx.core.djangoapps.enrollments.serializers import CourseEnrollmentsApiListSerializer
+from openedx.core.djangoapps.enrollments.serializers import (
+    CourseEnrollmentAllowedSerializer,
+    CourseEnrollmentsApiListSerializer
+)
 from openedx.core.djangoapps.user_api.accounts.permissions import CanRetireUser
 from openedx.core.djangoapps.user_api.models import UserRetirementStatus
 from openedx.core.djangoapps.user_api.preferences.api import update_email_opt_in
@@ -987,3 +991,165 @@ def get_queryset(self):
         if usernames:
             queryset = queryset.filter(user__username__in=usernames)
         return queryset
+
+
+class EnrollmentAllowedView(APIView):
+    """
+    A view that allows the retrieval and creation of enrollment allowed for a given user email and course id.
+    """
+    authentication_classes = (
+        JwtAuthentication,
+    )
+    permission_classes = (permissions.IsAdminUser,)
+    throttle_classes = (EnrollmentUserThrottle,)
+    serializer_class = CourseEnrollmentAllowedSerializer
+
+    def get(self, request):
+        """
+        Returns the enrollments allowed for a given user email.
+
+        **Example Requests**
+
+        GET /api/enrollment/v1/enrollment_allowed?email=user@example.com
+
+        **Parameters**
+
+        - `email` (optional, string, _query_params_) - defaults to the calling user if not provided.
+
+        **Responses**
+        - 200: Success.
+        - 403: Forbidden, you need to be staff.
+        """
+        user_email = request.query_params.get('email')
+        if not user_email:
+            user_email = request.user.email
+
+        enrollments_allowed = CourseEnrollmentAllowed.objects.filter(email=user_email) or []
+        serialized_enrollments_allowed = [
+            CourseEnrollmentAllowedSerializer(enrollment).data for enrollment in enrollments_allowed
+        ]
+
+        return Response(
+            status=status.HTTP_200_OK,
+            data=serialized_enrollments_allowed
+        )
+
+    def post(self, request):
+        """
+        Creates an enrollment allowed for a given user email and course id.
+
+        **Example Request**
+
+        POST /api/enrollment/v1/enrollment_allowed
+
+        Example request data:
+        ```
+        {
+            "email": "user@example.com",
+            "course_id": "course-v1:edX+DemoX+Demo_Course",
+            "auto_enroll": true
+        }
+        ```
+
+        **Parameters**
+
+        - `email` (**required**, string, _body_)
+
+        - `course_id` (**required**, string, _body_)
+
+        - `auto_enroll` (optional, bool: default=false, _body_)
+
+        **Responses**
+        - 200: Success, enrollment allowed found.
+        - 400: Bad request, missing data.
+        - 403: Forbidden, you need to be staff.
+        - 409: Conflict, enrollment allowed already exists.
+        """
+        is_bad_request_response, email, course_id = self.check_required_data(request)
+        auto_enroll = request.data.get('auto_enroll', False)
+        if is_bad_request_response:
+            return is_bad_request_response
+
+        try:
+            enrollment_allowed = CourseEnrollmentAllowed.objects.create(
+                email=email,
+                course_id=course_id,
+                auto_enroll=auto_enroll
+            )
+        except IntegrityError:
+            return Response(
+                status=status.HTTP_409_CONFLICT,
+                data={
+                    'message': f'An enrollment allowed with email {email} and course {course_id} already exists.'
+                }
+            )
+
+        serializer = CourseEnrollmentAllowedSerializer(enrollment_allowed)
+        return Response(
+            status=status.HTTP_201_CREATED,
+            data=serializer.data
+        )
+
+    def delete(self, request):
+        """
+        Deletes an enrollment allowed for a given user email and course id.
+
+        **Example Request**
+
+        DELETE /api/enrollment/v1/enrollment_allowed
+
+        Example request data:
+        ```
+        {
+            "email": "user@example.com",
+            "course_id": "course-v1:edX+DemoX+Demo_Course"
+        }
+        ```
+
+        **Parameters**
+
+        - `email` (**required**, string, _body_)
+
+        - `course_id` (**required**, string, _body_)
+
+        **Responses**
+        - 204: Enrollment allowed deleted.
+        - 400: Bad request, missing data.
+        - 403: Forbidden, you need to be staff.
+        - 404: Not found, the course enrollment allowed doesn't exists.
+        """
+        is_bad_request_response, email, course_id = self.check_required_data(request)
+        if is_bad_request_response:
+            return is_bad_request_response
+
+        try:
+            CourseEnrollmentAllowed.objects.get(
+                email=email,
+                course_id=course_id
+            ).delete()
+            return Response(
+                status=status.HTTP_204_NO_CONTENT,
+            )
+        except ObjectDoesNotExist:
+            return Response(
+                status=status.HTTP_404_NOT_FOUND,
+                data={
+                    'message': f"An enrollment allowed with email {email} and course {course_id} doesn't exists."
+                }
+            )
+
+    def check_required_data(self, request):
+        """
+        Check if the request has email and course_id.
+        """
+        email = request.data.get('email')
+        course_id = request.data.get('course_id')
+        if not email or not course_id:
+            is_bad_request = Response(
+                status=status.HTTP_400_BAD_REQUEST,
+                data={
+                    "message": "Please provide a value for 'email' and 'course_id' in the request data."
+                })
+        else:
+            is_bad_request = None
+        return (is_bad_request, email, course_id)