feat: Add tags.csv to export tarball

cms/djangoapps/contentstore/views/import_export.py

@@ -39,6 +39,7 @@
 from common.djangoapps.util.views import ensure_valid_course_key
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 
+from openedx.core.djangoapps.content_tagging.auth import has_view_object_tags_access
 from ..storage import course_import_export_storage
 from ..tasks import CourseExportTask, CourseImportTask, export_olx, import_olx
 from ..toggles import use_new_export_page, use_new_import_page
@@ -343,6 +344,11 @@ def export_handler(request, course_key_string):
     requested_format = request.GET.get('_accept', request.META.get('HTTP_ACCEPT', 'text/html'))
 
     if request.method == 'POST':
+        if not has_view_object_tags_access(request.user, str(course_key)):
+            raise PermissionDenied(
+                "You do not have permission to view object tags for this object_id."
+            )
+
         export_olx.delay(request.user.id, course_key_string, request.LANGUAGE_CODE)
         return JsonResponse({'ExportStatus': 1})
     elif 'text/html' in requested_format:

openedx/core/djangoapps/content_tagging/api.py

@@ -2,20 +2,19 @@
 Content Tagging APIs
 """
 from __future__ import annotations
+import io
 
 from itertools import groupby
 import csv
 from typing import Iterator
 from opaque_keys.edx.keys import UsageKey
 
 import openedx_tagging.core.tagging.api as oel_tagging
-from openedx_tagging.core.tagging import rules as oel_tagging_rules
 from django.db.models import Exists, OuterRef, Q, QuerySet
 from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import LibraryLocatorV2
 from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
 from organizations.models import Organization
-from rest_framework.exceptions import PermissionDenied, ValidationError
 from .helpers.objecttag_export_helpers import build_object_tree_with_objecttags, iterate_with_level
 
 from .models import TaxonomyOrg
@@ -208,24 +207,14 @@ def set_object_tags(
         )
 
 
-def generate_csv(object_id, user, buffer) -> Iterator[str]:
+def generate_csv_rows(object_id, buffer) -> Iterator[str]:
     """
-    Returns a CSV string with tags and taxonomies of all blocks of `content_key`
+    Returns a CSV string with tags and taxonomies of all blocks of `object_id`
     """
     content_key = get_content_key_from_string(object_id)
 
     if isinstance(content_key, UsageKey):
-        raise ValidationError("The object_id must be a CourseKey or a LibraryLocatorV2.")
-
-    # Check if the user has permission to view object tags for this object_id
-    if not user.has_perm(
-        "oel_tagging.view_objecttag",
-        # The obj arg expects a model, but we are passing an object
-        oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=object_id),  # type: ignore[arg-type]
-    ):
-        raise PermissionDenied(
-            "You do not have permission to view object tags for this object_id."
-        )
+        raise ValueError("The object_id must be a CourseKey or a LibraryLocatorV2.")
 
     all_object_tags, taxonomies = get_all_object_tags(content_key)
     tagged_content = build_object_tree_with_objecttags(content_key, all_object_tags)
@@ -258,6 +247,20 @@ def generate_csv(object_id, user, buffer) -> Iterator[str]:
         yield csv_writer.writerow(block_data)
 
 
+def export_tags_in_csv_file(object_id, file_dir, file_name) -> None:
+    """
+    Writes a CSV file with tags and taxonomies of all blocks of `object_id`
+    """
+    buffer = io.StringIO()
+    for _ in generate_csv_rows(object_id, buffer):
+        # The generate_csv_rows function is a generator,
+        # we don't need to do anything with the result here
+        pass
+
+    with file_dir.open(file_name, 'w') as csv_file:
+        buffer.seek(0)
+        csv_file.write(buffer.read())
+
 # Expose the oel_tagging APIs
 
 get_taxonomy = oel_tagging.get_taxonomy

openedx/core/djangoapps/content_tagging/auth.py

@@ -0,0 +1,14 @@
+"""
+Functions to validate the access in content tagging actions
+"""
+
+
+from openedx_tagging.core.tagging import rules as oel_tagging_rules
+
+
+def has_view_object_tags_access(user, object_id):
+    return user.has_perm(
+        "oel_tagging.view_objecttag",
+        # The obj arg expects a model, but we are passing an object
+        oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=object_id),  # type: ignore[arg-type]
+    )

openedx/core/djangoapps/content_tagging/helpers/test_objecttag_export_helpers.py

@@ -1,15 +1,189 @@
 """
 Test the objecttag_export_helpers module
 """
+import time
 from unittest.mock import patch
 
 from openedx.core.djangoapps.content_libraries import api as library_api
 from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
 
 from .. import api
-from ..tests.test_api import TestGetAllObjectTagsMixin
 from .objecttag_export_helpers import TaggedContent, build_object_tree_with_objecttags, iterate_with_level
+from openedx_tagging.core.tagging.models import ObjectTag, Tag
+from organizations.models import Organization
+
+
+class TestGetAllObjectTagsMixin:
+    """
+    Set up data to test get_all_object_tags functions
+    """
+
+    def setUp(self):
+        super().setUp()
+
+        self.orgA = Organization.objects.create(name="Organization A", short_name="orgA")
+        self.taxonomy_1 = api.create_taxonomy(name="Taxonomy 1")
+        api.set_taxonomy_orgs(self.taxonomy_1, all_orgs=True)
+        Tag.objects.create(
+            taxonomy=self.taxonomy_1,
+            value="Tag 1.1",
+        )
+        Tag.objects.create(
+            taxonomy=self.taxonomy_1,
+            value="Tag 1.2",
+        )
+
+        self.taxonomy_2 = api.create_taxonomy(name="Taxonomy 2")
+        api.set_taxonomy_orgs(self.taxonomy_2, all_orgs=True)
+
+        Tag.objects.create(
+            taxonomy=self.taxonomy_2,
+            value="Tag 2.1",
+        )
+        Tag.objects.create(
+            taxonomy=self.taxonomy_2,
+            value="Tag 2.2",
+        )
+
+        api.tag_object(
+            object_id="course-v1:orgA+test_course+test_run",
+            taxonomy=self.taxonomy_1,
+            tags=['Tag 1.1'],
+        )
+        self.course_tags = api.get_object_tags("course-v1:orgA+test_course+test_run")
+
+        # Tag blocks
+        api.tag_object(
+            object_id="block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            taxonomy=self.taxonomy_1,
+            tags=['Tag 1.1', 'Tag 1.2'],
+        )
+        self.sequential_tags1 = api.get_object_tags(
+            "block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            taxonomy_id=self.taxonomy_1.id,
+
+        )
+        api.tag_object(
+            object_id="block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            taxonomy=self.taxonomy_2,
+            tags=['Tag 2.1'],
+        )
+        self.sequential_tags2 = api.get_object_tags(
+            "block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            taxonomy_id=self.taxonomy_2.id,
+        )
+        api.tag_object(
+            object_id="block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1",
+            taxonomy=self.taxonomy_2,
+            tags=['Tag 2.2'],
+        )
+        self.vertical1_tags = api.get_object_tags(
+            "block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1"
+        )
+        api.tag_object(
+            object_id="block-v1:orgA+test_course+test_run+type@html+block@test_html",
+            taxonomy=self.taxonomy_2,
+            tags=['Tag 2.1'],
+        )
+        self.html_tags = api.get_object_tags("block-v1:orgA+test_course+test_run+type@html+block@test_html")
+
+        # Create "deleted" object tags, which will be omitted from the results.
+        for object_id in (
+            "course-v1:orgA+test_course+test_run",
+            "block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential",
+            "block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1",
+            "block-v1:orgA+test_course+test_run+type@html+block@test_html",
+        ):
+            ObjectTag.objects.create(
+                object_id=str(object_id),
+                taxonomy=None,
+                tag=None,
+                _value="deleted tag",
+                _name="deleted taxonomy",
+            )
+
+        self.expected_course_objecttags = {
+            "course-v1:orgA+test_course+test_run": {
+                self.taxonomy_1.id: list(self.course_tags),
+            },
+            "block-v1:orgA+test_course+test_run+type@sequential+block@test_sequential": {
+                self.taxonomy_1.id: list(self.sequential_tags1),
+                self.taxonomy_2.id: list(self.sequential_tags2),
+            },
+            "block-v1:orgA+test_course+test_run+type@vertical+block@test_vertical1": {
+                self.taxonomy_2.id: list(self.vertical1_tags),
+            },
+            "block-v1:orgA+test_course+test_run+type@html+block@test_html": {
+                self.taxonomy_2.id: list(self.html_tags),
+            },
+        }
+
+        # Library tags and library contents need a unique block_id that is persisted along test runs
+        self.block_suffix = str(round(time.time() * 1000))
+
+        api.tag_object(
+            object_id=f"lib:orgA:lib_{self.block_suffix}",
+            taxonomy=self.taxonomy_2,
+            tags=['Tag 2.1'],
+        )
+        self.library_tags = api.get_object_tags(f"lib:orgA:lib_{self.block_suffix}")
+
+        api.tag_object(
+            object_id=f"lb:orgA:lib_{self.block_suffix}:problem:problem1_{self.block_suffix}",
+            taxonomy=self.taxonomy_1,
+            tags=['Tag 1.1'],
+        )
+        self.problem1_tags = api.get_object_tags(
+            f"lb:orgA:lib_{self.block_suffix}:problem:problem1_{self.block_suffix}"
+        )
+
+        api.tag_object(
+            object_id=f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}",
+            taxonomy=self.taxonomy_1,
+            tags=['Tag 1.2'],
+        )
+        self.library_html_tags1 = api.get_object_tags(
+            object_id=f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}",
+            taxonomy_id=self.taxonomy_1.id,
+        )
+
+        api.tag_object(
+            object_id=f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}",
+            taxonomy=self.taxonomy_2,
+            tags=['Tag 2.2'],
+        )
+        self.library_html_tags2 = api.get_object_tags(
+            object_id=f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}",
+            taxonomy_id=self.taxonomy_2.id,
+        )
+
+        # Create "deleted" object tags, which will be omitted from the results.
+        for object_id in (
+            f"lib:orgA:lib_{self.block_suffix}",
+            f"lb:orgA:lib_{self.block_suffix}:problem:problem1_{self.block_suffix}",
+            f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}",
+        ):
+            ObjectTag.objects.create(
+                object_id=object_id,
+                taxonomy=None,
+                tag=None,
+                _value="deleted tag",
+                _name="deleted taxonomy",
+            )
+
+        self.expected_library_objecttags = {
+            f"lib:orgA:lib_{self.block_suffix}": {
+                self.taxonomy_2.id: list(self.library_tags),
+            },
+            f"lb:orgA:lib_{self.block_suffix}:problem:problem1_{self.block_suffix}": {
+                self.taxonomy_1.id: list(self.problem1_tags),
+            },
+            f"lb:orgA:lib_{self.block_suffix}:html:html_{self.block_suffix}": {
+                self.taxonomy_1.id: list(self.library_html_tags1),
+                self.taxonomy_2.id: list(self.library_html_tags2),
+            },
+        }
 
 
 class TaggedCourseMixin(TestGetAllObjectTagsMixin, ModuleStoreTestCase):  # type: ignore[misc]

openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py

@@ -1884,7 +1884,7 @@ def test_export_course_invalid_id(self) -> None:
         url = OBJECT_TAGS_EXPORT_URL.format(object_id="invalid")
         self.client.force_authenticate(user=self.staff)
         response = self.client.get(url)
-        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.status_code == status.HTTP_404_NOT_FOUND
 
 
 @skip_unless_cms

openedx/core/djangoapps/content_tagging/rest_api/v1/views.py

@@ -13,10 +13,11 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
+from ...auth import has_view_object_tags_access
 
 from ...api import (
     create_taxonomy,
-    generate_csv,
+    generate_csv_rows,
     get_taxonomies,
     get_taxonomies_for_org,
     get_taxonomy,
@@ -164,11 +165,15 @@ def write(self, value):
         object_id: str = kwargs.get('context_id', None)
         pseudo_buffer = Echo()
 
+        if not has_view_object_tags_access(self.request.user, object_id):
+            raise PermissionDenied(
+                "You do not have permission to view object tags for this object_id."
+            )
+
         try:
             return StreamingHttpResponse(
-                streaming_content=generate_csv(
+                streaming_content=generate_csv_rows(
                     object_id,
-                    self.request.user,
                     pseudo_buffer,
                 ),
                 content_type="text/csv",