feat: adds new fork migration strategy [FC-0097]

cms/djangoapps/modulestore_migrator/api.py

@@ -11,7 +11,6 @@
 from openedx.core.types.user import AuthUser
 
 from . import tasks
-from .data import RepeatHandlingStrategy
 from .models import ModulestoreSource
 
 __all__ = (
@@ -35,9 +34,6 @@ def start_migration_to_library(
     """
     Import a course or legacy library into a V2 library (or, a collection within a V2 library).
     """
-    # Can raise NotImplementedError for the Fork strategy
-    assert RepeatHandlingStrategy(repeat_handling_strategy).is_implemented()
-
     source, _ = ModulestoreSource.objects.get_or_create(key=source_key)
     target_library = get_library(target_library_key)
     # get_library ensures that the library is connected to a learning package.

cms/djangoapps/modulestore_migrator/data.py

@@ -70,12 +70,3 @@ def default(cls) -> RepeatHandlingStrategy:
         Returns the default repeat handling strategy.
         """
         return cls.Skip
-
-    def is_implemented(self) -> bool:
-        """
-        Returns True if the repeat handling strategy is implemented.
-        """
-        if self == self.Fork:
-            raise NotImplementedError("Forking is not implemented yet.")
-
-        return True

cms/djangoapps/modulestore_migrator/tasks.py

@@ -132,6 +132,13 @@ def should_update_strategy(self) -> bool:
         """
         return self.repeat_handling_strategy is RepeatHandlingStrategy.Update
 
+    @property
+    def should_fork_strategy(self) -> bool:
+        """
+        Determines whether the repeat handling strategy should fork the entity.
+        """
+        return self.repeat_handling_strategy is RepeatHandlingStrategy.Fork
+
 
 @shared_task(base=_MigrationTask, bind=True)
 # Note: The decorator @set_code_owner_attribute cannot be used here because the UserTaskMixin
@@ -601,8 +608,9 @@ def _get_distinct_target_container_key(
     Returns:
         LibraryContainerLocator: The target container key.
     """
-    # Check if we already processed this block
-    if context.is_already_migrated(source_key):
+    # Check if we already processed this block and we are not forking. If we are forking, we will
+    # want a new target key.
+    if context.is_already_migrated(source_key) and not context.should_fork_strategy:
         existing_version = context.get_existing_target(source_key)
 
         return LibraryContainerLocator(
@@ -646,8 +654,9 @@ def _get_distinct_target_usage_key(
     Raises:
         ValueError: If source_key is invalid
     """
-    # Check if we already processed this block
-    if context.is_already_migrated(source_key):
+    # Check if we already processed this block and we are not forking. If we are forking, we will
+    # want a new target key.
+    if context.is_already_migrated(source_key) and not context.should_fork_strategy:
         log.debug(f"Block {source_key} already exists, reusing existing target")
         existing_target = context.get_existing_target(source_key)
         block_id = existing_target.component.local_key

cms/djangoapps/modulestore_migrator/tests/test_api.py

@@ -15,6 +15,8 @@
 from common.djangoapps.student.tests.factories import UserFactory
 from openedx.core.djangoapps.content_libraries import api as lib_api
 
+from xmodule.modulestore.tests.factories import BlockFactory, LibraryFactory
+
 
 @pytest.mark.django_db
 class TestModulestoreMigratorAPI(LibraryTestCase):
@@ -95,24 +97,184 @@ def test_start_migration_to_library_with_collection(self):
         modulestoremigration = ModulestoreMigration.objects.get()
         assert modulestoremigration.target_collection.key == collection_key
 
-    def test_forking_is_not_implemented(self):
+    def test_start_migration_to_library_with_strategy_skip(self):
         """
-        Test that the API raises NotImplementedError for the Fork strategy.
+        Test that the API can start a migration to a library with a skip strategy.
         """
-        source = ModulestoreSourceFactory()
+        library = LibraryFactory.create(modulestore=self.store)
+        library_block = BlockFactory.create(
+            parent=library,
+            category="html",
+            display_name="Original Block",
+            publish_item=False,
+        )
+        source = ModulestoreSourceFactory(key=library.context_key)
+        user = UserFactory()
+
+        # Start a migration with the Skip strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Skip.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.get()
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Skip.value
+
+        migrated_components = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components) == 1
+
+        # Update the block, changing its name
+        library_block.display_name = "Updated Block"
+        self.store.update_item(library_block, user.id)
+
+        # Migrate again using the Skip strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Skip.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.last()
+        assert modulestoremigration is not None
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Skip.value
+
+        migrated_components_fork = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components_fork) == 1
+
+        component = lib_api.LibraryXBlockMetadata.from_component(
+            self.library_v2.library_key, migrated_components_fork[0]
+        )
+        assert component.display_name == "Original Block"
+
+    def test_start_migration_to_library_with_strategy_update(self):
+        """
+        Test that the API can start a migration to a library with a update strategy.
+        """
+        library = LibraryFactory.create(modulestore=self.store)
+        library_block = BlockFactory.create(
+            parent=library,
+            category="html",
+            display_name="Original Block",
+            publish_item=False,
+        )
+        source = ModulestoreSourceFactory(key=library.context_key)
+        user = UserFactory()
+
+        # Start a migration with the Skip strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Skip.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.get()
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Skip.value
+
+        migrated_components = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components) == 1
+
+        # Update the block, changing its name
+        library_block.display_name = "Updated Block"
+        self.store.update_item(library_block, user.id)
+
+        # Migrate again using the Skip strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Update.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.last()
+        assert modulestoremigration is not None
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Update.value
+
+        migrated_components_fork = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components_fork) == 1
+
+        component = lib_api.LibraryXBlockMetadata.from_component(
+            self.library_v2.library_key, migrated_components_fork[0]
+        )
+        assert component.display_name == "Updated Block"
+
+    def test_start_migration_to_library_with_strategy_forking(self):
+        """
+        Test that the API can start a migration to a library with a forking strategy.
+        """
+        library = LibraryFactory.create(modulestore=self.store)
+        library_block = BlockFactory.create(
+            parent=library,
+            category="html",
+            display_name="Original Block",
+            publish_item=False,
+        )
+        source = ModulestoreSourceFactory(key=library.context_key)
         user = UserFactory()
 
-        with pytest.raises(NotImplementedError):
-            api.start_migration_to_library(
-                user=user,
-                source_key=source.key,
-                target_library_key=self.library_v2.library_key,
-                target_collection_slug=None,
-                composition_level=CompositionLevel.Component.value,
-                repeat_handling_strategy=RepeatHandlingStrategy.Fork.value,
-                preserve_url_slugs=True,
-                forward_source_to_target=False,
-            )
+        # Start a migration with the Skip strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Skip.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.get()
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Skip.value
+
+        migrated_components = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components) == 1
+
+        # Update the block, changing its name
+        library_block.display_name = "Updated Block"
+        self.store.update_item(library_block, user.id)
+
+        # Migrate again using the Fork strategy
+        api.start_migration_to_library(
+            user=user,
+            source_key=source.key,
+            target_library_key=self.library_v2.library_key,
+            composition_level=CompositionLevel.Component.value,
+            repeat_handling_strategy=RepeatHandlingStrategy.Fork.value,
+            preserve_url_slugs=True,
+            forward_source_to_target=False,
+        )
+
+        modulestoremigration = ModulestoreMigration.objects.last()
+        assert modulestoremigration is not None
+        assert modulestoremigration.repeat_handling_strategy == RepeatHandlingStrategy.Fork.value
+
+        migrated_components_fork = lib_api.get_library_components(self.library_v2.library_key)
+        assert len(migrated_components_fork) == 2
+
+        first_component = lib_api.LibraryXBlockMetadata.from_component(
+            self.library_v2.library_key, migrated_components_fork[0]
+        )
+        assert first_component.display_name == "Original Block"
+
+        second_component = lib_api.LibraryXBlockMetadata.from_component(
+            self.library_v2.library_key, migrated_components_fork[1]
+        )
+        assert second_component.display_name == "Updated Block"
 
     def test_get_migration_info(self):
         """