fix: Cannot rerun courses - authz role assignment expects rerun to already exist

cms/djangoapps/contentstore/tasks.py

@@ -56,6 +56,7 @@
 from cms.djangoapps.contentstore.toggles import enable_course_optimizer_check_prev_run_links
 from cms.djangoapps.contentstore.utils import (
     IMPORTABLE_FILE_TYPES,
+    add_instructor,
     contains_course_reference,
     create_course_info_usage_key,
     create_or_update_xblock_upstream_link,
@@ -188,7 +189,9 @@ def rerun_course(source_course_key_string, destination_course_key_string, user_i
         update_unit_discussion_state_from_discussion_blocks(destination_course_key, user_id)
 
         # set initial permissions for the user to access the course.
-        initialize_permissions(destination_course_key, User.objects.get(id=user_id))
+        user = User.objects.get(id=user_id)
+        add_instructor(destination_course_key, user, user)
+        initialize_permissions(destination_course_key, user)
 
         # update state: Succeeded
         CourseRerunState.objects.succeeded(course_key=destination_course_key)

cms/djangoapps/contentstore/tests/test_clone_course.py

@@ -14,6 +14,7 @@
 from common.djangoapps.course_action_state.managers import CourseRerunUIStateManager
 from common.djangoapps.course_action_state.models import CourseRerunState
 from common.djangoapps.student.auth import has_course_author_access
+from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
 from xmodule.contentstore.content import StaticContent  # pylint: disable=wrong-import-order
 from xmodule.contentstore.django import contentstore  # pylint: disable=wrong-import-order
 from xmodule.modulestore import EdxJSONEncoder, ModuleStoreEnum  # pylint: disable=wrong-import-order
@@ -141,3 +142,47 @@ def test_rerun_course(self):
                 course_key=split_course4_id,
                 state=CourseRerunUIStateManager.State.FAILED
             )
+
+
+    def test_rerun_course_grants_instructor_access(self):
+        """
+        Test that the rerun_course task grants instructor and staff access
+        to the user after cloning. This verifies add_instructor is called
+        inside the task (needed when authz.enable_course_authoring is enabled
+        and add_instructor cannot be called pre-task).
+        """
+        org = 'edX'
+        course_number = 'CS101'
+        course_run = '2025_Q1'
+        display_name = 'rerun_instructor_test'
+        fields = {'display_name': display_name}
+
+        # Create a source course
+        source_course = CourseFactory.create(
+            org=org,
+            number=course_number,
+            run=course_run,
+            display_name=display_name,
+            default_store=ModuleStoreEnum.Type.split,
+        )
+
+        dest_course_id = CourseLocator(org=org, course=course_number, run="instructor_rerun")
+        CourseRerunState.objects.initiated(
+            source_course.id, dest_course_id, self.user, fields['display_name']
+        )
+
+        result = rerun_course.delay(
+            str(source_course.id),
+            str(dest_course_id),
+            self.user.id,
+            json.dumps(fields, cls=EdxJSONEncoder),
+        )
+        self.assertEqual(result.get(), "succeeded")  # noqa: PT009
+
+        # Verify the user has instructor and staff access on the new course
+        self.assertTrue(  # noqa: PT009
+            has_course_author_access(self.user, dest_course_id),
+            "User should have author access after rerun task completes",
+        )
+        self.assertTrue(CourseInstructorRole(dest_course_id).has_user(self.user))  # noqa: PT009
+        self.assertTrue(CourseStaffRole(dest_course_id).has_user(self.user))  # noqa: PT009

cms/djangoapps/contentstore/tests/test_course_create_rerun.py

@@ -14,16 +14,23 @@
 from django.test.client import RequestFactory
 from django.urls import reverse
 from opaque_keys.edx.keys import CourseKey
+from opaque_keys.edx.locator import CourseLocator
 from openedx_authz.constants.roles import COURSE_EDITOR
 from organizations.api import add_organization, get_course_organizations, get_organization_by_short_name
 from organizations.exceptions import InvalidOrganizationException
 from organizations.models import Organization
 
 from cms.djangoapps.contentstore.tests.utils import AjaxEnabledTestClient, parse_json
-from cms.djangoapps.contentstore.views.course import get_allowed_organizations, user_can_create_organizations
+from cms.djangoapps.contentstore.views.course import (
+    get_allowed_organizations,
+    get_in_process_course_actions,
+    user_can_create_organizations,
+)
 from cms.djangoapps.course_creators.admin import CourseCreatorAdmin
 from cms.djangoapps.course_creators.models import CourseCreator
-from common.djangoapps.student.auth import update_org_role
+from common.djangoapps.course_action_state.models import CourseRerunState
+from common.djangoapps.student.auth import has_course_author_access, update_org_role
+from common.djangoapps.student.models import CourseAccessRole
 from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole, OrgContentCreatorRole
 from common.djangoapps.student.tests.factories import AdminFactory, UserFactory
 from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin
@@ -491,3 +498,228 @@ def test_create_course_disabled_by_flag(self):
         })
 
         assert response.status_code == 403
+
+
+class TestCourseRerunAuthz(
+    CourseAuthoringAuthzTestMixin,
+    ModuleStoreTestCase,
+):
+    """
+    Tests for course rerun behavior when authz.enable_course_authoring is enabled.
+
+    Verifies that:
+    - Rerun succeeds without calling add_instructor pre-task (which would fail
+      because CourseOverview doesn't exist yet).
+    - add_instructor is called in the task after clone_course, when CourseOverview
+      can be resolved.
+    - The initiating user can see in-process rerun status via created_user check.
+    """
+
+    def setUp(self):
+        super().setUp()
+
+        self.url = reverse("course_handler")
+
+        # Create a source course
+        self.source_course = CourseFactory.create(
+            org='testorg',
+            number='101',
+            run='2025_Spring',
+            display_name='Source Course',
+        )
+        self.source_course_key = self.source_course.id
+
+        # Give the staff user instructor/staff access to the source course
+        for role in [CourseInstructorRole, CourseStaffRole]:
+            role(self.source_course_key).add_users(self.staff_user)
+
+        self.staff_client = AjaxEnabledTestClient()
+        self.staff_client.login(
+            username=self.staff_user.username,
+            password=self.password,
+        )
+
+    def test_rerun_succeeds_with_authz_enabled(self):
+        """
+        Test that course rerun completes successfully when authz.enable_course_authoring
+        is enabled. Previously this would fail with CourseOverview.DoesNotExist because
+        add_instructor was called before the course was cloned.
+        """
+        response = self.staff_client.ajax_post(self.url, {
+            'source_course_key': str(self.source_course_key),
+            'org': self.source_course_key.org,
+            'course': self.source_course_key.course,
+            'run': 'rerun_authz',
+            'display_name': 'Rerun with AuthZ',
+        })
+
+        assert response.status_code == 200
+        data = parse_json(response)
+        dest_course_key = CourseKey.from_string(data['destination_course_key'])
+
+        # Verify the rerun completed (task runs eagerly in tests)
+        dest_course = self.store.get_course(dest_course_key)
+        assert dest_course is not None
+        assert dest_course.display_name == 'Rerun with AuthZ'
+
+    def test_rerun_does_not_create_legacy_roles_with_authz_enabled(self):
+        """
+        Test that when authz.enable_course_authoring is enabled, the rerun does not
+        create legacy CourseAccessRole records pre-task. Legacy roles should not exist
+        when authz is enabled to avoid database inconsistencies.
+        """
+        response = self.staff_client.ajax_post(self.url, {
+            'source_course_key': str(self.source_course_key),
+            'org': self.source_course_key.org,
+            'course': self.source_course_key.course,
+            'run': 'rerun_no_legacy',
+            'display_name': 'Rerun No Legacy Roles',
+        })
+
+        assert response.status_code == 200
+        data = parse_json(response)
+        dest_course_key = CourseKey.from_string(data['destination_course_key'])
+
+        # Verify no legacy CourseAccessRole records were created for the destination course.
+        # When authz is enabled, permissions are managed entirely through the authz layer.
+        legacy_roles = CourseAccessRole.objects.filter(
+            user=self.staff_user,
+            course_id=dest_course_key,
+        )
+        assert not legacy_roles.exists()
+
+    def test_rerun_grants_authz_permissions_after_clone(self):
+        """
+        Test that after a successful rerun with authz enabled, the user has proper
+        permissions via the authz layer (add_instructor is called in the task after
+        clone_course completes and CourseOverview is available).
+        """
+        response = self.staff_client.ajax_post(self.url, {
+            'source_course_key': str(self.source_course_key),
+            'org': self.source_course_key.org,
+            'course': self.source_course_key.course,
+            'run': 'rerun_perms',
+            'display_name': 'Rerun Permissions Test',
+        })
+
+        assert response.status_code == 200
+        data = parse_json(response)
+        dest_course_key = CourseKey.from_string(data['destination_course_key'])
+
+        # Verify the user has author access to the new course via the role system
+        assert has_course_author_access(self.staff_user, dest_course_key)
+
+    def test_in_process_rerun_visible_to_initiating_user(self):
+        """
+        Test that the user who initiated the rerun can see the in-process status
+        via the created_user check in get_in_process_course_actions, even when
+        authz permissions haven't been fully established yet.
+        """
+        dest_course_key = CourseLocator(org='testorg', course='101', run='in_progress_rerun')
+        CourseRerunState.objects.initiated(
+            self.source_course_key, dest_course_key, self.staff_user, 'In Progress Rerun'
+        )
+
+        # Build a request for the staff user who initiated the rerun
+        request = RequestFactory().get('/')
+        request.user = self.staff_user
+
+        # The user should see the in-process action even without explicit course permissions
+        in_process_actions = get_in_process_course_actions(request)
+        action_course_keys = [action.course_key for action in in_process_actions]
+        assert dest_course_key in action_course_keys
+
+    def test_in_process_rerun_not_visible_to_other_users(self):
+        """
+        Test that a user who did NOT initiate the rerun cannot see the in-process
+        status (unless they have course permissions).
+        """
+        dest_course_key = CourseLocator(org='testorg', course='101', run='other_user_rerun')
+        CourseRerunState.objects.initiated(
+            self.source_course_key, dest_course_key, self.staff_user, 'Other User Rerun'
+        )
+
+        # Build a request for a different user who did not initiate the rerun
+        request = RequestFactory().get('/')
+        request.user = self.unauthorized_user
+
+        # The other user should NOT see the in-process action
+        in_process_actions = get_in_process_course_actions(request)
+        action_course_keys = [action.course_key for action in in_process_actions]
+        assert dest_course_key not in action_course_keys
+
+
+class TestCourseRerunLegacy(ModuleStoreTestCase):
+    """
+    Tests for course rerun behavior when authz.enable_course_authoring is DISABLED
+    (legacy mode). Verifies the original behavior is preserved.
+    """
+
+    def setUp(self):
+        super().setUp()
+
+        self.url = reverse("course_handler")
+        self.user = UserFactory()
+        self.client = AjaxEnabledTestClient()
+        self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
+
+        # Create a source course and give user access
+        self.source_course = CourseFactory.create(
+            org='legacyorg',
+            number='201',
+            run='2025_Fall',
+            display_name='Legacy Source Course',
+        )
+        self.source_course_key = self.source_course.id
+
+        for role in [CourseInstructorRole, CourseStaffRole]:
+            role(self.source_course_key).add_users(self.user)
+
+    def test_rerun_creates_legacy_roles_pre_task(self):
+        """
+        Test that when authz is disabled (legacy mode), add_instructor is called
+        pre-task and creates CourseAccessRole records immediately.
+        """
+        response = self.client.ajax_post(self.url, {
+            'source_course_key': str(self.source_course_key),
+            'org': self.source_course_key.org,
+            'course': self.source_course_key.course,
+            'run': 'legacy_rerun',
+            'display_name': 'Legacy Rerun',
+        })
+
+        assert response.status_code == 200
+        data = parse_json(response)
+        dest_course_key = CourseKey.from_string(data['destination_course_key'])
+
+        # Verify legacy CourseAccessRole records exist for the destination course
+        legacy_roles = CourseAccessRole.objects.filter(
+            user=self.user,
+            course_id=dest_course_key,
+        )
+        assert legacy_roles.filter(role='instructor').exists()
+        assert legacy_roles.filter(role='staff').exists()
+
+    def test_rerun_succeeds_with_authz_disabled(self):
+        """
+        Test that course rerun still works correctly when authz is disabled.
+        """
+        response = self.client.ajax_post(self.url, {
+            'source_course_key': str(self.source_course_key),
+            'org': self.source_course_key.org,
+            'course': self.source_course_key.course,
+            'run': 'legacy_rerun_success',
+            'display_name': 'Legacy Rerun Success',
+        })
+
+        assert response.status_code == 200
+        data = parse_json(response)
+        dest_course_key = CourseKey.from_string(data['destination_course_key'])
+
+        # Verify the course was created
+        dest_course = self.store.get_course(dest_course_key)
+        assert dest_course is not None
+        assert dest_course.display_name == 'Legacy Rerun Success'
+
+        # Verify author access
+        assert has_course_author_access(self.user, dest_course_key)

cms/djangoapps/contentstore/views/course.py

@@ -419,8 +419,15 @@ def get_in_process_course_actions(request):
             exclude_args={'state': CourseRerunUIStateManager.State.SUCCEEDED},
             should_display=True,
         )
-        if user_has_course_permission(
-            request.user, COURSES_VIEW_COURSE.identifier, course.course_key, LegacyAuthoringPermission.READ
+        if (
+            # The user who initiated the rerun can always see its status.
+            # This is needed because when the authz flag is enabled, permission
+            # checks require a CourseOverview which doesn't exist until the
+            # rerun task clones the course.
+            course.created_user == request.user
+            or user_has_course_permission(
+                request.user, COURSES_VIEW_COURSE.identifier, course.course_key, LegacyAuthoringPermission.READ
+            )
         )
     ]
 
@@ -1324,8 +1331,13 @@ def rerun_course(user, source_course_key, org, number, run, fields, background=T
             raise PermissionDenied()
 
     # Make sure user has instructor and staff access to the destination course
-    # so the user can see the updated status for that course
-    add_instructor(destination_course_key, user, user)
+    # so the user can see the updated status for that course.
+    # When authz is enabled, we skip this because the authz layer requires a
+    # CourseOverview (which doesn't exist until the course is cloned in the task).
+    # In that case, visibility of the rerun status is granted by checking
+    # created_user on CourseRerunState instead.
+    if not core_toggles.enable_authz_course_authoring(destination_course_key):
+        add_instructor(destination_course_key, user, user)
 
     # Mark the action as initiated
     CourseRerunState.objects.initiated(source_course_key, destination_course_key, user, fields['display_name'])