Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: adds new fork migration strategy (#37408)

Browse Source 
Implements the `fork` strategy, allowing the user to create new copies while migrating courses/legacy libraries to v2 libraries.
This commit is contained in:
Rômulo Penido
2025-10-02 21:35:41 -03:00
committed by GitHub
parent df1d91e8ca
commit 1aa862738e
4 changed files with 189 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cms/djangoapps/modulestore_migrator/api.py
View File

@@ -11,7 +11,6 @@ from openedx.core.djangoapps.content_libraries.api import get_library
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.

9
cms/djangoapps/modulestore_migrator/data.py
View File

@@ -70,12 +70,3 @@ class RepeatHandlingStrategy(Enum):
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

17
cms/djangoapps/modulestore_migrator/tasks.py
View File

@@ -132,6 +132,13 @@ class _MigrationContext:
"""
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

190
cms/djangoapps/modulestore_migrator/tests/test_api.py
View File

@@ -15,6 +15,8 @@ from cms.djangoapps.modulestore_migrator.tests.factories import ModulestoreSourc
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 @@ class TestModulestoreMigratorAPI(LibraryTestCase):
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()
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 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()
# 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):
"""