Keep record of copied courses in the DB

This will only work for instances that send a launch parameters with hte
course history, similarly to the `assignment.copied_from` feature.

lms/services/course.py

@@ -36,6 +36,8 @@ def any_with_setting(self, group, key, value=True) -> bool:
 
     def get_from_launch(self, product, lti_params):
         """Get the course this LTI launch based on the request's params."""
+        historical_course = None
+
         if existing_course := self.get_by_context_id(lti_params["context_id"]):
             # Keep existing `extra` instead of replacing it with the default
             extra = existing_course.extra
@@ -49,11 +51,14 @@ def get_from_launch(self, product, lti_params):
                         )
                     }
                 }
+            # Only make the query for the original course for new courses
+            historical_course = self._get_copied_from_course(lti_params)
 
         return self.upsert_course(
             context_id=lti_params["context_id"],
             name=lti_params["context_title"],
             extra=extra,
+            copied_from=historical_course,
         )
 
     def get_by_context_id(self, context_id, raise_on_missing=False) -> Course | None:
@@ -75,14 +80,22 @@ def get_by_context_id(self, context_id, raise_on_missing=False) -> Course | None
 
         return query.one_or_none()
 
-    def upsert_course(self, context_id, name, extra, settings=None) -> Course:
+    def upsert_course(  # pylint:disable=too-many-arguments
+        self,
+        context_id,
+        name,
+        extra,
+        settings=None,
+        copied_from: Grouping | None = None,
+    ) -> Course:
         """
         Create or update a course based on the provided values.
 
         :param context_id: The course id from LTI params
         :param name: The name of the course
         :param extra: Additional LMS specific values
         :param settings: A dict of settings for the course
+        :param copied_from: A reference to the course this one was copied from
         """
 
         return self._grouping_service.upsert_groupings(
@@ -95,6 +108,7 @@ def upsert_course(self, context_id, name, extra, settings=None) -> Course:
                 }
             ],
             type_=Grouping.Type.COURSE,
+            copied_from=copied_from,
         )[0]
 
     def find_group_set(self, group_set_id=None, name=None, context_id=None):
@@ -156,6 +170,23 @@ def _is_pre_sections(self, context_id):
             )
         )
 
+    def _get_copied_from_course(self, lti_params) -> Course | None:
+        """Return the course that the current one was copied from."""
+
+        history_params = [
+            "custom_Context.id.history",  # LTI 1.3
+        ]
+
+        for param in history_params:
+            if historical_context_id := lti_params.get(param):
+                # History might have a long chain of comma separated
+                # copies of copies, take the most recent one.
+                historical_context_id = historical_context_id.split(",")[0]
+                if historical_course := self.get_by_context_id(historical_context_id):
+                    return historical_course
+
+        return None
+
 
 def course_service_factory(_context, request):
     return CourseService(

lms/services/grouping/service.py

@@ -34,24 +34,27 @@ def upsert_groupings(
         grouping_dicts: list[dict],
         type_: Grouping.Type,
         parent: Grouping | None = None,
+        copied_from: Grouping | None = None,
     ) -> list[Grouping]:
         """
         Upsert a Grouping generating the authority_provided_id based on its parent.
 
         :param grouping_dicts: A list of dicts containing the grouping information
-        :param parent: Parent grouping for all upserted groups
         :param type_: Type of the groupings
+        :param parent: Parent grouping for all upserted groups
+        :param copied_from: Orignal grouping this one was copied from
         """
         if not grouping_dicts:
             return []
 
+        parent_id = None
         if parent:
             if not parent.id:
                 # Make sure we have a PK for the parent before upserting
                 self._db.flush()
             parent_id = parent.id
-        else:
-            parent_id = None
+
+        copied_from_id = copied_from.id if copied_from else None
 
         values = [
             {
@@ -63,6 +66,7 @@ def upsert_groupings(
                 "updated": func.now(),  # pylint:disable=not-callable
                 # From params
                 "parent_id": parent_id,
+                "copied_from_id": copied_from_id,
                 "type": type_,
                 # Things the caller provides
                 "lms_id": grouping["lms_id"],

tests/unit/lms/services/course_test.py

@@ -73,6 +73,7 @@ def test_get_from_launch_when_existing(
             context_id=sentinel.context_id,
             name=sentinel.context_title,
             extra={"existing": "extra"},
+            copied_from=None,
         )
         assert course == upsert_course.return_value
 
@@ -91,7 +92,10 @@ def test_get_from_launch_when_new(
 
         get_by_context_id.assert_called_once_with(sentinel.context_id)
         upsert_course.assert_called_once_with(
-            context_id=sentinel.context_id, name=sentinel.context_title, extra={}
+            context_id=sentinel.context_id,
+            name=sentinel.context_title,
+            extra={},
+            copied_from=None,
         )
         assert course == upsert_course.return_value
 
@@ -115,6 +119,68 @@ def test_get_from_launch_when_new_and_canvas(
             context_id=sentinel.context_id,
             name=sentinel.context_title,
             extra={"canvas": {"custom_canvas_course_id": sentinel.canvas_id}},
+            copied_from=None,
+        )
+        assert course == upsert_course.return_value
+
+    def test_get_from_launch_when_new_and_historical_course_doesnt_exists(
+        self, svc, upsert_course, product, grouping_service
+    ):
+        grouping_service.get_authority_provided_id.side_effect = [
+            "authority_new_context_id",
+            "authority_original_context_id",
+        ]
+
+        course = svc.get_from_launch(
+            product,
+            lti_params={
+                "context_id": "new_context_id",
+                "context_title": sentinel.context_title,
+                "custom_Context.id.history": "original_context_id",
+            },
+        )
+
+        upsert_course.assert_called_once_with(
+            context_id="new_context_id",
+            name=sentinel.context_title,
+            extra={},
+            copied_from=None,
+        )
+        assert course == upsert_course.return_value
+
+    def test_get_from_launch_when_new_and_historical_course_exists(
+        self,
+        svc,
+        upsert_course,
+        product,
+        application_instance,
+        grouping_service,
+    ):
+        grouping_service.get_authority_provided_id.side_effect = [
+            "authority_new_context_id",
+            "authority_original_context_id",
+        ]
+
+        historical_course = factories.Course(
+            application_instance=application_instance,
+            authority_provided_id="authority_original_context_id",
+            lms_id="original_context_id",
+        )
+
+        course = svc.get_from_launch(
+            product,
+            lti_params={
+                "context_id": "new_context_id",
+                "context_title": sentinel.context_title,
+                "custom_Context.id.history": "original_context_id",
+            },
+        )
+
+        upsert_course.assert_called_once_with(
+            context_id="new_context_id",
+            name=sentinel.context_title,
+            extra={},
+            copied_from=historical_course,
         )
         assert course == upsert_course.return_value
 
@@ -136,6 +202,7 @@ def test_upsert_course(self, svc, grouping_service):
                 }
             ],
             type_=Grouping.Type.COURSE,
+            copied_from=None,
         )
 
         assert course == grouping_service.upsert_groupings.return_value[0]
@@ -169,6 +236,7 @@ def test_upsert_course_sets_canvas_sections_enabled_based_on_legacy_rows(
                 )
             ],
             type_=Any(),
+            copied_from=None,
         )
 
     @pytest.mark.usefixtures("course_with_group_sets")