squash!: Remove use case for getting latest existing backup, now we always need to specify the task_id

Author: rodmgwgu

openedx/core/djangoapps/content_libraries/api/libraries.py

@@ -692,9 +692,8 @@ def revert_changes(library_key: LibraryLocatorV2, user_id: int | None = None) ->
 
 
 def get_backup_task_status(
-    library_key: LibraryLocatorV2,
     user_id: int,
-    task_id: str | None = None
+    task_id: str
 ) -> dict | None:
     """
     Get the status of a library backup task.
@@ -711,12 +710,6 @@ def get_backup_task_status(
             task_status = UserTaskStatus.objects.get(task_id=task_id, user_id=user_id)
         except UserTaskStatus.DoesNotExist:
             return None
-    else:
-        # Get the latest task for this user and library
-        task_status = UserTaskStatus.objects.filter(
-            user_id=user_id,
-            name__contains=str(library_key)
-        ).order_by('-created').first()
 
     if not task_status:
         return None

openedx/core/djangoapps/content_libraries/rest_api/libraries.py

@@ -719,10 +719,9 @@ class LibraryBackupView(APIView):
 
     **GET Parameters**
 
-        A GET request can include the following parameters:
+        A GET request must include the following parameters:
 
-        * task_id: (optional) The UUID of the task to check. If not provided,
-          the latest task for the library created by the calling user will be returned.
+        * task_id: (required) The UUID of the task to check.
 
     **GET Response Values**
 
@@ -766,22 +765,24 @@ def post(self, request, lib_key_str):
             apidocs.query_parameter(
                 'task_id',
                 str,
-                description="The ID of the backup task to retrieve. If not specified, retrieves the latest task."
+                description="The ID of the backup task to retrieve."
             ),
         ],
         responses={200: LibraryBackupTaskStatusSerializer}
     )
     @convert_exceptions
     def get(self, request, lib_key_str):
         """
-        Get the status of the specified or latest backup task for the specified library.
+        Get the status of the specified backup task for the specified library.
         """
         library_key = LibraryLocatorV2.from_string(lib_key_str)
         # Using CAN_EDIT_THIS_CONTENT_LIBRARY permission for now. This should eventually become its own permission
         api.require_permission_for_library_key(library_key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY)
 
         task_id = request.query_params.get('task_id', None)
-        result = get_backup_task_status(library_key, request.user.id, task_id)
+        if not task_id:
+            raise ValidationError(detail={'task_id': _('This field is required.')})
+        result = get_backup_task_status(request.user.id, task_id)
 
         if not result:
             raise NotFound(detail="No backup found for this library.")

openedx/core/djangoapps/content_libraries/tests/base.py

@@ -32,7 +32,8 @@
 URL_LIB_TEAM_USER = URL_LIB_TEAM + 'user/{username}/'  # Add/edit/remove a user's permission to use this library
 URL_LIB_TEAM_GROUP = URL_LIB_TEAM + 'group/{group_name}/'  # Add/edit/remove a group's permission to use this library
 URL_LIB_PASTE_CLIPBOARD = URL_LIB_DETAIL + 'paste_clipboard/'  # Paste user clipboard (POST) containing Xblock data
-URL_LIB_BACKUP = URL_LIB_DETAIL + 'backup/'  # Start or get status on a backup task for this library
+URL_LIB_BACKUP = URL_LIB_DETAIL + 'backup/'  # Start a backup task for this library
+URL_LIB_BACKUP_GET = URL_LIB_BACKUP + '?{query_params}'  # Get status on a backup task for this library
 URL_LIB_BLOCK = URL_PREFIX + 'blocks/{block_key}/'  # Get data about a block, or delete it
 URL_LIB_BLOCK_PUBLISH = URL_LIB_BLOCK + 'publish/'  # Publish changes from a specified XBlock
 URL_LIB_BLOCK_OLX = URL_LIB_BLOCK + 'olx/'  # Get or set the OLX of the specified XBlock
@@ -328,7 +329,7 @@ def _start_library_backup_task(self, lib_key, expect_response=200):
     def _get_library_backup_task(self, lib_key, task_id, expect_response=200):
         """ Get the status of a backup task for this library """
         query_params = urlencode({"task_id": task_id})
-        url = URL_LIB_BACKUP.format(lib_key=lib_key, query_params=query_params)
+        url = URL_LIB_BACKUP_GET.format(lib_key=lib_key, query_params=query_params)
         return self._api('get', url, None, expect_response)
 
     def _render_block_view(self, block_key, view_name, version=None, expect_response=200):

openedx/core/djangoapps/content_libraries/tests/test_api.py

@@ -1329,11 +1329,11 @@ def setUp(self) -> None:
         self.wrong_task_id = '11111111-1111-1111-1111-111111111111'
 
     def test_get_backup_task_status_no_task(self) -> None:
-        status = api.get_backup_task_status(self.lib1.library_key, self.user.id)
+        status = api.get_backup_task_status(self.user.id, None)
         assert status is None
 
     def test_get_backup_task_status_wrong_task_id(self) -> None:
-        status = api.get_backup_task_status(self.lib1.library_key, self.user.id, task_id=self.wrong_task_id)
+        status = api.get_backup_task_status(self.user.id, task_id=self.wrong_task_id)
         assert status is None
 
     def test_get_backup_task_status_in_progress(self) -> None:
@@ -1351,7 +1351,7 @@ def test_get_backup_task_status_in_progress(self) -> None:
         ) as mock_get:
             mock_get.return_value = mock_task
 
-            status = api.get_backup_task_status(self.lib1.library_key, self.user.id, task_id=task_id)
+            status = api.get_backup_task_status(self.user.id, task_id=task_id)
             assert status is not None
             assert status['state'] == UserTaskStatus.IN_PROGRESS
             assert status['url'] is None
@@ -1379,39 +1379,11 @@ def test_get_backup_task_status_succeeded(self) -> None:
             mock_get.return_value = mock_task
             mock_artifact_get.return_value = mock_artifact
 
-            status = api.get_backup_task_status(self.lib1.library_key, self.user.id, task_id=task_id)
+            status = api.get_backup_task_status(self.user.id, task_id=task_id)
             assert status is not None
             assert status['state'] == UserTaskStatus.SUCCEEDED
             assert status['url'] == "/media/user_tasks/2025/10/01/library-libOEXCSPROB_mOw1rPL.zip"
 
-    def test_get_backup_task_status_latest_task(self) -> None:
-        # Test getting the latest task when no task_id is provided
-        mock_task = UserTaskStatus(
-            task_id=str(uuid.uuid4()),
-            user_id=self.user.id,
-            name=f"Export of {self.lib1.library_key}",
-            state=UserTaskStatus.PENDING
-        )
-
-        mock_queryset = mock.Mock()
-        mock_queryset.order_by.return_value.first.return_value = mock_task
-
-        with mock.patch(
-            'openedx.core.djangoapps.content_libraries.api.libraries.UserTaskStatus.objects.filter'
-        ) as mock_filter:
-            mock_filter.return_value = mock_queryset
-
-            status = api.get_backup_task_status(self.lib1.library_key, self.user.id)
-            assert status is not None
-            assert status['state'] == UserTaskStatus.PENDING
-            assert status['url'] is None
-
-            # Verify the filter was called with correct parameters
-            mock_filter.assert_called_once_with(
-                user_id=self.user.id,
-                name__contains=str(self.lib1.library_key)
-            )
-
     def test_get_backup_task_status_failed(self) -> None:
         # Create a mock UserTaskStatus in FAILED state
         task_id = str(uuid.uuid4())
@@ -1427,7 +1399,7 @@ def test_get_backup_task_status_failed(self) -> None:
         ) as mock_get:
             mock_get.return_value = mock_task
 
-            status = api.get_backup_task_status(self.lib1.library_key, self.user.id, task_id=task_id)
+            status = api.get_backup_task_status(self.user.id, task_id=task_id)
             assert status is not None
             assert status['state'] == UserTaskStatus.FAILED
             assert status['url'] is None