squash!: Added migration test, formatted code

Author: rodmgwgu

openedx_authz/engine/utils.py

@@ -8,6 +8,9 @@
 
 from casbin import Enforcer
 
+from openedx_authz.api.users import assign_role_to_user_in_scope, batch_assign_role_to_users_in_scope
+from openedx_authz.constants.roles import LIBRARY_ADMIN, LIBRARY_AUTHOR, LIBRARY_USER
+
 logger = logging.getLogger(__name__)
 
 GROUPING_POLICY_PTYPES = ["g", "g2", "g3", "g4", "g5", "g6"]
@@ -69,3 +72,82 @@ def migrate_policy_between_enforcers(
     except Exception as e:
         logger.error(f"Error loading policies from file: {e}")
         raise
+
+
+def migrate_legacy_permissions(ContentLibraryPermission):
+    """
+    Migrate legacy permission data to the new Casbin-based authorization model.
+    This function reads legacy permissions from the ContentLibraryPermission model
+    and assigns equivalent roles in the new authorization system.
+
+    The old Library permissions are stored in the ContentLibraryPermission model, it consists of the following columns:
+
+    - library: FK to ContentLibrary
+    - user: optional FK to User
+    - group: optional FK to Group
+    - access_level: 'admin' | 'author' | 'read'
+
+    In the new Authz model, this would roughly translate to:
+
+    - library: scope
+    - user: subject
+    - access_level: role
+
+    Now, we don't have an equivalent concept to "Group", for this we will go through the users in the group and assign
+    roles independently.
+
+    param ContentLibraryPermission: The ContentLibraryPermission model to use.
+    """
+
+    legacy_permissions = ContentLibraryPermission.objects.select_related(
+        "library", "library__org", "user", "group"
+    ).all()
+
+    # List to keep track of any permissions that could not be migrated
+    permissions_with_errors = []
+
+    for permission in legacy_permissions:
+        # Migrate the permission to the new model
+
+        # Derive equivalent role based on access level
+        access_level_to_role = {
+            "admin": LIBRARY_ADMIN,
+            "author": LIBRARY_AUTHOR,
+            "read": LIBRARY_USER,
+        }
+
+        role = access_level_to_role.get(permission.access_level)
+        if role is None:
+            # This should not happen as there are no more access_levels defined
+            # in ContentLibraryPermission, log and skip
+            logger.error(f"Unknown access level: {permission.access_level} for User: {permission.user}")
+            permissions_with_errors.append(permission)
+            continue
+
+        # Generating scope based on library identifier
+        scope = f"lib:{permission.library.org.name}:{permission.library.slug}"
+
+        if permission.group:
+            # Permission applied to a group
+            users = [user.username for user in permission.group.user_set.all()]
+            logger.info(
+                f"Migrating permissions for Users: {users} in Group: {permission.group.name} "
+                f"to Role: {role.external_key} in Scope: {scope}"
+            )
+            batch_assign_role_to_users_in_scope(
+                users=users, role_external_key=role.external_key, scope_external_key=scope
+            )
+        else:
+            # Permission applied to individual user
+            logger.info(
+                f"Migrating permission for User: {permission.user.username} "
+                f"to Role: {role.external_key} in Scope: {scope}"
+            )
+
+            assign_role_to_user_in_scope(
+                user_external_key=permission.user.username,
+                role_external_key=role.external_key,
+                scope_external_key=scope,
+            )
+
+    return permissions_with_errors

openedx_authz/migrations/0002_migrate_legacy_permissions.py

@@ -0,0 +1,61 @@
+# Generated by Django 5.2.7 on 2025-11-03 20:39
+
+import logging
+
+from django.db import migrations
+
+from openedx_authz.engine.utils import migrate_legacy_permissions
+
+logger = logging.getLogger(__name__)
+
+
+def _log_migration_errors(permissions_with_errors: list) -> None:
+    """
+    Log the permissions that could not be migrated during the migration process.
+    Args:
+        permissions_with_errors (list): List of ContentLibraryPermission instances that failed to migrate.
+    """
+    logger.error(
+        f"Migration completed with errors for {len(permissions_with_errors)} permissions.\n"
+        "The following permissions could not be migrated:"
+    )
+    for permission in permissions_with_errors:
+        logger.error(
+            "Access level: %s, %sLibrary: %s",
+            permission.access_level,
+            f"User: {permission.user.username}, " if permission.user else f"Group: {permission.group.name}, ",
+            permission.library.slug,
+        )
+
+
+def apply_migrate_legacy_permissions(apps, schema_editor):
+    """
+    Wrapper to run the migration using the historical version of the ContentLibraryPermission model.
+    """
+    # ContentLibraryPermission model from the content_libraries app, here is where the legacy permissions are stored
+    try:
+        ContentLibraryPermission = apps.get_model("content_libraries", "ContentLibraryPermission")
+    except LookupError:
+        # Don't run the migration where the content_libraries app is not installed, like during development.
+        logger.warning("ContentLibraryPermission model not found. Skipping migration.")
+        return
+
+    permissions_with_errors = migrate_legacy_permissions(ContentLibraryPermission)
+
+    if permissions_with_errors:
+        _log_migration_errors(permissions_with_errors)
+
+
+class Migration(migrations.Migration):
+    """
+    Migration to transfer legacy permissions from ContentLibraryPermission
+    to the new Casbin-based authorization model.
+    """
+
+    dependencies = [
+        ("openedx_authz", "0004_contentlibraryscope"),
+    ]
+
+    operations = [
+        migrations.RunPython(apply_migrate_legacy_permissions),
+    ]

openedx_authz/migrations/0005_migrate_legacy_permissions.py

@@ -44,6 +44,7 @@ def get_registry(cls) -> dict[str, type["BaseRegistryModel"]]:
         """
         return cls._registry
 
+
 class ScopeManager(models.Manager):
     """Custom manager for Scope model that handles polymorphic behavior."""
 

openedx_authz/models/core.py

@@ -968,7 +968,7 @@ def test_assign_role_creates_extended_casbin_rule(self):
         new_count = ExtendedCasbinRule.objects.count()
         self.assertEqual(new_count, initial_count + 1)
 
-        extended_rule = ExtendedCasbinRule.objects.order_by('-id').first()
+        extended_rule = ExtendedCasbinRule.objects.order_by("-id").first()
         self.assertIsNotNone(extended_rule)
         self.assertIsNotNone(extended_rule.casbin_rule)
         self.assertIsNotNone(extended_rule.subject)

openedx_authz/tests/api/test_roles.py

@@ -8,6 +8,19 @@
 from opaque_keys.edx.locator import LibraryLocatorV2
 
 
+class Organization(models.Model):
+    """Stub model representing an organization for testing purposes.
+
+    .. no_pii:
+    """
+
+    name = models.CharField(max_length=255)
+    short_name = models.CharField(max_length=100)
+
+    def __str__(self):
+        return str(self.name)
+
+
 class ContentLibraryManager(models.Manager):
     """Manager for ContentLibrary model with helper methods."""
 
@@ -38,9 +51,37 @@ class ContentLibrary(models.Model):
 
     locator = models.CharField(max_length=255, unique=True, db_index=True)
     title = models.CharField(max_length=255, blank=True, null=True)
+    slug = models.SlugField(allow_unicode=True)
+    org = models.ForeignKey(Organization, on_delete=models.PROTECT, null=True)
     created_at = models.DateTimeField(auto_now_add=True)
 
     objects = ContentLibraryManager()
 
     def __str__(self):
-        return self.locator
+        return str(self.locator)
+
+
+# Legacy permission models for testing purposes
+class ContentLibraryPermission(models.Model):
+    """Stub model representing legacy content library permissions for testing purposes.
+
+    .. no_pii:
+    """
+
+    ADMIN_LEVEL = "admin"
+    AUTHOR_LEVEL = "author"
+    READ_LEVEL = "read"
+    ACCESS_LEVEL_CHOICES = (
+        (ADMIN_LEVEL, "Administer users and author content"),
+        (AUTHOR_LEVEL, "Author content"),
+        (READ_LEVEL, "Read-only"),
+    )
+
+    library = models.ForeignKey(ContentLibrary, on_delete=models.CASCADE)
+    user = models.ForeignKey("auth.User", on_delete=models.CASCADE, null=True, blank=True)
+    group = models.ForeignKey("auth.Group", on_delete=models.CASCADE, null=True, blank=True)
+    access_level = models.CharField(max_length=30, choices=ACCESS_LEVEL_CHOICES)
+
+    def __str__(self):
+        who = self.user.username if self.user else self.group.name
+        return f"ContentLibraryPermission ({self.access_level} for {who})"