Merge pull request #34305 from openedx/hajorg/au-1847-create-course-r…

…eset

feat: add post endpoint for course reset

lms/djangoapps/support/tests/test_views.py

@@ -54,7 +54,7 @@
 from common.djangoapps.third_party_auth.tests.factories import SAMLProviderConfigFactory
 from common.test.utils import disable_signal
 from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
-from lms.djangoapps.support.models import CourseResetAudit
+from lms.djangoapps.support.models import CourseResetAudit, CourseResetCourseOptIn
 from lms.djangoapps.support.serializers import ProgramEnrollmentSerializer
 from lms.djangoapps.support.tests.factories import CourseResetCourseOptInFactory, CourseResetAuditFactory
 from lms.djangoapps.verify_student.models import VerificationDeadline
@@ -2331,3 +2331,97 @@ def test_multiple_failed_audits(self):
             'can_reset': True,
             'status': most_recent_audit.status_message()
         }])
+
+
+class TestResetCourseViewPost(SupportViewTestCase):
+    """
+    Tests for creating course request
+    """
+
+    def setUp(self):
+        super().setUp()
+        SupportStaffRole().add_users(self.user)
+
+        self.course_id = 'course-v1:a+b+c'
+
+        self.other_user = User.objects.create(username='otheruser', password='test')
+
+        self.course = CourseFactory.create(
+            org='a',
+            course='b',
+            run='c',
+            enable_proctored_exams=True,
+            proctoring_provider=settings.PROCTORING_BACKENDS['DEFAULT'],
+        )
+        self.enrollment = CourseEnrollmentFactory(
+            is_active=True,
+            mode='verified',
+            course_id=self.course.id,
+            user=self.user
+        )
+        self.opt_in = CourseResetCourseOptInFactory.create(course_id=self.course.id)
+
+        self.other_course = CourseFactory.create(
+            org='x',
+            course='y',
+            run='z',
+        )
+
+    def _url(self, username):
+        return reverse("support:course_reset", kwargs={'username_or_email': username})
+
+    def test_wrong_username(self):
+        """
+        Test that a request with a username which does not exits returns 404
+        """
+        response = self.client.post(self._url(username='does_not_exist'), data={'course_id': 'course-v1:aa+bb+c'})
+        self.assertEqual(response.status_code, 404)
+
+    def test_learner_course_reset(self):
+        response = self.client.post(self._url(username=self.user.username), data={'course_id': self.course_id})
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(response.data, {
+            'course_id': self.course_id,
+            'status': response.data['status'],
+            'can_reset': False,
+            'display_name': self.course.display_name
+        })
+
+    def test_course_not_opt_in(self):
+        response = self.client.post(self._url(username=self.user.username), data={'course_id': 'course-v1:aa+bb+c'})
+        self.assertEqual(response.status_code, 404)
+
+    def test_course_reset_failed(self):
+        course = CourseFactory.create(
+            org='xx',
+            course='yy',
+            run='zz',
+        )
+        enrollment = CourseEnrollmentFactory(
+            is_active=True,
+            mode='verified',
+            course_id=course.id,
+            user=self.user
+        )
+
+        opt_in_course = CourseResetCourseOptIn.objects.create(
+            course_id=course.id,
+            active=True
+        )
+
+        CourseResetAudit.objects.create(
+            course=opt_in_course,
+            course_enrollment=enrollment,
+            reset_by=self.other_user,
+            status=CourseResetAudit.CourseResetStatus.FAILED
+        )
+        response = self.client.post(self._url(username=self.user.username), data={'course_id': course.id})
+        self.assertEqual(response.status_code, 200)
+
+    def test_course_reset_dupe(self):
+        CourseResetAuditFactory.create(
+            course=self.opt_in,
+            course_enrollment=self.enrollment,
+        )
+        response2 = self.client.post(self._url(username=self.user.username), data={'course_id': self.course_id})
+        self.assertEqual(response2.status_code, 204)

lms/djangoapps/support/views/course_reset.py

@@ -97,4 +97,69 @@ def get(self, request, username_or_email):
 
     @method_decorator(require_support_permission)
     def post(self, request, username_or_email):
-        """ Other Ticket """
+        """
+        Resets a course for the given learner
+
+        returns a dicts with the format {
+            'course_id': <course id>
+            'display_name': <course display name>
+            'status': <status of the enrollment wrt/reset, to be displayed to user>
+            'can_reset': (boolean) <can the course be reset for this learner>
+        }
+        """
+        course_id = request.data['course_id']
+        try:
+            user = get_user_by_username_or_email(username_or_email)
+        except User.DoesNotExist:
+            return Response({'error': 'User does not exist'}, status=404)
+        try:
+            opt_in_course = CourseResetCourseOptIn.objects.get(course_id=course_id)
+        except CourseResetCourseOptIn.DoesNotExist:
+            return Response({'error': 'Course is not eligible'}, status=404)
+        enrollment = CourseEnrollment.objects.get(
+            course=course_id,
+            user=user,
+            is_active=True
+        )
+        user_passed = user_has_passing_grade_in_course(enrollment=enrollment)
+        course_overview = enrollment.course_overview
+        course_reset_audit = CourseResetAudit.objects.filter(course_enrollment=enrollment).first()
+
+        if course_reset_audit and (
+            course_reset_audit.status == CourseResetAudit.CourseResetStatus.FAILED
+            and not user_passed
+        ):
+            course_reset_audit.status = CourseResetAudit.CourseResetStatus.ENQUEUED
+            course_reset_audit.save()
+            # Call celery task
+            resp = {
+                'course_id': course_id,
+                'status': course_reset_audit.status_message(),
+                'can_reset': False,
+                'display_name': course_overview.display_name
+            }
+            return Response(resp, status=200)
+
+        elif course_reset_audit and course_reset_audit.status in (
+            CourseResetAudit.CourseResetStatus.IN_PROGRESS,
+            CourseResetAudit.CourseResetStatus.ENQUEUED
+        ):
+            return Response(None, status=204)
+
+        if enrollment and opt_in_course and not user_passed:
+            course_reset_audit = CourseResetAudit.objects.create(
+                course=opt_in_course,
+                course_enrollment=enrollment,
+                reset_by=request.user,
+            )
+            resp = {
+                'course_id': course_id,
+                'status': course_reset_audit.status_message(),
+                'can_reset': False,
+                'display_name': course_overview.display_name
+            }
+
+            # Call celery task
+            return Response(resp, status=201)
+        else:
+            return Response(None, status=400)