feat: support role assignment for scopes that don't exist yet

openedx_authz/handlers.py

@@ -11,6 +11,7 @@
 
 from casbin_adapter.models import CasbinRule
 from django.conf import settings
+from django.db import transaction
 from django.db.models.signals import post_delete, post_save
 from django.dispatch import receiver
 from openedx_events.authz.signals import ROLE_ASSIGNMENT_CREATED, ROLE_ASSIGNMENT_DELETED
@@ -20,6 +21,7 @@
 from openedx_authz.engine.utils import run_course_authoring_migration
 from openedx_authz.models.authz_migration import MigrationType, ScopeType
 from openedx_authz.models.core import ExtendedCasbinRule, RoleAssignmentAudit
+from openedx_authz.models.scopes import ContentLibrary, ContentLibraryScope, CourseOverview, CourseScope
 from openedx_authz.models.subjects import UserSubject
 
 try:
@@ -141,6 +143,52 @@ def handle_org_waffle_flag_change(sender, instance, **kwargs) -> None:
     post_save.connect(handle_org_waffle_flag_change, sender=WaffleFlagOrgOverrideModel)
 
 
+def backfill_course_scope(sender, instance, **kwargs):  # pylint: disable=unused-argument
+    """
+    Link a pending CourseScope to the CourseOverview that was just created for it.
+
+    get_or_create_for_external_key() (openedx_authz/models/scopes.py) allows assigning a
+    role for a course key before its CourseOverview exists, leaving course_overview null.
+    This backfills that FK once the course shows up, e.g. after a course rerun finishes
+    cloning. See CourseScope.link_pending_scope() for the actual lookup/update.
+
+    This runs on every CourseOverview save, so errors are logged rather than raised:
+    a problem backfilling a pending scope must not break the caller's save(). The query
+    runs in its own savepoint so a failure doesn't poison the caller's transaction.
+    """
+    try:
+        with transaction.atomic():
+            CourseScope.link_pending_scope(instance)
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        logger.exception(
+            "Error linking pending CourseScope to CourseOverview %s", instance.pk, exc_info=exc,
+        )
+
+
+def backfill_content_library_scope(sender, instance, **kwargs):  # pylint: disable=unused-argument
+    """
+    Link a pending ContentLibraryScope to the ContentLibrary that was just created for it.
+
+    See backfill_course_scope() above; same idea for content libraries, including the
+    broad exception handling and the isolating savepoint.
+    """
+    try:
+        with transaction.atomic():
+            ContentLibraryScope.link_pending_scope(instance)
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        logger.exception(
+            "Error linking pending ContentLibraryScope to ContentLibrary %s", instance.pk, exc_info=exc,
+        )
+
+
+# Only register the handlers if the models are available (i.e., running in Open edX)
+if CourseOverview is not None:
+    post_save.connect(backfill_course_scope, sender=CourseOverview)
+
+if ContentLibrary is not None:
+    post_save.connect(backfill_content_library_scope, sender=ContentLibrary)
+
+
 # Match ``WaffleFlagCourseOverrideModel.OVERRIDE_CHOICES`` / ``override_value`` in edx-platform:
 # the flag is effectively forced on only when the row is enabled and ``override_choice`` is "on".
 WAFFLE_OVERRIDE_FORCE_ON = "on"

openedx_authz/migrations/0010_scope_external_key.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.16 on 2026-07-29 19:03
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('openedx_authz', '0009_roleassignmentaudit'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='scope',
+            name='external_key',
+            field=models.CharField(blank=True, max_length=255, null=True, unique=True),
+        ),
+    ]

openedx_authz/models/core.py

@@ -111,15 +111,95 @@ class Scope(BaseRegistryModel):
     This model can be extended to represent different types of scopes,
     such as courses or content libraries.
 
-    Subclasses should define a NAMESPACE class attribute (e.g., 'lib' for content libraries)
-    and implement get_or_create_for_external_key() classmethod.
+    Subclasses should define a NAMESPACE class attribute (e.g., 'lib' for content libraries),
+    a LINKED_OBJECT_FIELD naming their FK to the backing object (e.g. 'content_library'), and
+    implement external_key_for_object().
     """
 
     objects = ScopeManager()
 
+    # Name of the subclass's FK field pointing at its backing object (e.g. "content_library",
+    # "course_overview"), used by get_or_create_for_external_key()/link_pending_scope() below
+    # to work generically across scope types.
+    LINKED_OBJECT_FIELD: ClassVar[str] = None
+
+    # Canonical string form of the scope's key (e.g. a course-v1 course id), set on creation
+    # regardless of whether the backing object (CourseOverview, ContentLibrary, ...) exists yet.
+    # This is the only way to find a scope back again when its FK to that object is still null
+    # so it can be linked up once the object is created (see link_pending_scope() below and the
+    # backfill signal receivers in openedx_authz/handlers.py).
+    external_key = models.CharField(max_length=255, null=True, blank=True, unique=True)
+
     class Meta:
         abstract = False
 
+    @classmethod
+    def external_key_for_object(cls, linked_object) -> str:
+        """Return the canonical external_key string for one of this scope's backing objects.
+
+        Subclasses must override this to match how their ScopeData computes external_key
+        (e.g. the course id, or the library key).
+
+        Args:
+            linked_object: An instance of the model named by LINKED_OBJECT_FIELD.
+
+        Returns:
+            str: The canonical external_key for that instance.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def get_or_create_for_external_key(cls, scope) -> "Scope":
+        """Get or create a scope instance for the given external key.
+
+        The backing object (CourseOverview, ContentLibrary, ...) need not exist yet (e.g.
+        during a course rerun, the destination course id is known before the course is
+        cloned): the scope is created with its LINKED_OBJECT_FIELD left ``None`` in that
+        case, and gets linked up automatically once a matching object is saved (see
+        link_pending_scope() below and the backfill signal receivers in
+        openedx_authz/handlers.py).
+
+        Args:
+            scope: ScopeData object with an external_key attribute.
+
+        Returns:
+            Scope: The (possibly newly created) scope instance for this subclass.
+        """
+        external_key = scope.external_key
+        linked_object = scope.get_object()
+        if linked_object is None:
+            # The object doesn't exist yet: key the row by its external_key so it can be
+            # found and linked up later by link_pending_scope().
+            scope, _ = cls.objects.get_or_create(external_key=external_key)
+            return scope
+
+        # Look up by the FK first (as before the external_key field existed) so scopes
+        # created before this change are reused rather than duplicated.
+        scope, created = cls.objects.get_or_create(
+            **{cls.LINKED_OBJECT_FIELD: linked_object},
+            defaults={"external_key": external_key},
+        )
+        if not created and not scope.external_key:
+            scope.external_key = external_key
+            scope.save(update_fields=["external_key"])
+        return scope
+
+    @classmethod
+    def link_pending_scope(cls, linked_object) -> None:
+        """Link a pending scope to the object that was just created for it.
+
+        Called from the relevant post_save signal receiver in openedx_authz/handlers.py
+        once an object with a matching external_key shows up.
+
+        Args:
+            linked_object: The model instance (CourseOverview, ContentLibrary, ...) that
+                was just saved.
+        """
+        cls.objects.filter(
+            external_key=cls.external_key_for_object(linked_object),
+            **{f"{cls.LINKED_OBJECT_FIELD}__isnull": True},
+        ).update(**{cls.LINKED_OBJECT_FIELD: linked_object})
+
 
 class Subject(BaseRegistryModel):
     """Model representing a subject in the authorization system.

openedx_authz/models/scopes.py

@@ -8,8 +8,6 @@
 from django.apps import apps
 from django.conf import settings
 from django.db import models
-from opaque_keys.edx.keys import CourseKey
-from opaque_keys.edx.locator import LibraryLocatorV2
 
 from openedx_authz.models.core import Scope
 
@@ -61,6 +59,7 @@ class ContentLibraryScope(Scope):
     """
 
     NAMESPACE = "lib"
+    LINKED_OBJECT_FIELD = "content_library"
 
     # Link to the actual content library, if applicable. In other cases, this could be null.
     # Piggybacking on the existing ContentLibrary model to keep the ExtendedCasbinRule up to date
@@ -81,21 +80,9 @@ class ContentLibraryScope(Scope):
     )
 
     @classmethod
-    def get_or_create_for_external_key(cls, scope) -> "ContentLibraryScope":
-        """Get or create a ContentLibraryScope for the given external key.
-
-        Args:
-            scope: ScopeData object with an external_key attribute containing
-                a LibraryLocatorV2-compatible string.
-
-        Returns:
-            ContentLibraryScope: The Scope instance for the given ContentLibrary,
-                or None if the scope is a glob pattern (contains wildcard).
-        """
-        library_key = LibraryLocatorV2.from_string(scope.external_key)
-        content_library = ContentLibrary.objects.get_by_key(library_key)
-        scope, _ = cls.objects.get_or_create(content_library=content_library)
-        return scope
+    def external_key_for_object(cls, linked_object) -> str:
+        """Return the canonical external_key string for a ContentLibrary instance."""
+        return str(linked_object.library_key)
 
 
 class CourseScope(Scope):
@@ -105,6 +92,7 @@ class CourseScope(Scope):
     """
 
     NAMESPACE = "course-v1"
+    LINKED_OBJECT_FIELD = "course_overview"
 
     # Link to the actual course, if applicable. In other cases, this could be null.
     # Piggybacking on the existing CourseOverview model to keep the ExtendedCasbinRule up to date
@@ -125,18 +113,6 @@ class CourseScope(Scope):
     )
 
     @classmethod
-    def get_or_create_for_external_key(cls, scope) -> "CourseScope":
-        """Get or create a CourseScope for the given external key.
-
-        Args:
-            scope: ScopeData object with an external_key attribute containing
-                a CourseKey string.
-
-        Returns:
-            CourseScope: The Scope instance for the given CourseOverview,
-                or None if the scope is a glob pattern (contains wildcard).
-        """
-        course_key = CourseKey.from_string(scope.external_key)
-        course_overview = CourseOverview.get_from_id(course_key)
-        scope, _ = cls.objects.get_or_create(course_overview=course_overview)
-        return scope
+    def external_key_for_object(cls, linked_object) -> str:
+        """Return the canonical external_key string for a CourseOverview instance."""
+        return str(linked_object.id)

openedx_authz/tests/integration/test_models.py

@@ -26,15 +26,20 @@
 from django.contrib.auth import get_user_model
 from django.db import IntegrityError
 from django.test import TestCase, override_settings
+from opaque_keys.edx.locator import CourseLocator
+from openedx.core.djangoapps.content.course_overviews.tests.factories import (  # pylint: disable=import-error
+    CourseOverviewFactory,
+)
 from organizations.api import ensure_organization
 from organizations.models import Organization
 
-from openedx_authz.api.data import ContentLibraryData, RoleData, SubjectData, UserData
+from openedx_authz.api.data import ContentLibraryData, CourseOverviewData, RoleData, SubjectData, UserData
 from openedx_authz.api.roles import assign_role_to_subject_in_scope
 from openedx_authz.engine.enforcer import AuthzEnforcer
 from openedx_authz.models import (
     ContentLibrary,
     ContentLibraryScope,
+    CourseScope,
     ExtendedCasbinRule,
     Scope,
     Subject,
@@ -1356,3 +1361,114 @@ def test_extended_casbin_rule_with_null_subject_deletion(self):
         self.assertFalse(CasbinRule.objects.filter(id=casbin_rule_id).exists())
 
         self.assertTrue(Scope.objects.filter(id=scope_id).exists())
+
+
+@pytest.mark.integration
+@override_settings(OPENEDX_AUTHZ_CONTENT_LIBRARY_MODEL="content_libraries.ContentLibrary")
+class TestScopeForPendingObjects(TestCase):
+    """Test cases for scopes whose backing object doesn't exist yet.
+
+    Covers https://github.com/openedx/openedx-authz/issues/352: role assignment must succeed
+    for a course/library key before the CourseOverview/ContentLibrary behind it is created (e.g.
+    during a course rerun, the destination course id is known before the course is cloned), and
+    the Scope must link up to the real object once it's created (see the backfill signal
+    receivers in openedx_authz/handlers.py).
+    """
+
+    def test_course_scope_created_without_course_overview(self):
+        """Assigning a role for a course that doesn't exist yet must not raise."""
+        course_key = CourseLocator("PendingOrg", "PendingCourse", "2024_T1")
+        scope_data = CourseOverviewData(external_key=str(course_key))
+
+        scope = Scope.objects.get_or_create_for_external_key(scope_data)
+
+        self.assertIsInstance(scope, CourseScope)
+        self.assertIsNone(scope.course_overview)
+        self.assertEqual(scope.external_key, str(course_key))
+
+    def test_course_scope_backfills_when_course_overview_created(self):
+        """Creating the CourseOverview later links up the previously-pending CourseScope."""
+        course_key = CourseLocator("PendingOrg2", "PendingCourse2", "2024_T1")
+        scope_data = CourseOverviewData(external_key=str(course_key))
+        scope = Scope.objects.get_or_create_for_external_key(scope_data)
+        self.assertIsNone(scope.course_overview)
+
+        course_overview = CourseOverviewFactory(id=course_key)
+
+        scope.refresh_from_db()
+        self.assertEqual(scope.course_overview, course_overview)
+
+    def test_course_scope_get_or_create_is_idempotent_while_pending(self):
+        """Calling get_or_create_for_external_key twice before the course exists reuses the same row."""
+        course_key = CourseLocator("PendingOrg3", "PendingCourse3", "2024_T1")
+        scope_data = CourseOverviewData(external_key=str(course_key))
+
+        scope1 = Scope.objects.get_or_create_for_external_key(scope_data)
+        scope2 = Scope.objects.get_or_create_for_external_key(scope_data)
+
+        self.assertEqual(scope1.id, scope2.id)
+        self.assertEqual(CourseScope.objects.filter(external_key=str(course_key)).count(), 1)
+
+    def test_assign_role_succeeds_for_course_that_does_not_exist_yet(self):
+        """The public API assign_role_to_subject_in_scope must not crash for a not-yet-existing course."""
+        test_username = "pending_course_instructor"
+        User.objects.create_user(username=test_username)
+        course_key = CourseLocator("PendingOrg4", "PendingCourse4", "2024_T1")
+
+        subject_data = UserData(external_key=test_username)
+        role_data = RoleData(external_key="instructor")
+        scope_data = CourseOverviewData(external_key=str(course_key))
+
+        result = assign_role_to_subject_in_scope(subject_data, role_data, scope_data)
+
+        self.assertTrue(result)
+        scope = CourseScope.objects.get(external_key=str(course_key))
+        self.assertIsNone(scope.course_overview)
+
+        # The course gets created later (e.g. once course rerun finishes cloning).
+        course_overview = CourseOverviewFactory(id=course_key)
+        scope.refresh_from_db()
+        self.assertEqual(scope.course_overview, course_overview)
+
+    def test_content_library_scope_created_without_content_library(self):
+        """Assigning a role for a library that doesn't exist yet must not raise."""
+        library_key = "lib:PendingOrg:pendinglib"
+        scope_data = ContentLibraryData(external_key=library_key)
+
+        scope = Scope.objects.get_or_create_for_external_key(scope_data)
+
+        self.assertIsInstance(scope, ContentLibraryScope)
+        self.assertIsNone(scope.content_library)
+        self.assertEqual(scope.external_key, library_key)
+
+    def test_content_library_scope_backfills_when_library_created(self):
+        """Creating the ContentLibrary later links up the previously-pending ContentLibraryScope."""
+        org_short_name = "PendingLibOrg"
+        slug = "pendinglib2"
+        library_key = f"lib:{org_short_name}:{slug}"
+        scope_data = ContentLibraryData(external_key=library_key)
+        scope = Scope.objects.get_or_create_for_external_key(scope_data)
+        self.assertIsNone(scope.content_library)
+
+        _, _, content_library = create_test_library(org_short_name=org_short_name, slug=slug)
+
+        scope.refresh_from_db()
+        self.assertEqual(scope.content_library, content_library)
+
+    def test_scope_created_before_external_key_field_is_reused_not_duplicated(self):
+        """A Scope row created before this fix (FK set, no external_key) isn't duplicated on reassignment.
+
+        Simulates a pre-existing row the way get_or_create_for_external_key used to create it:
+        keyed only by the FK, with no external_key.
+        """
+        course_key = CourseLocator("LegacyOrg", "LegacyCourse", "2024_T1")
+        course_overview = CourseOverviewFactory(id=course_key)
+        legacy_scope = CourseScope.objects.create(course_overview=course_overview)
+        self.assertIsNone(legacy_scope.external_key)
+
+        scope_data = CourseOverviewData(external_key=str(course_key))
+        scope = Scope.objects.get_or_create_for_external_key(scope_data)
+
+        self.assertEqual(scope.id, legacy_scope.id)
+        self.assertEqual(scope.external_key, str(course_key))
+        self.assertEqual(CourseScope.objects.filter(course_overview=course_overview).count(), 1)