feat: add new endpoint for cloning course (#31794)

Co-authored-by: Maxim Beder <maxim@opencraft.com>

cms/djangoapps/api/v1/serializers/course_runs.py

@@ -5,6 +5,7 @@ from django.contrib.auth import get_user_model
 from django.db import transaction
 from django.utils.translation import gettext_lazy as _
 from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
 from rest_framework import serializers
 from rest_framework.fields import empty
 
@@ -198,8 +199,50 @@ class CourseRunRerunSerializer(CourseRunSerializerCommonFieldsMixin, CourseRunTe
             'display_name': instance.display_name
         }
         fields.update(validated_data)
-        new_course_run_key = rerun_course(user, course_run_key, course_run_key.org, number, run, fields, False)
+        new_course_run_key = rerun_course(
+            user, course_run_key, course_run_key.org, number, run, fields, background=False,
+        )
 
         course_run = get_course_and_check_access(new_course_run_key, user)
         self.update_team(course_run, team)
         return course_run
+
+
+class CourseCloneSerializer(serializers.Serializer):  # lint-amnesty, pylint: disable=abstract-method, missing-class-docstring
+    source_course_id = serializers.CharField()
+    destination_course_id = serializers.CharField()
+
+    def validate(self, attrs):
+        source_course_id = attrs.get('source_course_id')
+        destination_course_id = attrs.get('destination_course_id')
+        store = modulestore()
+        source_key = CourseKey.from_string(source_course_id)
+        dest_key = CourseKey.from_string(destination_course_id)
+
+        # Check if the source course exists
+        if not store.has_course(source_key):
+            raise serializers.ValidationError('Source course does not exist.')
+
+        # Check if the destination course already exists
+        if store.has_course(dest_key):
+            raise serializers.ValidationError('Destination course already exists.')
+        return attrs
+
+    def create(self, validated_data):
+        source_course_id = validated_data.get('source_course_id')
+        destination_course_id = validated_data.get('destination_course_id')
+        user = self.context['request'].user
+        source_course_key = CourseKey.from_string(source_course_id)
+        destination_course_key = CourseKey.from_string(destination_course_id)
+        source_course_run = get_course_and_check_access(source_course_key, user)
+        fields = {
+            'display_name': source_course_run.display_name,
+        }
+
+        destination_course_run_key = rerun_course(
+            user, source_course_key, destination_course_key.org, destination_course_key.course,
+            destination_course_key.run, fields, background=False,
+        )
+
+        destination_course_run = get_course_and_check_access(destination_course_run_key, user)
+        return destination_course_run

cms/djangoapps/api/v1/tests/test_views/test_course_runs.py

@@ -402,3 +402,54 @@ class CourseRunViewSetTests(ModuleStoreTestCase):
         assert response.data == {'non_field_errors': [
             'Invalid key supplied. Ensure there are no special characters in the Course Number.'
         ]}
+
+    def test_clone_course(self):
+        course = CourseFactory()
+        url = reverse('api:v1:course_run-clone')
+        data = {
+            'source_course_id': str(course.id),
+            'destination_course_id': 'course-v1:destination+course+id',
+        }
+        response = self.client.post(url, data, format='json')
+        assert response.status_code == 201
+        self.assertEqual(response.data, {"message": "Course cloned successfully."})
+
+    def test_clone_course_with_missing_source_id(self):
+        url = reverse('api:v1:course_run-clone')
+        data = {
+            'destination_course_id': 'course-v1:destination+course+id',
+        }
+        response = self.client.post(url, data, format='json')
+        assert response.status_code == 400
+        self.assertEqual(response.data, {'source_course_id': ['This field is required.']})
+
+    def test_clone_course_with_missing_dest_id(self):
+        url = reverse('api:v1:course_run-clone')
+        data = {
+            'source_course_id': 'course-v1:source+course+id',
+        }
+        response = self.client.post(url, data, format='json')
+        assert response.status_code == 400
+        self.assertEqual(response.data, {'destination_course_id': ['This field is required.']})
+
+    def test_clone_course_with_nonexistent_source_course(self):
+        url = reverse('api:v1:course_run-clone')
+        data = {
+            'source_course_id': 'course-v1:nonexistent+source+course_id',
+            'destination_course_id': 'course-v1:destination+course+id',
+        }
+        response = self.client.post(url, data, format='json')
+        assert response.status_code == 400
+        assert str(response.data.get('non_field_errors')[0]) == 'Source course does not exist.'
+
+    def test_clone_course_with_existing_dest_course(self):
+        url = reverse('api:v1:course_run-clone')
+        course = CourseFactory()
+        existing_dest_course = CourseFactory()
+        data = {
+            'source_course_id': str(course.id),
+            'destination_course_id': str(existing_dest_course.id),
+        }
+        response = self.client.post(url, data, format='json')
+        assert response.status_code == 400
+        assert str(response.data.get('non_field_errors')[0]) == 'Destination course already exists.'

cms/djangoapps/api/v1/views/course_runs.py

@@ -11,6 +11,7 @@ from rest_framework.response import Response
 from cms.djangoapps.contentstore.views.course import _accessible_courses_iter, get_course_and_check_access
 
 from ..serializers.course_runs import (
+    CourseCloneSerializer,
     CourseRunCreateSerializer,
     CourseRunImageSerializer,
     CourseRunRerunSerializer,
@@ -90,3 +91,49 @@ class CourseRunViewSet(viewsets.GenericViewSet):  # lint-amnesty, pylint: disabl
         new_course_run = serializer.save()
         serializer = self.get_serializer(new_course_run)
         return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    @action(detail=False, methods=['post'])
+    def clone(self, request, *args, **kwargs):  # lint-amnesty, pylint: disable=unused-argument
+        """
+        **Use Case**
+
+            This endpoint can be used for course cloning.
+
+            Unlike reruns, cloning a course allows creating a copy of an existing
+            course under a different organization name and with a different course
+            name.
+
+        **Example Request**
+
+            POST /api/v1/course_runs/clone/ {
+                "source_course_id": "course-v1:edX+DemoX+Demo_Course",
+                "destination_course_id": "course-v1:newOrg+newDemoX+Demo_Course_Clone"
+            }
+
+            **POST Parameters**
+
+                * source_course_id: a full course id of the course that will be
+                  cloned. Has to be an id of an existing course.
+                * destination_course_id: a full course id of the destination
+                  course. The organization, course name and course run of the
+                  new course will be determined from the provided id. Has to be
+                  an id of a course that doesn't exist yet.
+
+        **Response Values**
+
+            If the request parameters are valid and a course has been cloned
+            succesfully, an HTTP 201 "Created" response is returned.
+
+            If source course id and/or destination course id are invalid, or
+            source course doesn't exist, or destination course already exist,
+            an HTTP 400 "Bad Request" response is returned.
+
+            If the user that is making the request doesn't have the access to
+            either of the courses, an HTTP 401 "Unauthorized" response is
+            returned.
+        """
+        serializer = CourseCloneSerializer(data=request.data, context=self.get_serializer_context())
+        serializer.is_valid(raise_exception=True)
+        new_course_run = serializer.save()
+        serializer = self.get_serializer(new_course_run)
+        return Response({"message": "Course cloned successfully."}, status=status.HTTP_201_CREATED)