Create license audit task and special permissions model

contentcuration/contentcuration/migrations/0158_add_audited_special_permissions_license.py

@@ -0,0 +1,39 @@
+# Generated by Django 3.2.24 on 2025-11-05 22:17
+import uuid
+
+from django.db import migrations
+from django.db import models
+
+import contentcuration.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("contentcuration", "0157_merge_20251015_0333"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="AuditedSpecialPermissionsLicense",
+            fields=[
+                (
+                    "id",
+                    contentcuration.models.UUIDField(
+                        default=uuid.uuid4,
+                        max_length=32,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("description", models.TextField(db_index=True, unique=True)),
+                ("distributable", models.BooleanField(default=False)),
+            ],
+        ),
+        migrations.AddIndex(
+            model_name="auditedspecialpermissionslicense",
+            index=models.Index(
+                fields=["description"], name="audited_special_perms_desc_idx"
+            ),
+        ),
+    ]

contentcuration/contentcuration/models.py

@@ -2708,6 +2708,29 @@ class Meta:
         ]
 
 
+class AuditedSpecialPermissionsLicense(models.Model):
+    """
+    Stores special permission license descriptions that have been audited
+    for community library submissions. When a channel contains resources with
+    "Special Permissions" licenses, their license descriptions are stored here
+    for admin review.
+    """
+
+    id = UUIDField(primary_key=True, default=uuid.uuid4)
+    description = models.TextField(unique=True, db_index=True)
+    distributable = models.BooleanField(default=False)
+
+    def __str__(self):
+        return (
+            self.description[:100] if len(self.description) > 100 else self.description
+        )
+
+    class Meta:
+        indexes = [
+            models.Index(fields=["description"], name="audited_special_perms_desc_idx"),
+        ]
+
+
 ASSESSMENT_ID_INDEX_NAME = "assessment_id_idx"
 
 

contentcuration/contentcuration/tasks.py

@@ -16,6 +16,7 @@
 from contentcuration.models import Change
 from contentcuration.models import ContentNode
 from contentcuration.models import User
+from contentcuration.utils.audit_channel_licenses import audit_channel_licenses
 from contentcuration.utils.csv_writer import write_user_csv
 from contentcuration.utils.nodes import calculate_resource_size
 from contentcuration.utils.nodes import generate_diff
@@ -152,7 +153,7 @@ def sendcustomemails_task(subject, message, query):
         text = message.format(
             current_date=time.strftime("%A, %B %d"),
             current_time=time.strftime("%H:%M %Z"),
-            **recipient.__dict__
+            **recipient.__dict__,
         )
         text = render_to_string("registration/custom_email.txt", {"message": text})
         recipient.email_user(
@@ -165,3 +166,8 @@ def sendcustomemails_task(subject, message, query):
 @app.task(name="ensure_versioned_database_exists_task")
 def ensure_versioned_database_exists_task(channel_id, channel_version):
     ensure_versioned_database_exists(channel_id, channel_version)
+
+
+@app.task(name="audit-channel-licenses")
+def audit_channel_licenses_task(channel_id, user_id):
+    audit_channel_licenses(channel_id, user_id)

contentcuration/contentcuration/tests/test_asynctask.py

@@ -2,16 +2,22 @@
 import time
 import uuid
 
+import mock
 import pytest
 from celery import states
 from celery.result import allow_join_result
 from celery.utils.log import get_task_logger
 from django.core.management import call_command
 from django.test import TransactionTestCase
 from django_celery_results.models import TaskResult
+from le_utils.constants import licenses
+from mock import patch
 
 from . import testdata
+from .base import StudioTestCase
 from .helpers import clear_tasks
+from .helpers import EagerTasksTestMixin
+from contentcuration import models as cc
 from contentcuration.celery import app
 
 logger = get_task_logger(__name__)
@@ -273,3 +279,229 @@ def test_revoke_task(self):
             TaskResult.objects.get(task_id=async_result.task_id, status=states.REVOKED)
         except TaskResult.DoesNotExist:
             self.fail("Missing revoked task result")
+
+
+class AuditChannelLicensesTaskTestCase(EagerTasksTestMixin, StudioTestCase):
+    """Tests for the audit_channel_licenses_task"""
+
+    def setUp(self):
+        super().setUp()
+        self.setUpBase()
+        self.channel.main_tree.published = True
+        self.channel.main_tree.save()
+        self.channel.version = 1
+        self.channel.save()
+
+    @patch("contentcuration.utils.audit_channel_licenses.KolibriContentNode")
+    @patch(
+        "contentcuration.utils.audit_channel_licenses.using_temp_migrated_content_database"
+    )
+    @patch("contentcuration.utils.audit_channel_licenses.storage.exists")
+    def test_audit_licenses_task__no_invalid_or_special_permissions(
+        self, mock_storage_exists, mock_using_db, mock_kolibri_node
+    ):
+        """Test audit task when channel has no invalid or special permissions licenses"""
+        from contentcuration.tasks import audit_channel_licenses_task
+
+        license1, _ = cc.License.objects.get_or_create(license_name="CC BY")
+        license2, _ = cc.License.objects.get_or_create(license_name="CC BY-SA")
+        cc.License.objects.get_or_create(license_name=licenses.SPECIAL_PERMISSIONS)
+        node1 = testdata.node({"kind_id": "video", "title": "Video Node"})
+        node1.parent = self.channel.main_tree
+        node1.license = license1
+        node1.save()
+        node1.published = True
+        node1.save()
+
+        node2 = testdata.node({"kind_id": "video", "title": "Video Node 2"})
+        node2.parent = self.channel.main_tree
+        node2.license = license2
+        node2.save()
+        node2.published = True
+        node2.save()
+
+        mock_storage_exists.return_value = True
+
+        mock_context = mock.MagicMock()
+        mock_using_db.return_value.__enter__ = mock.Mock(return_value=mock_context)
+        mock_using_db.return_value.__exit__ = mock.Mock(return_value=None)
+
+        # Mock KolibriContentNode to return license names from the nodes we created
+        mock_license_names_distinct = ["CC BY", "CC BY-SA"]
+        mock_license_names_values_list = mock.Mock()
+        mock_license_names_values_list.distinct.return_value = (
+            mock_license_names_distinct
+        )
+        mock_license_names_exclude3 = mock.Mock()
+        mock_license_names_exclude3.values_list.return_value = (
+            mock_license_names_values_list
+        )
+        mock_license_names_exclude2 = mock.Mock()
+        mock_license_names_exclude2.exclude.return_value = mock_license_names_exclude3
+        mock_license_names_exclude1 = mock.Mock()
+        mock_license_names_exclude1.exclude.return_value = mock_license_names_exclude2
+
+        mock_kolibri_node.objects = mock.Mock()
+        mock_kolibri_node.objects.exclude = mock.Mock(
+            return_value=mock_license_names_exclude1
+        )
+
+        audit_channel_licenses_task.apply(
+            kwargs={"channel_id": self.channel.id, "user_id": self.user.id}
+        )
+
+        self.channel.refresh_from_db()
+        version_str = str(self.channel.version)
+        self.assertIn(version_str, self.channel.published_data)
+        published_data_version = self.channel.published_data[version_str]
+
+        self.assertIn("included_licenses", published_data_version)
+        self.assertIsNone(
+            published_data_version.get("community_library_invalid_licenses")
+        )
+        self.assertIsNone(
+            published_data_version.get("community_library_special_permissions")
+        )
+
+    @patch("contentcuration.utils.audit_channel_licenses.KolibriContentNode")
+    @patch(
+        "contentcuration.utils.audit_channel_licenses.using_temp_migrated_content_database"
+    )
+    @patch("contentcuration.utils.audit_channel_licenses.storage.exists")
+    def test_audit_licenses_task__with_all_rights_reserved(
+        self, mock_storage_exists, mock_using_db, mock_kolibri_node
+    ):
+        """Test audit task when channel has All Rights Reserved license"""
+        from contentcuration.tasks import audit_channel_licenses_task
+
+        all_rights_license, _ = cc.License.objects.get_or_create(
+            license_name=licenses.ALL_RIGHTS_RESERVED
+        )
+
+        mock_storage_exists.return_value = True
+
+        mock_context = mock.MagicMock()
+        mock_using_db.return_value.__enter__ = mock.Mock(return_value=mock_context)
+        mock_using_db.return_value.__exit__ = mock.Mock(return_value=None)
+
+        mock_license_names_distinct = [licenses.ALL_RIGHTS_RESERVED]
+        mock_license_names_values_list = mock.Mock()
+        mock_license_names_values_list.distinct.return_value = (
+            mock_license_names_distinct
+        )
+        mock_license_names_exclude3 = mock.Mock()
+        mock_license_names_exclude3.values_list.return_value = (
+            mock_license_names_values_list
+        )
+        mock_license_names_exclude2 = mock.Mock()
+        mock_license_names_exclude2.exclude.return_value = mock_license_names_exclude3
+        mock_license_names_exclude1 = mock.Mock()
+        mock_license_names_exclude1.exclude.return_value = mock_license_names_exclude2
+        mock_license_names_base = mock.Mock()
+        mock_license_names_base.exclude.return_value = mock_license_names_exclude1
+
+        mock_kolibri_node.objects = mock.Mock()
+        mock_kolibri_node.objects.exclude = mock.Mock(
+            return_value=mock_license_names_exclude1
+        )
+
+        audit_channel_licenses_task.apply(
+            kwargs={"channel_id": self.channel.id, "user_id": self.user.id}
+        )
+
+        self.channel.refresh_from_db()
+        version_str = str(self.channel.version)
+        published_data_version = self.channel.published_data[version_str]
+
+        self.assertEqual(
+            published_data_version.get("community_library_invalid_licenses"),
+            [all_rights_license.id],
+        )
+
+    @patch("contentcuration.utils.audit_channel_licenses.KolibriContentNode")
+    @patch(
+        "contentcuration.utils.audit_channel_licenses.using_temp_migrated_content_database"
+    )
+    @patch("contentcuration.utils.audit_channel_licenses.storage.exists")
+    def test_audit_licenses_task__with_special_permissions(
+        self, mock_storage_exists, mock_using_db, mock_kolibri_node
+    ):
+        """Test audit task when channel has Special Permissions licenses"""
+        from contentcuration.tasks import audit_channel_licenses_task
+
+        special_perms_license, _ = cc.License.objects.get_or_create(
+            license_name="Special Permissions"
+        )
+        node = testdata.node({"kind_id": "video", "title": "Video Node"})
+        node.parent = self.channel.main_tree
+        node.license = special_perms_license
+        node.save()
+        node.published = True
+        node.save()
+
+        mock_storage_exists.return_value = True
+
+        mock_context = mock.MagicMock()
+        mock_using_db.return_value.__enter__ = mock.Mock(return_value=mock_context)
+        mock_using_db.return_value.__exit__ = mock.Mock(return_value=None)
+
+        mock_license_names_distinct = [licenses.SPECIAL_PERMISSIONS]
+        mock_license_names_values_list = mock.Mock()
+        mock_license_names_values_list.distinct.return_value = (
+            mock_license_names_distinct
+        )
+        mock_license_names_exclude3 = mock.Mock()
+        mock_license_names_exclude3.values_list.return_value = (
+            mock_license_names_values_list
+        )
+        mock_license_names_exclude2 = mock.Mock()
+        mock_license_names_exclude2.exclude.return_value = mock_license_names_exclude3
+        mock_license_names_exclude1 = mock.Mock()
+        mock_license_names_exclude1.exclude.return_value = mock_license_names_exclude2
+        mock_license_names_base = mock.Mock()
+        mock_license_names_base.exclude.return_value = mock_license_names_exclude1
+        mock_special_perms_distinct = ["Custom permission 1", "Custom permission 2"]
+        mock_special_perms_values_list = mock.Mock()
+        mock_special_perms_values_list.distinct.return_value = (
+            mock_special_perms_distinct
+        )
+        mock_special_perms_exclude3 = mock.Mock()
+        mock_special_perms_exclude3.values_list.return_value = (
+            mock_special_perms_values_list
+        )
+        mock_special_perms_exclude2 = mock.Mock()
+        mock_special_perms_exclude2.exclude.return_value = mock_special_perms_exclude3
+        mock_special_perms_exclude1 = mock.Mock()
+        mock_special_perms_exclude1.exclude.return_value = mock_special_perms_exclude2
+        mock_special_perms_filter = mock.Mock()
+        mock_special_perms_filter.exclude.return_value = mock_special_perms_exclude1
+
+        # Set up the mock to return different querysets based on the method called
+        mock_kolibri_node.objects = mock.Mock()
+        mock_kolibri_node.objects.exclude = mock.Mock(
+            return_value=mock_license_names_exclude1
+        )
+        mock_kolibri_node.objects.filter = mock.Mock(
+            return_value=mock_special_perms_filter
+        )
+
+        audit_channel_licenses_task.apply(
+            kwargs={"channel_id": self.channel.id, "user_id": self.user.id}
+        )
+
+        self.channel.refresh_from_db()
+        version_str = str(self.channel.version)
+        published_data_version = self.channel.published_data[version_str]
+
+        special_perms = published_data_version.get(
+            "community_library_special_permissions"
+        )
+        self.assertIsNotNone(special_perms)
+        self.assertEqual(len(special_perms), 2)
+
+        from contentcuration.models import AuditedSpecialPermissionsLicense
+
+        audited_licenses = AuditedSpecialPermissionsLicense.objects.filter(
+            description__in=["Custom permission 1", "Custom permission 2"]
+        )
+        self.assertEqual(audited_licenses.count(), 2)