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

openedx_authz/handlers.py

@@ -20,6 +20,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 +142,39 @@ 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.
+
+    CourseScope.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 and external_key set to the course id. This backfills that FK
+    once the course shows up, e.g. after a course rerun finishes cloning.
+    """
+    CourseScope.objects.filter(
+        external_key=str(instance.id), course_overview__isnull=True
+    ).update(course_overview=instance)
+
+
+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.
+    """
+    ContentLibraryScope.objects.filter(
+        external_key=str(instance.library_key), content_library__isnull=True
+    ).update(content_library=instance)
+
+
+# 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-28 00:04
+
+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, db_index=True, max_length=255, null=True, unique=True),
+        ),
+    ]

openedx_authz/models/core.py

@@ -117,6 +117,13 @@ class Scope(BaseRegistryModel):
 
     objects = ScopeManager()
 
+    # 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
+    # (see get_or_create_for_external_key() in openedx_authz/models/scopes.py) so it can be
+    # linked up once the object is created (see openedx_authz/handlers.py backfill receivers).
+    external_key = models.CharField(max_length=255, null=True, blank=True, unique=True, db_index=True)
+
     class Meta:
         abstract = False
 

openedx_authz/models/scopes.py

@@ -84,6 +84,12 @@ class ContentLibraryScope(Scope):
     def get_or_create_for_external_key(cls, scope) -> "ContentLibraryScope":
         """Get or create a ContentLibraryScope for the given external key.
 
+        The backing ContentLibrary need not exist yet (e.g. it may be created
+        later as part of an in-progress operation): the scope is created with
+        ``content_library=None`` in that case and gets linked up automatically
+        once a ContentLibrary with a matching key is saved (see the backfill
+        signal receiver in openedx_authz/handlers.py).
+
         Args:
             scope: ScopeData object with an external_key attribute containing
                 a LibraryLocatorV2-compatible string.
@@ -93,8 +99,23 @@ def get_or_create_for_external_key(cls, scope) -> "ContentLibraryScope":
                 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)
+        try:
+            content_library = ContentLibrary.objects.get_by_key(library_key)
+        except ContentLibrary.DoesNotExist:
+            # The library doesn't exist yet: key the row by its external_key so it can be
+            # found and linked up later by the backfill signal in openedx_authz/handlers.py.
+            scope, _ = cls.objects.get_or_create(external_key=str(library_key))
+            return scope
+
+        # Look up by the FK first (as before this change) so scopes created before the
+        # external_key field existed are reused rather than duplicated.
+        scope, created = cls.objects.get_or_create(
+            content_library=content_library,
+            defaults={"external_key": str(library_key)},
+        )
+        if not created and not scope.external_key:
+            scope.external_key = str(library_key)
+            scope.save(update_fields=["external_key"])
         return scope
 
 
@@ -128,6 +149,13 @@ class CourseScope(Scope):
     def get_or_create_for_external_key(cls, scope) -> "CourseScope":
         """Get or create a CourseScope for the given external key.
 
+        The backing CourseOverview 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 ``course_overview=None`` in that
+        case and gets linked up automatically once a CourseOverview with a
+        matching id is saved (see the backfill signal receiver in
+        openedx_authz/handlers.py).
+
         Args:
             scope: ScopeData object with an external_key attribute containing
                 a CourseKey string.
@@ -137,6 +165,21 @@ def get_or_create_for_external_key(cls, scope) -> "CourseScope":
                 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)
+        try:
+            course_overview = CourseOverview.get_from_id(course_key)
+        except CourseOverview.DoesNotExist:
+            # The course doesn't exist yet: key the row by its external_key so it can be
+            # found and linked up later by the backfill signal in openedx_authz/handlers.py.
+            scope, _ = cls.objects.get_or_create(external_key=str(course_key))
+            return scope
+
+        # Look up by the FK first (as before this change) so scopes created before the
+        # external_key field existed are reused rather than duplicated.
+        scope, created = cls.objects.get_or_create(
+            course_overview=course_overview,
+            defaults={"external_key": str(course_key)},
+        )
+        if not created and not scope.external_key:
+            scope.external_key = str(course_key)
+            scope.save(update_fields=["external_key"])
         return scope

openedx_authz/tests/integration/test_models.py

@@ -29,12 +29,19 @@
 from organizations.api import ensure_organization
 from organizations.models import Organization
 
-from openedx_authz.api.data import ContentLibraryData, RoleData, SubjectData, UserData
+from opaque_keys.edx.locator import CourseLocator
+from openedx.core.djangoapps.content.course_overviews.tests.factories import (  # pylint: disable=import-error
+    CourseOverviewFactory,
+)
+
+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,
+    CourseOverview,
+    CourseScope,
     ExtendedCasbinRule,
     Scope,
     Subject,
@@ -1356,3 +1363,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)