From 3d5ed3e71b5f3f47f6d13f78ac8d81274d192c18 Mon Sep 17 00:00:00 2001 From: Diana Olarte Date: Thu, 17 Sep 2026 10:57:17 +1000 Subject: [PATCH] feat: add authz validation o course optimizer tool --- .../test_course_optimizer_permissions.py | 137 ++++++++++++++++++ .../rest_api/v0/views/course_optimizer.py | 34 ++++- 2 files changed, 164 insertions(+), 7 deletions(-) create mode 100644 cms/djangoapps/contentstore/rest_api/v0/tests/test_course_optimizer_permissions.py diff --git a/cms/djangoapps/contentstore/rest_api/v0/tests/test_course_optimizer_permissions.py b/cms/djangoapps/contentstore/rest_api/v0/tests/test_course_optimizer_permissions.py new file mode 100644 index 000000000000..33d3cfd78a8b --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v0/tests/test_course_optimizer_permissions.py @@ -0,0 +1,137 @@ +""" +Integration tests verifying authz permissions for v0 course optimizer REST API views. +""" +from unittest.mock import patch + +from django.urls import reverse +from openedx_authz.constants.roles import COURSE_AUDITOR, COURSE_EDITOR, COURSE_STAFF + +from cms.djangoapps.contentstore.tests.utils import CourseTestCase +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin + +VIEWS_MODULE = 'cms.djangoapps.contentstore.rest_api.v0.views.course_optimizer' + + +class CourseOptimizerV0AuthzTest(CourseAuthoringAuthzTestMixin, CourseTestCase): + """ + Integration tests for v0 course optimizer API authz permissions. + + All endpoints require courses.edit_course_content. + """ + + def setUp(self): + super().setUp() + self.link_check_url = reverse( + 'cms.djangoapps.contentstore:v0:link_check', + kwargs={'course_id': self.course.id}, + ) + self.link_check_status_url = reverse( + 'cms.djangoapps.contentstore:v0:link_check_status', + kwargs={'course_id': self.course.id}, + ) + self.rerun_link_update_url = reverse( + 'cms.djangoapps.contentstore:v0:rerun_link_update', + kwargs={'course_id': self.course.id}, + ) + self.rerun_link_update_status_url = reverse( + 'cms.djangoapps.contentstore:v0:rerun_link_update_status', + kwargs={'course_id': self.course.id}, + ) + + prev_run_links_patcher = patch( + f'{VIEWS_MODULE}.enable_course_optimizer_check_prev_run_links', + return_value=True, + ) + prev_run_links_patcher.start() + self.addCleanup(prev_run_links_patcher.stop) + + # --- LinkCheckView (POST) --- + + @patch(f'{VIEWS_MODULE}.check_broken_links') + def test_editor_can_start_link_check(self, mock_task): + self.add_user_to_role_in_course(self.authorized_user, COURSE_EDITOR.external_key, self.course.id) + resp = self.authorized_client.post(self.link_check_url) + assert resp.status_code == 200 + mock_task.delay.assert_called_once() + + @patch(f'{VIEWS_MODULE}.check_broken_links') + def test_staff_can_start_link_check(self, mock_task): + self.add_user_to_role_in_course(self.authorized_user, COURSE_STAFF.external_key, self.course.id) + resp = self.authorized_client.post(self.link_check_url) + assert resp.status_code == 200 + mock_task.delay.assert_called_once() + + @patch(f'{VIEWS_MODULE}.check_broken_links') + def test_auditor_cannot_start_link_check(self, mock_task): + self.add_user_to_role_in_course(self.authorized_user, COURSE_AUDITOR.external_key, self.course.id) + resp = self.authorized_client.post(self.link_check_url) + assert resp.status_code == 403 + mock_task.delay.assert_not_called() + + @patch(f'{VIEWS_MODULE}.check_broken_links') + def test_unauthorized_cannot_start_link_check(self, mock_task): + resp = self.unauthorized_client.post(self.link_check_url) + assert resp.status_code == 403 + mock_task.delay.assert_not_called() + + # --- LinkCheckStatusView (GET) --- + + def test_editor_can_get_link_check_status(self): + self.add_user_to_role_in_course(self.authorized_user, COURSE_EDITOR.external_key, self.course.id) + resp = self.authorized_client.get(self.link_check_status_url) + assert resp.status_code == 200 + + def test_auditor_cannot_get_link_check_status(self): + self.add_user_to_role_in_course(self.authorized_user, COURSE_AUDITOR.external_key, self.course.id) + resp = self.authorized_client.get(self.link_check_status_url) + assert resp.status_code == 403 + + def test_unauthorized_cannot_get_link_check_status(self): + resp = self.unauthorized_client.get(self.link_check_status_url) + assert resp.status_code == 403 + + # --- RerunLinkUpdateView (POST) --- + + @patch(f'{VIEWS_MODULE}.update_course_rerun_links') + def test_editor_can_start_rerun_link_update(self, mock_task): + self.add_user_to_role_in_course(self.authorized_user, COURSE_EDITOR.external_key, self.course.id) + resp = self.authorized_client.post(self.rerun_link_update_url, data={'action': 'all'}, format='json') + assert resp.status_code == 200 + mock_task.delay.assert_called_once() + + @patch(f'{VIEWS_MODULE}.update_course_rerun_links') + def test_auditor_cannot_start_rerun_link_update(self, mock_task): + self.add_user_to_role_in_course(self.authorized_user, COURSE_AUDITOR.external_key, self.course.id) + resp = self.authorized_client.post(self.rerun_link_update_url, data={'action': 'all'}, format='json') + assert resp.status_code == 403 + mock_task.delay.assert_not_called() + + @patch(f'{VIEWS_MODULE}.update_course_rerun_links') + def test_unauthorized_cannot_start_rerun_link_update(self, mock_task): + resp = self.unauthorized_client.post(self.rerun_link_update_url, data={'action': 'all'}, format='json') + assert resp.status_code == 403 + mock_task.delay.assert_not_called() + + # --- RerunLinkUpdateStatusView (GET) --- + + def test_editor_can_get_rerun_link_update_status(self): + self.add_user_to_role_in_course(self.authorized_user, COURSE_EDITOR.external_key, self.course.id) + resp = self.authorized_client.get(self.rerun_link_update_status_url) + assert resp.status_code == 200 + + def test_auditor_cannot_get_rerun_link_update_status(self): + self.add_user_to_role_in_course(self.authorized_user, COURSE_AUDITOR.external_key, self.course.id) + resp = self.authorized_client.get(self.rerun_link_update_status_url) + assert resp.status_code == 403 + + def test_unauthorized_cannot_get_rerun_link_update_status(self): + resp = self.unauthorized_client.get(self.rerun_link_update_status_url) + assert resp.status_code == 403 + + # --- Superuser bypass --- + + @patch(f'{VIEWS_MODULE}.check_broken_links') + def test_superuser_can_start_link_check(self, mock_task): + resp = self.super_client.post(self.link_check_url) + assert resp.status_code == 200 + mock_task.delay.assert_called_once() diff --git a/cms/djangoapps/contentstore/rest_api/v0/views/course_optimizer.py b/cms/djangoapps/contentstore/rest_api/v0/views/course_optimizer.py index 9b012d918e06..077ba1bd9aec 100644 --- a/cms/djangoapps/contentstore/rest_api/v0/views/course_optimizer.py +++ b/cms/djangoapps/contentstore/rest_api/v0/views/course_optimizer.py @@ -3,6 +3,7 @@ import edx_api_doc_tools as apidocs from opaque_keys import InvalidKeyError from opaque_keys.edx.keys import CourseKey +from openedx_authz.constants.permissions import COURSES_EDIT_COURSE_CONTENT from rest_framework import status from rest_framework.request import Request from rest_framework.response import Response @@ -21,8 +22,9 @@ ) from cms.djangoapps.contentstore.tasks import check_broken_links, update_course_rerun_links from cms.djangoapps.contentstore.toggles import enable_course_optimizer_check_prev_run_links -from common.djangoapps.student.auth import has_course_author_access, has_studio_read_access from common.djangoapps.util.json_request import JsonResponse +from openedx.core.djangoapps.authz.constants import LegacyAuthoringPermission +from openedx.core.djangoapps.authz.decorators import user_has_course_permission from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, verify_course_exists, view_auth_classes @@ -58,7 +60,12 @@ def post(self, request: Request, course_id: str): """ course_key = CourseKey.from_string(course_id) - if not has_studio_read_access(request.user, course_key): + if not user_has_course_permission( + request.user, + COURSES_EDIT_COURSE_CONTENT.identifier, + course_key, + LegacyAuthoringPermission.READ, + ): self.permission_denied(request) check_broken_links.delay(request.user.id, course_id, request.LANGUAGE_CODE) @@ -206,7 +213,12 @@ def get(self, request: Request, course_id: str): } """ course_key = CourseKey.from_string(course_id) - if not has_course_author_access(request.user, course_key): + if not user_has_course_permission( + request.user, + COURSES_EDIT_COURSE_CONTENT.identifier, + course_key, + LegacyAuthoringPermission.WRITE, + ): self.permission_denied(request) link_check_data = get_link_check_data(request, course_id) @@ -280,8 +292,12 @@ def post(self, request: Request, course_id: str): status=status.HTTP_404_NOT_FOUND, ) - # Check course author permissions - if not has_course_author_access(request.user, course_key): + if not user_has_course_permission( + request.user, + COURSES_EDIT_COURSE_CONTENT.identifier, + course_key, + LegacyAuthoringPermission.WRITE, + ): self.permission_denied(request) if not enable_course_optimizer_check_prev_run_links(course_key): @@ -401,8 +417,12 @@ def get(self, request: Request, course_id: str): status=status.HTTP_404_NOT_FOUND, ) - # Check course author permissions - if not has_course_author_access(request.user, course_key): + if not user_has_course_permission( + request.user, + COURSES_EDIT_COURSE_CONTENT.identifier, + course_key, + LegacyAuthoringPermission.WRITE, + ): self.permission_denied(request) if not enable_course_optimizer_check_prev_run_links(course_key):