Merge branch 'master' into marslan/10256-course_detail_title

Author: pdpinch

cms/djangoapps/contentstore/api/tests/base.py

@@ -18,6 +18,8 @@ class BaseCourseViewTest(SharedModuleStoreTestCase, APITestCase):
     Base test class for course data views.
     """
     view_name = None  # The name of the view to use in reverse() call in self.get_url()
+    course_key_arg_name = 'course_id'
+    extra_request_args = {}
 
     @classmethod
     def setUpClass(cls):
@@ -86,9 +88,10 @@ def get_url(self, course_id):
         """
         Helper function to create the url
         """
+        args = {
+            self.course_key_arg_name: course_id,
+        }
         return reverse(
             self.view_name,
-            kwargs={
-                'course_id': course_id
-            }
+            kwargs= args | self.extra_request_args
         )

cms/djangoapps/contentstore/asset_storage_handlers.py

@@ -17,15 +17,23 @@
 from django.views.decorators.csrf import ensure_csrf_cookie
 from django.views.decorators.http import require_http_methods, require_POST
 from opaque_keys.edx.keys import AssetKey, CourseKey
+from openedx.core.djangoapps.authz.constants import LegacyAuthoringPermission
+from openedx.core.djangoapps.authz.decorators import user_has_course_permission
 from pymongo import ASCENDING, DESCENDING
 
-from common.djangoapps.student.auth import has_course_author_access
 from common.djangoapps.util.date_utils import get_default_time_display
 from common.djangoapps.util.json_request import JsonResponse
 from openedx.core.djangoapps.contentserver.caching import del_cached_content
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
 from openedx.core.djangoapps.user_api.models import UserPreference
+from openedx_authz.constants.permissions import (
+    COURSES_VIEW_FILES,
+    COURSES_CREATE_FILES,
+    COURSES_DELETE_FILES,
+    COURSES_EDIT_FILES,
+)
 from openedx_filters.content_authoring.filters import LMSPageURLRequested
+from openedx.core.toggles import enable_authz_course_authoring
 from xmodule.contentstore.content import StaticContent  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.contentstore.django import contentstore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.exceptions import NotFoundError  # lint-amnesty, pylint: disable=wrong-import-order
@@ -73,8 +81,8 @@ def handle_assets(request, course_key_string=None, asset_key_string=None):
         json: delete an asset
     '''
     course_key = CourseKey.from_string(course_key_string)
-    if not has_course_author_access(request.user, course_key):
-        raise PermissionDenied()
+    # Enforce file permissions.
+    _authz_enforce_file_permissions(request, course_key)
 
     response_format = get_response_format(request)
     if request_response_format_is_json(request, response_format):
@@ -91,12 +99,60 @@ def handle_assets(request, course_key_string=None, asset_key_string=None):
     return HttpResponseNotFound()
 
 
+def _authz_enforce_file_permissions(request, course_key):
+    """
+    Enforce permissions for file operations in asset handler.
+    When the authz.enable_course_authoring flag is enabled for the specified course,
+    This function enforces the appropriate file permission depending on request content.
+    When the flag is disabled, it enforces the legacy has_studio_write_access permission.
+    """
+    # Enforce permission to view files.
+    # This is the minimum permission needed for handling assets.
+    if not user_has_course_permission(
+        request.user,
+        COURSES_VIEW_FILES.identifier,
+        course_key,
+        LegacyAuthoringPermission.WRITE
+    ):
+        raise PermissionDenied()
+
+    if enable_authz_course_authoring(course_key):
+        # Check create, edit and delete permissions for AuthZ-enabled courses.
+        if request.method in ('PUT', 'POST'):
+            permission = (
+                COURSES_CREATE_FILES.identifier
+                if 'file' in request.FILES
+                else COURSES_EDIT_FILES.identifier
+            )
+
+            if not user_has_course_permission(
+                request.user,
+                permission,
+                course_key,
+                LegacyAuthoringPermission.WRITE
+            ):
+                raise PermissionDenied()
+
+        if request.method == 'DELETE' and not user_has_course_permission(
+            request.user,
+            COURSES_DELETE_FILES.identifier,
+            course_key,
+            LegacyAuthoringPermission.WRITE
+        ):
+            raise PermissionDenied()
+
+
 def get_asset_usage_path_json(request, course_key, asset_key_string):
     """
     Get a list of units with ancestors that use given asset.
     """
     course_key = CourseKey.from_string(course_key)
-    if not has_course_author_access(request.user, course_key):
+    if not user_has_course_permission(
+        request.user,
+        COURSES_VIEW_FILES.identifier,
+        course_key,
+        LegacyAuthoringPermission.WRITE
+    ):
         raise PermissionDenied()
     asset_location = AssetKey.from_string(asset_key_string) if asset_key_string else None
     usage_locations = _get_asset_usage_path(course_key, [{'asset_key': asset_location}])

cms/djangoapps/contentstore/rest_api/v1/views/certificates.py

@@ -6,11 +6,14 @@
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
+from openedx_authz.constants.permissions import COURSES_MANAGE_CERTIFICATES
+
 from cms.djangoapps.contentstore.utils import get_certificates_context
 from cms.djangoapps.contentstore.rest_api.v1.serializers import (
     CourseCertificatesSerializer,
 )
-from common.djangoapps.student.auth import has_studio_write_access
+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,
@@ -96,7 +99,12 @@ def get(self, request: Request, course_id: str):
         course_key = CourseKey.from_string(course_id)
         store = modulestore()
 
-        if not has_studio_write_access(request.user, course_key):
+        if not user_has_course_permission(
+            request.user,
+            COURSES_MANAGE_CERTIFICATES.identifier,
+            course_key,
+            LegacyAuthoringPermission.WRITE
+        ):
             self.permission_denied(request)
 
         with store.bulk_operations(course_key):

cms/djangoapps/contentstore/rest_api/v1/views/tests/test_certificates.py

@@ -4,8 +4,11 @@
 from django.urls import reverse
 from rest_framework import status
 
+from openedx_authz.constants.roles import COURSE_STAFF, COURSE_EDITOR
+
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
 from cms.djangoapps.contentstore.views.tests.test_certificates import HelperMethods
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin
 
 from ...mixins import PermissionAccessMixin
 
@@ -36,3 +39,35 @@ def test_success_response(self):
         self.assertEqual(response_data["course_number_override"], self.course.display_coursenumber)
         self.assertEqual(response_data["course_title"], self.course.display_name_with_default)
         self.assertEqual(response_data["course_number"], self.course.number)
+
+
+class CourseCertificatesAuthzViewTest(
+        CourseAuthoringAuthzTestMixin, CourseTestCase, PermissionAccessMixin, HelperMethods
+    ):
+    """
+    Tests for CourseCertificatesView with AuthZ enabled.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.url = reverse(
+            "cms.djangoapps.contentstore:v1:certificates",
+            kwargs={"course_id": self.course.id},
+        )
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self._add_course_certificates(count=2, signatory_count=2)
+        self.add_user_to_role_in_course(self.authorized_user, COURSE_STAFF.external_key, self.course.id)
+        resp = self.authorized_client.get(self.url)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_non_staff_user_cannot_access(self):
+        """
+        User without permissions should be denied.
+        This case validates that a non-staff user cannot access.
+        """
+        self._add_course_certificates(count=2, signatory_count=2)
+        self.add_user_to_role_in_course(self.authorized_user, COURSE_EDITOR.external_key, self.course.id)
+        resp = self.authorized_client.get(self.url)
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstream_sync_integration.py

@@ -128,7 +128,7 @@ def _create_block_from_upstream(
             "library_content_key": upstream_key,
         }, expect_response=expect_response)
 
-    def _update_course_block_fields(self, usage_key: str, fields: dict[str, Any] = None):
+    def _update_course_block_fields(self, usage_key: str, fields: dict[str, Any] | None = None):
         """ Update fields of an XBlock """
         return self._api('patch', f"/xblock/{usage_key}", {
             "metadata": fields,
@@ -158,7 +158,7 @@ def _get_downstream_links(
             data["use_top_level_parents"] = str(use_top_level_parents)
         return self.client.get("/api/contentstore/v2/downstreams/", data=data)
 
-    def assertXmlEqual(self, xml_str_a: str, xml_str_b: str) -> bool:
+    def assertXmlEqual(self, xml_str_a: str, xml_str_b: str) -> None:
         """ Assert that the given XML strings are equal, ignoring attribute order and some whitespace variations. """
         self.assertEqual(
             ElementTree.canonicalize(xml_str_a, strip_text=True),

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py

@@ -11,6 +11,7 @@
 from freezegun import freeze_time
 from opaque_keys.edx.keys import ContainerKey, UsageKey
 from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
+from openedx_content import models_api as content_models
 from organizations.models import Organization
 
 from cms.djangoapps.contentstore.helpers import StaticFileNotices
@@ -221,6 +222,11 @@ def setUp(self):
         self._publish_library_block(self.video_lib_id)
         self._publish_library_block(self.html_lib_id)
 
+    def tearDown(self):
+        # If we're working with Containers in test cases, we need this line:
+        content_models.Container.reset_cache()
+        return super().tearDown()
+
     def _api(self, method, url, data, expect_response):
         """
         Call a REST API

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_home.py

@@ -3,10 +3,9 @@
 """
 
 from collections import OrderedDict
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import ddt
-import pytz
 from django.conf import settings
 from django.urls import reverse
 from rest_framework import status
@@ -36,7 +35,7 @@ def setUp(self):
             display_name="Demo Course (Sample)",
             id=archived_course_key,
             org=archived_course_key.org,
-            end=(datetime.now() - timedelta(days=365)).replace(tzinfo=pytz.UTC),
+            end=(datetime.now() - timedelta(days=365)).replace(tzinfo=timezone.utc),
         )
         self.non_staff_client, _ = self.create_non_staff_authed_user_client()
 
@@ -256,7 +255,7 @@ def test_filter_and_ordering_courses(
             display_name="Course (Demo)",
             id=archived_course_key,
             org=archived_course_key.org,
-            end=(datetime.now() - timedelta(days=365)).replace(tzinfo=pytz.UTC),
+            end=(datetime.now() - timedelta(days=365)).replace(tzinfo=timezone.utc),
         )
         active_course_key = self.store.make_course_key("foo-org", "foo-number", "foo-run")
         CourseOverviewFactory.create(

cms/djangoapps/contentstore/tests/test_utils.py

@@ -9,7 +9,6 @@
 from django.conf import settings
 from django.test import TestCase
 from django.test.utils import override_settings
-from edx_toggles.toggles.testutils import override_waffle_flag
 from opaque_keys.edx.keys import CourseKey
 from opaque_keys.edx.locator import CourseLocator, LibraryLocator
 from openedx_events.tests.utils import OpenEdxEventsTestMixin
@@ -24,7 +23,6 @@
 from cms.djangoapps.contentstore.utils import send_course_update_notification
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.tests.factories import GlobalStaffFactory, InstructorFactory, UserFactory
-from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
 from openedx.core.djangoapps.notifications.models import Notification
 from openedx.core.djangoapps.site_configuration.tests.test_util import with_site_configuration_context
 from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
@@ -938,7 +936,6 @@ def test_update_course_details_instructor_paced(self, mock_update):
         mock_update.assert_called_once_with(self.course.id, payload, mock_request.user)
 
 
-@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
 class CourseUpdateNotificationTests(OpenEdxEventsTestMixin, ModuleStoreTestCase):
     """
     Unit tests for the course_update notification.