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

feat: Implementation of library v2 backup endpoints

Browse Source
This commit is contained in:
Rodrigo Mendez
2025-10-01 12:47:33 -06:00
committed by David Ormsbee
parent 913598076c
commit 0c9997ce92
9 changed files with 446 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
openedx/core/djangoapps/content_libraries/api/libraries.py
View File

@@ -41,9 +41,10 @@ could be promoted to the core XBlock API and made generic.
"""
from __future__ import annotations
from dataclasses import dataclass, field as dataclass_field
from datetime import datetime
import logging
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from datetime import datetime
from django.conf import settings
from django.contrib.auth.models import AbstractUser, AnonymousUser, Group
@@ -53,29 +54,24 @@ from django.db import IntegrityError, transaction
from django.db.models import Q, QuerySet
from django.utils.translation import gettext as _
from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
from openedx_events.content_authoring.data import (
ContentLibraryData,
)
from openedx_events.content_authoring.data import ContentLibraryData
from openedx_events.content_authoring.signals import (
CONTENT_LIBRARY_CREATED,
CONTENT_LIBRARY_DELETED,
CONTENT_LIBRARY_UPDATED,
CONTENT_LIBRARY_UPDATED
)
from openedx_learning.api import authoring as authoring_api
from openedx_learning.api.authoring_models import Component
from organizations.models import Organization
from user_tasks.models import UserTaskArtifact, UserTaskStatus
from xblock.core import XBlock
from openedx.core.types import User as UserType
from .. import permissions
from .. import permissions, tasks
from ..constants import ALL_RIGHTS_RESERVED
from ..models import ContentLibrary, ContentLibraryPermission
from .. import tasks
from .exceptions import (
LibraryAlreadyExists,
LibraryPermissionIntegrityError,
)
from .exceptions import LibraryAlreadyExists, LibraryPermissionIntegrityError
log = logging.getLogger(__name__)
@@ -105,6 +101,7 @@ __all__ = [
"get_allowed_block_types",
"publish_changes",
"revert_changes",
"get_backup_task_status",
]
@@ -692,3 +689,30 @@ def revert_changes(library_key: LibraryLocatorV2, user_id: int | None = None) ->
# Call the event handlers as needed.
tasks.wait_for_post_revert_events(draft_change_log, library_key)
def get_backup_task_status(
user_id: int,
task_id: str
) -> dict | None:
"""
Get the status of a library backup task.
Returns a dictionary with the following keys:
- state: One of "Pending", "Exporting", "Succeeded", "Failed"
- url: If state is "Succeeded", the URL where the exported .zip file can be downloaded. Otherwise, None.
If no task is found, returns None.
"""
try:
task_status = UserTaskStatus.objects.get(task_id=task_id, user_id=user_id)
except UserTaskStatus.DoesNotExist:
return None
result = {'state': task_status.state, 'url': None}
if task_status.state == UserTaskStatus.SUCCEEDED:
artifact = UserTaskArtifact.objects.get(status=task_status, name='Output')
result['url'] = artifact.file.storage.url(artifact.file.name)
return result

127
openedx/core/djangoapps/content_libraries/rest_api/libraries.py
View File

@@ -66,6 +66,7 @@ import itertools
import json
import logging
import edx_api_doc_tools as apidocs
from django.conf import settings
from django.contrib.auth import authenticate, get_user_model, login
from django.contrib.auth.models import Group
@@ -78,14 +79,12 @@ from django.views.decorators.clickjacking import xframe_options_exempt
from django.views.decorators.csrf import csrf_exempt
from django.views.generic.base import TemplateResponseMixin, View
from drf_yasg.utils import swagger_auto_schema
from pylti1p3.contrib.django import DjangoCacheDataStorage, DjangoDbToolConf, DjangoMessageLaunch, DjangoOIDCLogin
from pylti1p3.exception import LtiException, OIDCException
import edx_api_doc_tools as apidocs
from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
from organizations.api import ensure_organization
from organizations.exceptions import InvalidOrganizationException
from organizations.models import Organization
from pylti1p3.contrib.django import DjangoCacheDataStorage, DjangoDbToolConf, DjangoMessageLaunch, DjangoOIDCLogin
from pylti1p3.exception import LtiException, OIDCException
from rest_framework import status
from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError
from rest_framework.generics import GenericAPIView
@@ -93,12 +92,15 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.viewsets import GenericViewSet
import openedx.core.djangoapps.site_configuration.helpers as configuration_helpers
from cms.djangoapps.contentstore.views.course import (
get_allowed_organizations_for_libraries,
user_can_create_organizations,
user_can_create_organizations
)
from openedx.core.djangoapps.content_libraries import api, permissions
from openedx.core.djangoapps.content_libraries.api.libraries import get_backup_task_status
from openedx.core.djangoapps.content_libraries.rest_api.serializers import (
ContentLibraryAddPermissionByEmailSerializer,
ContentLibraryBlockImportTaskCreateSerializer,
ContentLibraryBlockImportTaskSerializer,
ContentLibraryFilterSerializer,
@@ -106,20 +108,20 @@ from openedx.core.djangoapps.content_libraries.rest_api.serializers import (
ContentLibraryPermissionLevelSerializer,
ContentLibraryPermissionSerializer,
ContentLibraryUpdateSerializer,
LibraryBackupResponseSerializer,
LibraryBackupTaskStatusSerializer,
LibraryXBlockCreationSerializer,
LibraryXBlockMetadataSerializer,
LibraryXBlockTypeSerializer,
ContentLibraryAddPermissionByEmailSerializer,
PublishableItemSerializer,
PublishableItemSerializer
)
import openedx.core.djangoapps.site_configuration.helpers as configuration_helpers
from openedx.core.lib.api.view_utils import view_auth_classes
from openedx.core.djangoapps.content_libraries.tasks import backup_library
from openedx.core.djangoapps.safe_sessions.middleware import mark_user_change_as_expected
from openedx.core.djangoapps.xblock import api as xblock_api
from openedx.core.lib.api.view_utils import view_auth_classes
from .utils import convert_exceptions
from ..models import ContentLibrary, LtiGradedResource, LtiProfile
from .utils import convert_exceptions
User = get_user_model()
log = logging.getLogger(__name__)
@@ -685,6 +687,109 @@ class LibraryImportTaskViewSet(GenericViewSet):
return Response(ContentLibraryBlockImportTaskSerializer(import_task).data)
# Library Backup Views
# ====================
@method_decorator(non_atomic_requests, name="dispatch")
@view_auth_classes()
class LibraryBackupView(APIView):
"""
**Use Case**
* Start an asynchronous task to back up the content of a library to a .zip file
* Get a status on an asynchronous export task
**Example Requests**
POST /api/libraries/v2/{library_id}/backup/
GET /api/libraries/v2/{library_id}/backup/?task_id={task_id}
**POST Response Values**
If the import task is started successfully, an HTTP 200 "OK" response is
returned.
The HTTP 200 response has the following values:
* task_id: UUID of the created task, usable for checking status
**Example POST Response**
{
"task_id": "7069b95b-ccea-4214-b6db-e00f27065bf7"
}
**GET Parameters**
A GET request must include the following parameters:
* task_id: (required) The UUID of the task to check.
**GET Response Values**
If the import task is found successfully by the UUID provided, an HTTP
200 "OK" response is returned.
The HTTP 200 response has the following values:
* state: String description of the state of the task.
Possible states: "Pending", "Exporting", "Succeeded", "Failed".
* url: (may be null) If the task is complete, a URL to download the .zip file
**Example GET Response**
{
"state": "Succeeded",
"url": "/media/user_tasks/2025/10/03/lib-wgu-csprob-2025-10-03-153633.zip"
}
"""
@apidocs.schema(
body=None,
responses={200: LibraryBackupResponseSerializer}
)
@convert_exceptions
def post(self, request, lib_key_str):
"""
Start 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)
async_result = backup_library.delay(request.user.id, str(library_key))
result = {'task_id': async_result.task_id}
return Response(LibraryBackupResponseSerializer(result).data)
@apidocs.schema(
parameters=[
apidocs.query_parameter(
'task_id',
str,
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 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)
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.")
return Response(LibraryBackupTaskStatusSerializer(result).data)
# LTI 1.3 Views
# =============

35
openedx/core/djangoapps/content_libraries/rest_api/serializers.py
View File

@@ -3,26 +3,22 @@ Serializers for the content libraries REST API
"""
# pylint: disable=abstract-method
from django.core.validators import validate_unicode_slug
from opaque_keys import InvalidKeyError, OpaqueKey
from opaque_keys.edx.locator import LibraryContainerLocator, LibraryUsageLocatorV2
from openedx_learning.api.authoring_models import Collection
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from opaque_keys import OpaqueKey
from opaque_keys.edx.locator import LibraryContainerLocator, LibraryUsageLocatorV2
from opaque_keys import InvalidKeyError
from openedx_learning.api.authoring_models import Collection
from openedx.core.djangoapps.content_libraries.api.containers import ContainerType
from openedx.core.djangoapps.content_libraries.constants import (
ALL_RIGHTS_RESERVED,
LICENSE_OPTIONS,
)
from openedx.core.djangoapps.content_libraries.constants import ALL_RIGHTS_RESERVED, LICENSE_OPTIONS
from openedx.core.djangoapps.content_libraries.models import (
ContentLibraryPermission, ContentLibraryBlockImportTask,
ContentLibrary
ContentLibrary,
ContentLibraryBlockImportTask,
ContentLibraryPermission
)
from openedx.core.lib.api.serializers import CourseKeyField
from .. import permissions
from .. import permissions
DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
@@ -416,3 +412,18 @@ class ContainerHierarchySerializer(serializers.Serializer):
units = serializers.ListField(child=ContainerHierarchyMemberSerializer(), allow_empty=True)
components = serializers.ListField(child=ContainerHierarchyMemberSerializer(), allow_empty=True)
object_key = OpaqueKeySerializer()
class LibraryBackupResponseSerializer(serializers.Serializer):
"""
Serializer for the response after requesting a backup of a content library.
"""
task_id = serializers.CharField()
class LibraryBackupTaskStatusSerializer(serializers.Serializer):
"""
Serializer for checking the status of a library backup task.
"""
state = serializers.CharField()
url = serializers.URLField(allow_null=True)

96
openedx/core/djangoapps/content_libraries/tasks.py
View File

@@ -17,37 +17,44 @@ Architecture note:
from __future__ import annotations
import logging
import os
from datetime import datetime
from tempfile import mkdtemp
from celery import shared_task
from celery_utils.logged_task import LoggedTask
from celery.utils.log import get_task_logger
from edx_django_utils.monitoring import set_code_owner_attribute, set_code_owner_attribute_from_module
from celery_utils.logged_task import LoggedTask
from django.core.files import File
from django.utils.text import slugify
from edx_django_utils.monitoring import (
set_code_owner_attribute,
set_code_owner_attribute_from_module,
set_custom_attribute
)
from opaque_keys.edx.keys import CourseKey
from opaque_keys.edx.locator import (
BlockUsageLocator,
LibraryCollectionLocator,
LibraryContainerLocator,
LibraryLocatorV2,
)
from openedx_learning.api import authoring as authoring_api
from openedx_learning.api.authoring_models import DraftChangeLog, PublishLog
from openedx_events.content_authoring.data import (
LibraryBlockData,
LibraryCollectionData,
LibraryContainerData,
LibraryLocatorV2
)
from openedx_events.content_authoring.data import LibraryBlockData, LibraryCollectionData, LibraryContainerData
from openedx_events.content_authoring.signals import (
LIBRARY_BLOCK_CREATED,
LIBRARY_BLOCK_DELETED,
LIBRARY_BLOCK_UPDATED,
LIBRARY_BLOCK_PUBLISHED,
LIBRARY_BLOCK_UPDATED,
LIBRARY_COLLECTION_UPDATED,
LIBRARY_CONTAINER_CREATED,
LIBRARY_CONTAINER_DELETED,
LIBRARY_CONTAINER_UPDATED,
LIBRARY_CONTAINER_PUBLISHED,
LIBRARY_CONTAINER_UPDATED
)
from openedx_learning.api import authoring as authoring_api
from openedx_learning.api.authoring import create_zip_file as create_lib_zip_file
from openedx_learning.api.authoring_models import DraftChangeLog, PublishLog
from path import Path
from user_tasks.models import UserTaskArtifact
from user_tasks.tasks import UserTask, UserTaskStatus
from xblock.fields import Scope
@@ -477,3 +484,66 @@ def _copy_overrides(
dest_block=store.get_item(dest_child_key),
)
store.update_item(dest_block, user_id)
class LibraryBackupTask(UserTask): # pylint: disable=abstract-method
"""
Base class for tasks related with Library backup functionality.
"""
@classmethod
def generate_name(cls, arguments_dict) -> str:
"""
Create a name for this particular backup task instance.
Should be both:
a. semi human-friendly
b. something we can query in order to determine whether the library has a task in progress
Arguments:
arguments_dict (dict): The arguments given to the task function
Returns:
str: The generated name
"""
key = arguments_dict['library_key_str']
return f'Backup of {key}'
@shared_task(base=LibraryBackupTask, bind=True)
# Note: The decorator @set_code_owner_attribute cannot be used here because the UserTaskMixin
# does stack inspection and can't handle additional decorators.
def backup_library(self, user_id: int, library_key_str: str) -> None:
"""
Export a library to a .zip archive and prepare it for download.
Possible Task states:
- Pending: Task is created but not started yet.
- Exporting: Task is running and the library is being exported.
- Succeeded: Task completed successfully and the exported file is available for download.
- Failed: Task failed and the export did not complete.
"""
ensure_cms("backup_library may only be executed in a CMS context")
set_code_owner_attribute_from_module(__name__)
library_key = LibraryLocatorV2.from_string(library_key_str)
try:
self.status.set_state('Exporting')
set_custom_attribute("exporting_started", str(library_key))
root_dir = Path(mkdtemp())
sanitized_lib_key = str(library_key).replace(":", "-")
sanitized_lib_key = slugify(sanitized_lib_key, allow_unicode=True)
timestamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
filename = f'{sanitized_lib_key}-{timestamp}.zip'
file_path = os.path.join(root_dir, filename)
create_lib_zip_file(lp_key=str(library_key), path=file_path)
set_custom_attribute("exporting_completed", str(library_key))
with open(file_path, 'rb') as zipfile:
artifact = UserTaskArtifact(status=self.status, name='Output')
artifact.file.save(name=os.path.basename(zipfile.name), content=File(zipfile))
artifact.save()
except Exception as exception: # pylint: disable=broad-except
TASK_LOGGER.exception('Error exporting library %s', library_key, exc_info=True)
if self.status.state != UserTaskStatus.FAILED:
self.status.fail({'raw_error_msg': str(exception)})

13
openedx/core/djangoapps/content_libraries/tests/base.py
View File

@@ -32,6 +32,8 @@ URL_LIB_TEAM = URL_LIB_DETAIL + 'team/' # Get the list of users/groups authoriz
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 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
@@ -319,6 +321,17 @@ class ContentLibrariesRestApiTest(APITransactionTestCase):
url = URL_LIB_PASTE_CLIPBOARD.format(lib_key=lib_key)
return self._api('post', url, {}, expect_response)
def _start_library_backup_task(self, lib_key, expect_response=200):
""" Start a backup task for this library """
url = URL_LIB_BACKUP.format(lib_key=lib_key)
return self._api('post', url, {}, expect_response)
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_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):
"""
Render an XBlock's view in the active application's runtime.

94
openedx/core/djangoapps/content_libraries/tests/test_api.py
View File

@@ -4,9 +4,11 @@ Tests for Content Library internal api.
import base64
import hashlib
import uuid
from unittest import mock
from django.test import TestCase
from user_tasks.models import UserTaskStatus
from opaque_keys.edx.keys import (
CourseKey,
@@ -1309,3 +1311,95 @@ class ContentLibraryContainersTest(ContentLibrariesRestApiTest):
),
},
)
class ContentLibraryExportTest(ContentLibrariesRestApiTest):
"""
Tests for Content Library API export methods.
"""
def setUp(self) -> None:
super().setUp()
# Create Content Libraries
self._create_library("test-lib-exp-1", "Test Library Export 1")
# Fetch the created ContentLibrary objects so we can access their learning_package.id
self.lib1 = ContentLibrary.objects.get(slug="test-lib-exp-1")
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.user.id, "")
assert status is None
def test_get_backup_task_status_wrong_task_id(self) -> None:
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:
# Create a mock UserTaskStatus in IN_PROGRESS state
task_id = str(uuid.uuid4())
mock_task = UserTaskStatus(
task_id=task_id,
user_id=self.user.id,
name=f"Export of {self.lib1.library_key}",
state=UserTaskStatus.IN_PROGRESS
)
with mock.patch(
'openedx.core.djangoapps.content_libraries.api.libraries.UserTaskStatus.objects.get'
) as mock_get:
mock_get.return_value = mock_task
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
def test_get_backup_task_status_succeeded(self) -> None:
# Create a mock UserTaskStatus in SUCCEEDED state
task_id = str(uuid.uuid4())
mock_task = UserTaskStatus(
task_id=task_id,
user_id=self.user.id,
name=f"Export of {self.lib1.library_key}",
state=UserTaskStatus.SUCCEEDED
)
# Create a mock UserTaskArtifact
mock_artifact = mock.Mock()
mock_artifact.file.storage.url.return_value = "/media/user_tasks/2025/10/01/library-libOEXCSPROB_mOw1rPL.zip"
with mock.patch(
'openedx.core.djangoapps.content_libraries.api.libraries.UserTaskStatus.objects.get'
) as mock_get, mock.patch(
'openedx.core.djangoapps.content_libraries.api.libraries.UserTaskArtifact.objects.get'
) as mock_artifact_get:
mock_get.return_value = mock_task
mock_artifact_get.return_value = mock_artifact
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_failed(self) -> None:
# Create a mock UserTaskStatus in FAILED state
task_id = str(uuid.uuid4())
mock_task = UserTaskStatus(
task_id=task_id,
user_id=self.user.id,
name=f"Export of {self.lib1.library_key}",
state=UserTaskStatus.FAILED
)
with mock.patch(
'openedx.core.djangoapps.content_libraries.api.libraries.UserTaskStatus.objects.get'
) as mock_get:
mock_get.return_value = mock_task
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

34
openedx/core/djangoapps/content_libraries/tests/test_content_libraries.py
View File

@@ -823,6 +823,40 @@ class ContentLibrariesTestCase(ContentLibrariesRestApiTest):
"id": f"lb:CL-TEST:test_lib_paste_clipboard:problem:{pasted_usage_key.block_id}",
})
def test_start_library_backup(self):
"""
Test starting a backup operation on a content library.
"""
author = UserFactory.create(username="Author", email="author@example.com", is_staff=True)
with self.as_user(author):
lib = self._create_library(
slug="test_lib_backup",
title="Backup Test Library",
description="Testing backup for library"
)
lib_id = lib["id"]
response = self._start_library_backup_task(lib_id)
assert response["task_id"] is not None
def test_get_library_backup_status(self):
"""
Test getting the status of a backup operation on a content library.
"""
author = UserFactory.create(username="Author", email="author@example.com", is_staff=True)
with self.as_user(author):
lib = self._create_library(
slug="test_lib_backup_status",
title="Backup Status Test Library",
description="Testing backup status for library"
)
lib_id = lib["id"]
response = self._start_library_backup_task(lib_id)
task_id = response["task_id"]
# Now check the status of the backup task
status_response = self._get_library_backup_task(lib_id, task_id)
assert status_response["state"] in ["Pending", "Exporting", "Succeeded", "Failed"]
@override_settings(LIBRARY_ENABLED_BLOCKS=['problem', 'video', 'html'])
def test_library_get_enabled_blocks(self):
expected = [

45
openedx/core/djangoapps/content_libraries/tests/test_tasks.py Normal file
View File

@@ -0,0 +1,45 @@
"""
Unit tests for content libraries Celery tasks
"""
from ..models import ContentLibrary
from .base import ContentLibrariesRestApiTest
from openedx.core.djangoapps.content_libraries.tasks import backup_library
from user_tasks.models import UserTaskArtifact
class ContentLibraryBackupTaskTest(ContentLibrariesRestApiTest):
"""
Tests for Content Library export task.
"""
def setUp(self) -> None:
super().setUp()
# Create Content Libraries
self._create_library("test-lib-task-1", "Test Library Task 1")
# Fetch the created ContentLibrary objects so we can access their learning_package.id
self.lib1 = ContentLibrary.objects.get(slug="test-lib-task-1")
self.wrong_task_id = '11111111-1111-1111-1111-111111111111'
def test_backup_task_returns_task_id(self):
result = backup_library.delay(self.user.id, str(self.lib1.library_key))
assert result.task_id is not None
def test_backup_task_success(self):
result = backup_library.delay(self.user.id, str(self.lib1.library_key))
assert result.state == 'SUCCESS'
# Ensure an artifact was created with the output file
artifact = UserTaskArtifact.objects.filter(status__task_id=result.task_id, name='Output').first()
assert artifact is not None
assert artifact.file.name.endswith('.zip')
def test_backup_task_failure(self):
result = backup_library.delay(self.user.id, self.wrong_task_id)
assert result.state == 'FAILURE'
# Ensure an error artifact was created
artifact = UserTaskArtifact.objects.filter(status__task_id=result.task_id, name='Error').first()
assert artifact is not None
assert artifact.text is not None

2
openedx/core/djangoapps/content_libraries/urls.py
View File

@@ -54,6 +54,8 @@ urlpatterns = [
path('import_blocks/', include(import_blocks_router.urls)),
# Paste contents of clipboard into library
path('paste_clipboard/', libraries.LibraryPasteClipboardView.as_view()),
# Start a backup task for this library
path('backup/', libraries.LibraryBackupView.as_view()),
# Library Collections
path('', include(library_collections_router.urls)),
])),