feat: Add unlinkable to xblock actions and update top_level_parent_key on unlink [FC-0097]

cms/djangoapps/contentstore/models.py

@@ -128,6 +128,10 @@ def published_at(self) -> str | None:
     class Meta:
         abstract = True
 
+    @classmethod
+    def get_by_downstream_usage_key(cls, downstream_usage_key: UsageKey):
+        return cls.objects.get(downstream_usage_key=downstream_usage_key)
+
 
 class ComponentLink(EntityLinkBase):
     """
@@ -523,10 +527,6 @@ def update_or_create(
             link.save()
         return link
 
-    @classmethod
-    def get_by_downstream_usage_key(cls, downstream_usage_key: UsageKey):
-        return cls.objects.get(downstream_usage_key=downstream_usage_key)
-
 
 class LearningContextLinksStatusChoices(models.TextChoices):
     """

cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py

@@ -455,16 +455,39 @@ def delete(self, request: _AuthenticatedRequest, usage_key_string: str) -> Respo
         Sever an XBlock's link to upstream content.
         """
         downstream = _load_accessible_block(request.user, usage_key_string, require_write_access=True)
+        affected_blocks: list[XBlock] = []
         try:
-            sever_upstream_link(downstream)
+            # Try to get the upstream key before severing the link, so we can delete
+            # the corresponding ComponentLink or ContainerLink below.
+            try:
+                upstream_key = UpstreamLink.get_for_block(downstream).upstream_key
+            except NoUpstream:
+                # Even if we don't have an UpstreamLink, we still need to check
+                # if the block has the upstream key set, so we don't want to
+                # raise an exception here.
+                upstream_key = None
+
+            affected_blocks = sever_upstream_link(downstream)
+
+            # Remove the ComponentLink or ContainerLink, if it exists.
+            if upstream_key:
+                if isinstance(upstream_key, LibraryUsageLocatorV2):
+                    ComponentLink.get_by_downstream_usage_key(downstream.usage_key).delete()
+                elif isinstance(upstream_key, LibraryContainerLocator):
+                    ContainerLink.get_by_downstream_usage_key(downstream.usage_key).delete()
         except NoUpstream:
             logger.exception(
                 "Tried to DELETE upstream link of '%s', but it wasn't linked to anything in the first place. "
                 "Will do nothing. ",
                 usage_key_string,
             )
-        else:
-            modulestore().update_item(downstream, request.user.id)
+        finally:
+            if affected_blocks:
+                # If we successfully severed the upstream link, then we need to update the affected blocks.
+                with modulestore().bulk_operations(downstream.usage_key.context_key):
+                    for block in affected_blocks:
+                        modulestore().update_item(block, request.user.id)
+
         return Response(status=204)
 
 

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py

@@ -2,29 +2,29 @@
 Unit tests for /api/contentstore/v2/downstreams/* JSON APIs.
 """
 import json
-import ddt
 from datetime import datetime, timezone
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
+import ddt
 from django.conf import settings
 from django.urls import reverse
 from freezegun import freeze_time
+from opaque_keys.edx.keys import ContainerKey, UsageKey
+from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
 from organizations.models import Organization
 
 from cms.djangoapps.contentstore.helpers import StaticFileNotices
-from cms.lib.xblock.upstream_sync import BadUpstream, UpstreamLink
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
 from cms.djangoapps.contentstore.xblock_storage_handlers import view_handlers as xblock_view_handlers
 from cms.djangoapps.contentstore.xblock_storage_handlers.xblock_helpers import get_block_key_dict
-from opaque_keys.edx.keys import ContainerKey, UsageKey
-from opaque_keys.edx.locator import LibraryLocatorV2
-from common.djangoapps.student.tests.factories import UserFactory
+from cms.lib.xblock.upstream_sync import BadUpstream, UpstreamLink
 from common.djangoapps.student.auth import add_users
 from common.djangoapps.student.roles import CourseStaffRole
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.content_libraries import api as lib_api
 from xmodule.modulestore.django import modulestore
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
-from openedx.core.djangoapps.content_libraries import api as lib_api
 
 from .. import downstreams as downstreams_views
 
@@ -42,7 +42,7 @@
 def _get_upstream_link_good_and_syncable(downstream):
     return UpstreamLink(
         upstream_ref=downstream.upstream,
-        upstream_key=UsageKey.from_string(downstream.upstream),
+        upstream_key=LibraryUsageLocatorV2.from_string(downstream.upstream),
         downstream_key=str(downstream.usage_key),
         version_synced=downstream.upstream_version,
         version_available=(downstream.upstream_version or 0) + 1,
@@ -433,6 +433,45 @@ def test_204_no_upstream(self, mock_sever):
         assert response.status_code == 204
         assert mock_sever.call_count == 1
 
+    def test_unlink_parent_should_update_children(self):
+        """
+        If we unlink a parent block, do all children get unlinked as well?
+        """
+        self.client.login(username="superuser", password="password")
+
+        all_downstreams = self.client.get(
+            "/api/contentstore/v2/downstreams/",
+            data={"course_id": str(self.course.id)},
+        )
+        assert all_downstreams.data["count"] == 11
+
+        response = self.call_api(self.top_level_downstream_chapter.usage_key)
+        assert response.status_code == 204
+
+        # Check that all children have their top_level_downstream_parent_key updated
+        subsection = modulestore().get_item(self.top_level_downstream_sequential.usage_key)
+        assert subsection.top_level_downstream_parent_key is None
+
+        unit = modulestore().get_item(self.top_level_downstream_unit_2.usage_key)
+        # The sequential is the top-level parent for the unit
+        assert unit.top_level_downstream_parent_key == {
+            "id": str(self.top_level_downstream_sequential.usage_key.block_id),
+            "type": str(self.top_level_downstream_sequential.usage_key.block_type),
+        }
+
+        video = modulestore().get_item(self.top_level_downstream_video_key)
+        # The sequential is the top-level parent for the video
+        assert video.top_level_downstream_parent_key == {
+            "id": str(self.top_level_downstream_sequential.usage_key.block_id),
+            "type": str(self.top_level_downstream_sequential.usage_key.block_type),
+        }
+
+        all_downstreams = self.client.get(
+            "/api/contentstore/v2/downstreams/",
+            data={"course_id": str(self.course.id)},
+        )
+        assert all_downstreams.data["count"] == 10
+
 
 class _DownstreamSyncViewTestMixin(SharedErrorTestCases):
     """
@@ -562,7 +601,7 @@ class GetUpstreamViewTest(
     SharedModuleStoreTestCase,
 ):
     """
-    Test that `GET /api/v2/contentstore/downstreams-all?...` returns list of links based on the provided filter.
+    Test that `GET /api/v2/contentstore/downstreams?...` returns list of links based on the provided filter.
     """
 
     def call_api(

cms/djangoapps/contentstore/tasks.py

@@ -1654,7 +1654,7 @@ def handle_create_xblock_upstream_link(usage_key):
             # The top-level parent link does not exist yet,
             # it is necessary to create it first.
             handle_create_xblock_upstream_link(str(top_level_parent_usage_key))
-    create_or_update_xblock_upstream_link(xblock, xblock.course_id)
+    create_or_update_xblock_upstream_link(xblock)
 
 
 @shared_task
@@ -1671,7 +1671,7 @@ def handle_update_xblock_upstream_link(usage_key):
         return
     if not xblock.upstream or not xblock.upstream_version:
         return
-    create_or_update_xblock_upstream_link(xblock, xblock.course_id)
+    create_or_update_xblock_upstream_link(xblock)
 
 
 @shared_task
@@ -1711,7 +1711,7 @@ def create_or_update_upstream_links(
         course_status.update_status(LearningContextLinksStatusChoices.FAILED)
         return
     for xblock in xblocks:
-        create_or_update_xblock_upstream_link(xblock, course_key, created)
+        create_or_update_xblock_upstream_link(xblock, created)
     course_status.update_status(LearningContextLinksStatusChoices.COMPLETED)
 
 

cms/djangoapps/contentstore/utils.py

@@ -2385,7 +2385,7 @@ def get_xblock_render_context(request, block):
     return ""
 
 
-def _create_or_update_component_link(course_key: CourseKey, created: datetime | None, xblock):
+def _create_or_update_component_link(created: datetime | None, xblock):
     """
     Create or update upstream->downstream link for components in database for given xblock.
     """
@@ -2399,7 +2399,7 @@ def _create_or_update_component_link(course_key: CourseKey, created: datetime |
     top_level_parent_usage_key = None
     if xblock.top_level_downstream_parent_key is not None:
         top_level_parent_usage_key = BlockUsageLocator(
-            course_key,
+            xblock.usage_key.course_key,
             xblock.top_level_downstream_parent_key.get('type'),
             xblock.top_level_downstream_parent_key.get('id'),
         )
@@ -2408,7 +2408,7 @@ def _create_or_update_component_link(course_key: CourseKey, created: datetime |
         lib_component,
         upstream_usage_key=upstream_usage_key,
         upstream_context_key=str(upstream_usage_key.context_key),
-        downstream_context_key=course_key,
+        downstream_context_key=xblock.usage_key.course_key,
         downstream_usage_key=xblock.usage_key,
         top_level_parent_usage_key=top_level_parent_usage_key,
         version_synced=xblock.upstream_version,
@@ -2417,7 +2417,7 @@ def _create_or_update_component_link(course_key: CourseKey, created: datetime |
     )
 
 
-def _create_or_update_container_link(course_key: CourseKey, created: datetime | None, xblock):
+def _create_or_update_container_link(created: datetime | None, xblock):
     """
     Create or update upstream->downstream link for containers in database for given xblock.
     """
@@ -2431,7 +2431,7 @@ def _create_or_update_container_link(course_key: CourseKey, created: datetime |
     top_level_parent_usage_key = None
     if xblock.top_level_downstream_parent_key is not None:
         top_level_parent_usage_key = BlockUsageLocator(
-            course_key,
+            xblock.usage_key.course_key,
             xblock.top_level_downstream_parent_key.get('type'),
             xblock.top_level_downstream_parent_key.get('id'),
         )
@@ -2440,7 +2440,7 @@ def _create_or_update_container_link(course_key: CourseKey, created: datetime |
         lib_component,
         upstream_container_key=upstream_container_key,
         upstream_context_key=str(upstream_container_key.context_key),
-        downstream_context_key=course_key,
+        downstream_context_key=xblock.usage_key.course_key,
         downstream_usage_key=xblock.usage_key,
         version_synced=xblock.upstream_version,
         top_level_parent_usage_key=top_level_parent_usage_key,
@@ -2449,19 +2449,19 @@ def _create_or_update_container_link(course_key: CourseKey, created: datetime |
     )
 
 
-def create_or_update_xblock_upstream_link(xblock, course_key: CourseKey, created: datetime | None = None) -> None:
+def create_or_update_xblock_upstream_link(xblock, created: datetime | None = None) -> None:
     """
     Create or update upstream->downstream link in database for given xblock.
     """
     if not xblock.upstream:
         return None
     try:
         # Try to create component link
-        _create_or_update_component_link(course_key, created, xblock)
+        _create_or_update_component_link(created, xblock)
     except InvalidKeyError:
         # It is possible that the upstream is a container and UsageKeyV2 parse failed
         # Create upstream container link and raise InvalidKeyError if xblock.upstream is a valid key.
-        _create_or_update_container_link(course_key, created, xblock)
+        _create_or_update_container_link(created, xblock)
 
 
 def get_previous_run_course_key(course_key):

cms/djangoapps/contentstore/xblock_storage_handlers/view_handlers.py

@@ -1118,11 +1118,14 @@ def create_xblock_info(  # lint-amnesty, pylint: disable=too-many-statements
 
     # defining the default value 'True' for delete, duplicate, drag and add new child actions
     # in xblock_actions for each xblock.
+    # The unlinkable action is set to None by default, which means the action is not applicable for
+    # any xblock unless explicitly set to True or False for a specific xblock condition.
     xblock_actions = {
         "deletable": True,
         "draggable": True,
         "childAddable": True,
         "duplicable": True,
+        "unlinkable": None,
     }
     explanatory_message = None
 
@@ -1320,9 +1323,13 @@ def create_xblock_info(  # lint-amnesty, pylint: disable=too-many-statements
         # Also add upstream info
         upstream_info = UpstreamLink.try_get_for_block(xblock, log_error=False).to_json()
         xblock_info["upstream_info"] = upstream_info
-        # Disable adding or removing children component if xblock is imported from library
+
         if upstream_info["upstream_ref"]:
+            # Disable adding or removing children component if xblock is imported from library
             xblock_actions["childAddable"] = False
+            # Enable unlinking only for top level imported components
+            xblock_actions["unlinkable"] = not upstream_info["has_top_level_parent"]
+
         if is_xblock_unit:
             # if xblock is a Unit we add the discussion_enabled option
             xblock_info["discussion_enabled"] = xblock.discussion_enabled

cms/lib/xblock/upstream_sync.py

@@ -28,6 +28,7 @@
 from xblock.exceptions import XBlockNotFoundError
 from xblock.fields import Scope, String, Integer, Dict
 from xblock.core import XBlockMixin, XBlock
+from xmodule.util.keys import BlockKey
 
 if t.TYPE_CHECKING:
     from django.contrib.auth.models import User  # pylint: disable=imported-auth-user
@@ -190,14 +191,14 @@ def try_get_for_block(cls, downstream: XBlock, log_error: bool = True) -> t.Self
                     downstream.upstream,
                 )
             return cls(
-                upstream_ref=getattr(downstream, "upstream", ""),
+                upstream_ref=getattr(downstream, "upstream", None),
                 upstream_key=None,
                 downstream_key=str(getattr(downstream, "usage_key", "")),
                 version_synced=getattr(downstream, "upstream_version", None),
                 version_available=None,
                 version_declined=None,
                 error_message=str(exc),
-                has_top_level_parent=False,
+                has_top_level_parent=getattr(downstream, "top_level_downstream_parent_key", None) is not None,
             )
 
     @classmethod
@@ -311,7 +312,37 @@ def decline_sync(downstream: XBlock, user_id=None) -> None:
         store.update_item(downstream, user_id)
 
 
-def sever_upstream_link(downstream: XBlock) -> None:
+def _update_children_top_level_parent(
+    downstream: XBlock,
+    new_top_level_parent_key: dict[str, str] | None
+) -> list[XBlock]:
+    """
+    Given a new top-level parent block, update the `top_level_downstream_parent_key` field on the downstream block
+    and all of its children.
+
+    If `new_top_level_parent_key` is None, use the current downstream block's usage_key for its children.
+
+    Returns a list of all affected blocks.
+    """
+    if not downstream.has_children:
+        return []
+
+    affected_blocks = []
+    for child in downstream.get_children():
+        child.top_level_downstream_parent_key = new_top_level_parent_key
+        affected_blocks.append(child)
+        # If the `new_top_level_parent_key` is None, the current level assume the top-level
+        # parent key for its children.
+        child_top_level_parent_key = new_top_level_parent_key if new_top_level_parent_key is not None else (
+            BlockKey.from_usage_key(child.usage_key)._asdict()
+        )
+
+        affected_blocks.extend(_update_children_top_level_parent(child, child_top_level_parent_key))
+
+    return affected_blocks
+
+
+def sever_upstream_link(downstream: XBlock) -> list[XBlock]:
     """
     Given an XBlock that is linked to upstream content, disconnect the link, such that authors are never again prompted
     to sync upstream updates. Erase all `.upstream*` fields from the downtream block.
@@ -320,10 +351,12 @@ def sever_upstream_link(downstream: XBlock) -> None:
     because once a downstream block has been de-linked from source (e.g., a Content Library block), it is no different
     than if the block had just been copy-pasted in the first place.
 
-    Does not save `downstream` to the store. That is left up to the caller.
+    Does not save `downstream` (or its children)  to the store. That is left up to the caller.
 
     If `downstream` lacks a link, then this raises NoUpstream (though it is reasonable for callers to handle such
     exception and ignore it, as the end result is the same: `downstream.upstream is None`).
+
+    Returns a list of affected blocks, which includes the `downstream` block itself and all of its children.
     """
     if not downstream.upstream:
         raise NoUpstream()
@@ -336,6 +369,14 @@ def sever_upstream_link(downstream: XBlock) -> None:
             continue
         setattr(downstream, fetched_upstream_field, None)  # Null out upstream_display_name, et al.
 
+    # Set the top_level_dowwnstream_parent_key to None, and calls `_update_children_top_level_parent` to
+    # update all children with the new top_level_dowwnstream_parent_key for each of them.
+    downstream.top_level_downstream_parent_key = None
+    affected_blocks = _update_children_top_level_parent(downstream, None)
+
+    # Return the list of affected blocks, which includes the `downstream` block itself.
+    return [downstream, *affected_blocks]
+
 
 def _get_library_xblock_url(usage_key: LibraryUsageLocatorV2):
     """