diff --git a/cms/djangoapps/contentstore/rest_api/v2/authoring_urls.py b/cms/djangoapps/contentstore/rest_api/v2/authoring_urls.py new file mode 100644 index 000000000000..1c0d693dec4f --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/authoring_urls.py @@ -0,0 +1,21 @@ +"""Contentstore API v2 authoring URLs.""" + +from django.urls import path + +from cms.djangoapps.contentstore.rest_api.v2.views.textbooks import CourseTextbooksViewSet +from cms.djangoapps.contentstore.rest_api.v2.views.video_settings import CourseVideoSettingsViewSet + +app_name = "authoring_v2" + +urlpatterns = [ + path( + "courses//textbooks/", + CourseTextbooksViewSet.as_view({"get": "list"}), + name="course_textbook_list", + ), + path( + "courses//video_settings/", + CourseVideoSettingsViewSet.as_view({"get": "retrieve"}), + name="course_video_settings", + ), +] diff --git a/cms/djangoapps/contentstore/rest_api/v2/serializers/textbooks.py b/cms/djangoapps/contentstore/rest_api/v2/serializers/textbooks.py new file mode 100644 index 000000000000..42feaaa07ddb --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/serializers/textbooks.py @@ -0,0 +1,69 @@ +"""Serializers for course PDF textbooks.""" + +from rest_framework import serializers + + +class TextbookChapterSerializer(serializers.Serializer): + """ + One chapter of a course PDF textbook. + + A stored chapter missing ``title`` or ``url``, or holding null for either, + is returned with an empty string in its place rather than failing the page. + """ + + title = serializers.CharField( + default="", + help_text="Chapter title shown in the textbook's table of contents. Empty when the stored chapter has none.", + ) + url = serializers.CharField( + default="", + help_text=( + "Address of the chapter's PDF file, usually a course asset path. " + "Empty when the stored chapter has none." + ), + ) + + class Meta: + ref_name = "TextbookChapter" + + def to_representation(self, instance): + data = super().to_representation(instance) + return {key: "" if value is None else value for key, value in data.items()} + + +class CourseTextbookSerializer(serializers.Serializer): + """ + One PDF textbook of a course, as stored in the course's settings. + + A stored textbook can lack ``id``, ``chapters`` or ``tab_title``: imported + courses may carry textbooks written by other tools, such as the single-file + form that has a ``url`` and no chapters. Such a textbook is returned with + ``null`` for the missing scalar and an empty chapter list, rather than + failing the whole page. A stored ``null`` chapter list is returned empty + too. Stored keys not declared here are not returned. + """ + + id = serializers.CharField( + allow_null=True, + default=None, + help_text="Identifier of the textbook, unique within its course. Null when the stored textbook has none.", + ) + chapters = TextbookChapterSerializer( + many=True, + default=list, + help_text="Chapters in reading order. Empty when the stored textbook has no chapter list.", + ) + tab_title = serializers.CharField( + allow_null=True, + default=None, + help_text="Title of the course tab that opens the textbook. Null when the stored textbook has none.", + ) + + class Meta: + ref_name = "CourseTextbook" + + def to_representation(self, instance): + data = super().to_representation(instance) + if data.get("chapters") is None: + data["chapters"] = [] + return data diff --git a/cms/djangoapps/contentstore/rest_api/v2/serializers/video_settings.py b/cms/djangoapps/contentstore/rest_api/v2/serializers/video_settings.py new file mode 100644 index 000000000000..06029d05108d --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/serializers/video_settings.py @@ -0,0 +1,145 @@ +"""Serializers for a course's video upload and transcript settings.""" + +from rest_framework import serializers + + +class VideoSettingsImageSerializer(serializers.Serializer): + """Limits that apply to video thumbnail images uploaded to the course.""" + + video_image_upload_enabled = serializers.BooleanField( + help_text="Whether thumbnail images may be uploaded for the course's videos.", + ) + max_size = serializers.IntegerField(help_text="Largest accepted image, in bytes.") + min_size = serializers.IntegerField(help_text="Smallest accepted image, in bytes.") + max_width = serializers.IntegerField(help_text="Largest accepted image width, in pixels.") + max_height = serializers.IntegerField(help_text="Largest accepted image height, in pixels.") + supported_file_formats = serializers.DictField( + child=serializers.CharField(), + help_text="Accepted image file extensions, each mapped to its MIME type.", + ) + + class Meta: + ref_name = "VideoSettingsImage" + + +class VideoSettingsTranscriptSerializer(serializers.Serializer): + """Studio handler addresses and options for the course's video transcripts.""" + + transcript_download_handler_url = serializers.CharField( + help_text="Studio address that downloads a transcript.", + ) + transcript_upload_handler_url = serializers.CharField( + help_text="Studio address that uploads a transcript.", + ) + transcript_delete_handler_url = serializers.CharField( + help_text="Studio address that deletes a transcript of this course.", + ) + transcript_download_file_format = serializers.CharField( + source="trancript_download_file_format", + help_text="File format transcripts are downloaded in.", + ) + transcript_preferences_handler_url = serializers.CharField( + allow_null=True, + help_text="Studio address that saves the course's transcript preferences. Null when transcripts are disabled.", + ) + transcript_credentials_handler_url = serializers.CharField( + allow_null=True, + help_text="Studio address that saves transcript provider credentials. Null when transcripts are disabled.", + ) + transcription_plans = serializers.DictField( + child=serializers.DictField(), + allow_null=True, + help_text=( + "Third-party transcription plans, keyed by provider name, with each plan's options as the " + "provider defines them. Null when transcripts are disabled." + ), + ) + + class Meta: + ref_name = "VideoSettingsTranscript" + + +class VideoSettingsTranscriptPreferencesSerializer(serializers.Serializer): + """The course's saved preferences for third-party transcription.""" + + course_id = serializers.CharField(help_text="Key of the course the preferences belong to.") + provider = serializers.CharField(help_text="Transcription provider.") + cielo24_fidelity = serializers.CharField(allow_null=True, help_text="Cielo24 fidelity level, when Cielo24 is used.") + cielo24_turnaround = serializers.CharField( + allow_null=True, + help_text="Cielo24 turnaround, when Cielo24 is used.", + ) + three_play_turnaround = serializers.CharField( + allow_null=True, + help_text="3Play Media turnaround, when 3Play Media is used.", + ) + preferred_languages = serializers.ListField( + child=serializers.CharField(), + help_text="Language codes transcripts are requested in.", + ) + video_source_language = serializers.CharField( + allow_null=True, + help_text="Language code of the videos' spoken language.", + ) + modified = serializers.CharField(help_text="When the preferences were last changed.") + + class Meta: + ref_name = "VideoSettingsTranscriptPreferences" + + +class TranscriptLanguageSerializer(serializers.Serializer): + """A language a transcript may be written in.""" + + language_code = serializers.CharField(help_text="Language code.") + language_text = serializers.CharField(help_text="Display name of the language.") + + class Meta: + ref_name = "TranscriptLanguage" + + +class CourseVideoSettingsSerializer(serializers.Serializer): + """The course's video upload and transcript settings.""" + + image_upload_url = serializers.CharField(help_text="Studio address that uploads a video's thumbnail image.") + video_handler_url = serializers.CharField(help_text="Studio address that manages the course's video uploads.") + encodings_download_url = serializers.CharField( + help_text="Studio address that downloads the course's video encodings as CSV.", + ) + default_video_image_url = serializers.CharField(help_text="Address of the image shown for videos without one.") + concurrent_upload_limit = serializers.IntegerField( + help_text="How many videos a client may upload at the same time.", + ) + video_supported_file_formats = serializers.ListField( + child=serializers.CharField(), + help_text="Accepted video file extensions.", + ) + video_upload_max_file_size = serializers.IntegerField(help_text="Largest accepted video file, in gigabytes.") + video_image_settings = VideoSettingsImageSerializer(help_text="Limits for video thumbnail images.") + is_video_transcript_enabled = serializers.BooleanField( + help_text="Whether third-party transcription is enabled for the course.", + ) + is_ai_translations_enabled = serializers.BooleanField( + help_text="Whether AI translation of transcripts is enabled for the course.", + ) + active_transcript_preferences = VideoSettingsTranscriptPreferencesSerializer( + allow_null=True, + help_text="The course's transcription preferences. Null when transcripts are disabled or none are saved.", + ) + transcript_credentials = serializers.DictField( + child=serializers.BooleanField(), + allow_null=True, + help_text=( + "Whether the course's organization has credentials for each transcription provider, " + "keyed by provider name. Null when transcripts are disabled." + ), + ) + transcript_available_languages = TranscriptLanguageSerializer( + many=True, + help_text="Every language a transcript may be written in, ordered by display name.", + ) + video_transcript_settings = VideoSettingsTranscriptSerializer( + help_text="Studio handler addresses and options for transcripts.", + ) + + class Meta: + ref_name = "CourseVideoSettings" diff --git a/cms/djangoapps/contentstore/rest_api/v2/textbooks_service.py b/cms/djangoapps/contentstore/rest_api/v2/textbooks_service.py new file mode 100644 index 000000000000..dc663d2c0932 --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/textbooks_service.py @@ -0,0 +1,27 @@ +"""Service layer for the v2 course textbooks endpoint.""" + +import logging + +from rest_framework.exceptions import NotFound + +from cms.djangoapps.contentstore.utils import get_textbooks_context +from xmodule.modulestore.django import modulestore + +log = logging.getLogger(__name__) + +COURSE_NOT_FOUND_MESSAGE = "The requested course does not exist." + + +def get_course_textbooks(course_key): + """ + Return the stored PDF textbooks of the course ``course_key``, in stored order. + + Raises ``NotFound`` when the course cannot be loaded from the modulestore. + """ + store = modulestore() + with store.bulk_operations(course_key): + course = store.get_course(course_key) + if course is None: + log.info("Textbooks requested for course %s, which is not in the modulestore.", course_key) + raise NotFound(COURSE_NOT_FOUND_MESSAGE) + return get_textbooks_context(course)["textbooks"] diff --git a/cms/djangoapps/contentstore/rest_api/v2/video_settings_service.py b/cms/djangoapps/contentstore/rest_api/v2/video_settings_service.py new file mode 100644 index 000000000000..feeb6116c090 --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/video_settings_service.py @@ -0,0 +1,29 @@ +"""Service layer for the v2 course video settings endpoint.""" + +import logging + +from rest_framework.exceptions import NotFound + +from cms.djangoapps.contentstore.utils import get_course_video_settings_context +from xmodule.modulestore.django import modulestore + +log = logging.getLogger(__name__) + +COURSE_NOT_FOUND_MESSAGE = "The requested course does not exist." + + +def get_course_video_settings(course_key): + """ + Return the video settings of the course ``course_key``. + + Raises ``NotFound`` when the course cannot be loaded from the modulestore. + Nothing is written. + """ + store = modulestore() + with store.bulk_operations(course_key): + course = store.get_course(course_key) + if course is None: + log.info("Video settings requested for course %s, which is not in the modulestore.", course_key) + raise NotFound(COURSE_NOT_FOUND_MESSAGE) + + return get_course_video_settings_context(course) diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/tests/helpers.py b/cms/djangoapps/contentstore/rest_api/v2/views/tests/helpers.py new file mode 100644 index 000000000000..e956150f300d --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/views/tests/helpers.py @@ -0,0 +1,97 @@ +"""Shared helpers for the v2 authoring endpoint tests.""" + +from unittest.mock import patch + +from crum import signals as crum_signals +from django.core import signals as core_signals +from django.db.models import signals as model_signals +from django.dispatch import Signal +from drf_spectacular.generators import SchemaGenerator +from drf_spectacular.settings import patched_settings +from openedx_events.tooling import OpenEdxPublicSignal +from rest_framework.test import APIClient + +PASSWORD = "test" + +# The schema settings the CMS publishes its authoring schema with. The test +# settings module configures none. +SCHEMA_SETTINGS = { + "PREPROCESSING_HOOKS": ["cms.lib.spectacular.cms_api_filter"], + "SCHEMA_PATH_PREFIX": r"/api/(contentstore|authoring)", +} + + +def logged_in_client(user): + """Return a client holding a real session for ``user``, created with ``PASSWORD``.""" + client = APIClient() + assert client.login(username=user.username, password=PASSWORD) + return client + + +def writes(queries): + """Return the SQL statements in ``queries`` that insert, update or delete rows.""" + return [q["sql"] for q in queries if q["sql"].lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE"))] + + +class SignalRecorder: + """ + Record every Django signal and openedx-event sent while active. + + ``unexpected()`` leaves out only the signals any request sends without + writing anything: the request boundaries, model instance construction on + every row read, and django-crum's current-user bookkeeping. + """ + + NEUTRAL = ( + core_signals.request_started, + core_signals.request_finished, + model_signals.pre_init, + model_signals.post_init, + crum_signals.current_user_getter, + crum_signals.current_user_setter, + ) + + def __init__(self): + self.sent = [] + self._patches = [] + + def _wrap(self, owner, name): + """Prepare a patch of ``owner.name`` that records each send before performing it.""" + original = getattr(owner, name) + recorder = self + + def wrapper(signal, *args, **kwargs): + recorder.sent.append((name, signal)) + return original(signal, *args, **kwargs) + + self._patches.append(patch.object(owner, name, wrapper)) + + def __enter__(self): + self._wrap(Signal, "send") + self._wrap(Signal, "send_robust") + self._wrap(OpenEdxPublicSignal, "send_event") + for active in self._patches: + active.start() + return self + + def __exit__(self, *exc): + for active in self._patches: + active.stop() + + def unexpected(self): + """Return the recorded ``(method, signal)`` pairs other than the neutral signals.""" + return [(name, signal) for name, signal in self.sent if not any(signal is n for n in self.NEUTRAL)] + + +def generate_cms_schema(): + """Generate the CMS authoring schema with the settings the service publishes it with.""" + with patched_settings(SCHEMA_SETTINGS): + return SchemaGenerator().get_schema(request=None, public=True) + + +def resolve_ref(schema, node): + """Return the component ``node`` points at, or ``node`` itself when it is inline.""" + ref = node.get("$ref") + if ref is None: + return node + return schema["components"]["schemas"][ref.rsplit("/", 1)[1]] diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_textbooks.py b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_textbooks.py new file mode 100644 index 000000000000..c00d7f61255c --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_textbooks.py @@ -0,0 +1,588 @@ +"""Tests for the v2 course textbooks endpoint.""" + +from pathlib import Path +from unittest.mock import patch + +import ddt +from ccx_keys.locator import CCXLocator +from django.db import connection +from django.db.models import signals as model_signals +from django.test import Client +from django.test.utils import CaptureQueriesContext +from django.urls import resolve, reverse +from edx_rest_framework_extensions.testing import assert_error_envelope +from openedx_authz.constants.roles import COURSE_AUDITOR, COURSE_LIMITED_STAFF, COURSE_STAFF +from openedx_authz.models.engine import PolicyCacheControl +from rest_framework.test import APIClient + +from cms.djangoapps.contentstore.rest_api.v1.views.textbooks import CourseTextbooksView +from cms.djangoapps.contentstore.rest_api.v2.views.tests.helpers import ( + PASSWORD, + SignalRecorder, + generate_cms_schema, + logged_in_client, + resolve_ref, + writes, +) +from cms.djangoapps.contentstore.rest_api.v2.views.textbooks import CourseTextbooksViewSet +from cms.djangoapps.contentstore.tests.utils import CourseTestCase +from cms.lib.spectacular import cms_api_filter +from common.djangoapps.student.roles import CourseInstructorRole, CourseLimitedStaffRole, CourseStaffRole +from common.djangoapps.student.tests.factories import UserFactory +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin +from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user +from openedx.core.djangoapps.oauth_dispatch.tests.factories import AccessTokenFactory, ApplicationFactory +from xmodule.modulestore.tests.factories import CourseFactory + +URL_NAME = "authoring_v2:course_textbook_list" +LEGACY_URL_NAME = "cms.djangoapps.contentstore:v1:textbooks" +PAGE_KEYS = {"count", "num_pages", "current_page", "start", "next", "previous", "results"} +MISSING_COURSE_KEY = "course-v1:NoSuch+Course+Run" + + +def _textbook(index, chapter_count=1): + """Return a stored textbook with ``chapter_count`` chapters.""" + return { + "id": f"{index}Book", + "tab_title": f"Book {index}", + "chapters": [{"title": f"Chapter {n}", "url": f"/static/book{index}_{n}.pdf"} for n in range(chapter_count)], + } + + +class TextbooksTestBase(CourseTestCase): + """A course with textbooks and the callers the tests need.""" + + def setUp(self): + super().setUp() + self.api_client = APIClient() + self.other_course = CourseFactory.create(org="OtherX", course="Other", run="Run") + + self.no_role_user = UserFactory(password=PASSWORD) + self.course_staff = UserFactory(password=PASSWORD) + CourseStaffRole(self.course.id).add_users(self.course_staff) + self.course_instructor = UserFactory(password=PASSWORD) + CourseInstructorRole(self.course.id).add_users(self.course_instructor) + self.limited_staff = UserFactory(password=PASSWORD) + CourseLimitedStaffRole(self.course.id).add_users(self.limited_staff) + self.other_course_staff = UserFactory(password=PASSWORD) + CourseStaffRole(self.other_course.id).add_users(self.other_course_staff) + self.other_course_instructor = UserFactory(password=PASSWORD) + CourseInstructorRole(self.other_course.id).add_users(self.other_course_instructor) + self.global_staff = UserFactory(is_staff=True, password=PASSWORD) + self.superuser = UserFactory(is_superuser=True, password=PASSWORD) + # Role checks create the authorization engine's cache-version row on + # first use; create it here so request-level measurements exclude it. + PolicyCacheControl.get() + + def set_textbooks(self, textbooks): + self.course.pdf_textbooks = textbooks + self.save_course() + + def url(self, course_key=None): + return reverse(URL_NAME, kwargs={"course_key": str(course_key or self.course.id)}) + + def legacy_url(self, course_key=None): + return reverse(LEGACY_URL_NAME, kwargs={"course_id": str(course_key or self.course.id)}) + + def get(self, user, course_key=None, method="get", **params): + client = logged_in_client(user) if user is not None else APIClient() + return getattr(client, method)(self.url(course_key), params) + + def get_legacy(self, user, course_key=None): + client = logged_in_client(user) if user is not None else APIClient() + return client.get(self.legacy_url(course_key)) + + +@ddt.ddt +class CourseTextbooksAccessTest(TextbooksTestBase): + """Who may list a course's textbooks, with course-authoring policy disabled.""" + + def setUp(self): + super().setUp() + self.set_textbooks([_textbook(1)]) + + @ddt.data("course_staff", "course_instructor", "global_staff", "superuser") + def test_allowed_callers_get_the_textbooks(self, caller): + for method in ("get", "head"): + response = self.get(getattr(self, caller), method=method) + assert response.status_code == 200, (caller, method) + response = self.get(getattr(self, caller)) + assert response.data["results"] == [_textbook(1)] + assert list(response.data["results"][0]) == ["id", "chapters", "tab_title"] + + def test_anonymous_is_refused_with_a_challenge(self): + response = self.get(None) + assert_error_envelope(response, expected_status=401, expected_type_slug="authn") + assert response["WWW-Authenticate"] == 'JWT realm="api"' + + @ddt.data("no_role_user", "limited_staff", "other_course_staff", "other_course_instructor") + def test_callers_without_access_to_this_course_are_refused(self, caller): + for method in ("get", "head"): + response = self.get(getattr(self, caller), method=method) + assert response.status_code == 403, (caller, method) + assert_error_envelope(self.get(getattr(self, caller)), expected_status=403, expected_type_slug="authz") + + def test_the_legacy_route_refuses_the_same_callers(self): + for caller in ("no_role_user", "limited_staff", "other_course_staff", "other_course_instructor"): + assert self.get_legacy(getattr(self, caller)).status_code == 403, caller + for caller in ("course_staff", "course_instructor", "global_staff", "superuser"): + assert self.get_legacy(getattr(self, caller)).status_code == 200, caller + + def test_a_missing_course_is_not_found_for_a_privileged_caller(self): + response = self.get(self.global_staff, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + assert response.data["detail"] == "The requested course does not exist." + + def test_a_missing_course_is_refused_not_revealed_to_a_caller_without_access(self): + response = self.get(self.no_role_user, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=403, expected_type_slug="authz") + + def test_the_legacy_route_reveals_a_missing_course_to_a_caller_without_access(self): + response = self.get_legacy(self.no_role_user, course_key=MISSING_COURSE_KEY) + assert response.status_code == 404 + assert response.data["error_code"] == "course_does_not_exist" + + def test_a_ccx_course_is_refused_even_to_global_staff(self): + ccx_key = CCXLocator.from_course_locator(self.course.id, "1") + response = self.get(self.global_staff, course_key=ccx_key) + assert_error_envelope(response, expected_status=403, expected_type_slug="authz") + + def test_an_inactive_course_author_is_authenticated_then_refused_as_on_legacy(self): + inactive = UserFactory(is_active=False, password=PASSWORD) + CourseStaffRole(self.course.id).add_users(inactive) + assert_error_envelope(self.get(inactive), expected_status=403, expected_type_slug="authz") + assert self.get_legacy(inactive).status_code == 403 + + def test_a_jwt_caller_is_served(self): + client = APIClient() + response = client.get(self.url(), HTTP_AUTHORIZATION=f"JWT {create_jwt_for_user(self.course_staff)}") + assert response.status_code == 200 + + def test_a_bearer_token_is_not_accepted(self): + token = AccessTokenFactory(user=self.course_staff, application=ApplicationFactory()).token + response = APIClient().get(self.url(), HTTP_AUTHORIZATION=f"Bearer {token}") + assert_error_envelope(response, expected_status=401, expected_type_slug="authn") + + def test_the_legacy_route_still_accepts_a_bearer_token(self): + token = AccessTokenFactory(user=self.course_staff, application=ApplicationFactory()).token + response = APIClient().get(self.legacy_url(), HTTP_AUTHORIZATION=f"Bearer {token}") + assert response.status_code == 200 + + +class CourseTextbooksAuthzEnabledTest(CourseAuthoringAuthzTestMixin, TextbooksTestBase): + """Who may list a course's textbooks when course-authoring policy decides.""" + + def setUp(self): + super().setUp() + self.set_textbooks([_textbook(1)]) + + def _assert_both_routes(self, user, expected_status): + new = self.get(user) + assert new.status_code == expected_status + assert self.get_legacy(user).status_code == expected_status + return new + + def test_policy_staff_and_auditor_are_served(self): + for role in (COURSE_STAFF, COURSE_AUDITOR): + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, role.external_key, self.course.id) + response = self._assert_both_routes(user, 200) + assert response.data["results"] == [_textbook(1)] + + def test_policy_limited_staff_is_refused(self): + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, COURSE_LIMITED_STAFF.external_key, self.course.id) + assert_error_envelope(self._assert_both_routes(user, 403), expected_status=403, expected_type_slug="authz") + + def test_a_policy_role_on_another_course_is_refused(self): + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, COURSE_STAFF.external_key, self.other_course.id) + self._assert_both_routes(user, 403) + + def test_a_course_role_granted_while_policy_decides_is_honoured(self): + self._assert_both_routes(self.course_staff, 200) + + def test_an_inactive_caller_with_a_policy_role_is_served_through_a_session(self): + inactive = UserFactory(is_active=False, password=PASSWORD) + self.add_user_to_role_in_course(inactive, COURSE_STAFF.external_key, self.course.id) + self._assert_both_routes(inactive, 200) + + def test_a_caller_with_no_role_is_refused(self): + self._assert_both_routes(self.no_role_user, 403) + + def test_a_superuser_is_served_and_sees_not_found_for_a_missing_course(self): + self._assert_both_routes(self.superuser, 200) + response = self.get(self.superuser, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + + def test_a_missing_course_is_refused_to_a_caller_without_access(self): + response = self.get(self.no_role_user, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=403, expected_type_slug="authz") + + +class CourseTextbooksParityTest(TextbooksTestBase): + """The textbooks match the legacy endpoint's, spread over standard pages.""" + + def _all_pages(self, user=None, page_size=None): + """Follow the pages to the end and return every result, checking each envelope.""" + user = user or self.course_staff + results, page = [], 1 + while True: + params = {"page": page} + if page_size: + params["page_size"] = page_size + response = self.get(user, **params) + assert response.status_code == 200 + assert set(response.data) == PAGE_KEYS + results.extend(response.data["results"]) + if response.data["next"] is None: + return results + page += 1 + + def _assert_parity(self, textbooks): + """Store ``textbooks``; assert both routes return them identically, in the same key order.""" + self.set_textbooks(textbooks) + legacy = self.get_legacy(self.course_staff) + assert legacy.status_code == 200 + new_results = self._all_pages() + assert new_results == legacy.data["textbooks"] + for new_item, legacy_item in zip(new_results, legacy.data["textbooks"], strict=True): + assert list(new_item) == list(legacy_item) + return new_results + + def test_no_textbooks(self): + self.set_textbooks([]) + response = self.get(self.course_staff) + assert response.data == { + "count": 0, "num_pages": 1, "current_page": 1, "start": 0, + "next": None, "previous": None, "results": [], + } + assert self.get_legacy(self.course_staff).data == {"textbooks": []} + + def test_one_textbook(self): + assert self._assert_parity([_textbook(1)]) == [_textbook(1)] + + def test_textbooks_with_one_none_and_five_chapters(self): + textbooks = [_textbook(1, 1), _textbook(2, 0), _textbook(3, 5)] + assert self._assert_parity(textbooks) == textbooks + + def test_undeclared_stored_keys_are_dropped_on_both_routes(self): + stored = [{"id": "1T", "tab_title": "T", "x": 2, "chapters": [{"title": "c", "url": "/u", "extra": 1}]}] + expected = [{"id": "1T", "chapters": [{"title": "c", "url": "/u"}], "tab_title": "T"}] + assert self._assert_parity(stored) == expected + + def test_twelve_textbooks_are_paged_in_stored_order(self): + textbooks = [_textbook(i) for i in range(12)] + self.set_textbooks(textbooks) + first = self.get(self.course_staff) + assert first.data["count"] == 12 + assert first.data["num_pages"] == 2 + assert first.data["current_page"] == 1 + assert first.data["start"] == 0 + assert first.data["previous"] is None + assert first.data["next"] == f"http://testserver{self.url()}?page=2" + assert first.data["results"] == textbooks[:10] + second = self.get(self.course_staff, page=2) + assert second.data["current_page"] == 2 + assert second.data["start"] == 10 + assert second.data["next"] is None + assert second.data["results"] == textbooks[10:] + assert self._assert_parity(textbooks) == textbooks + + def test_page_size_is_honoured_and_capped(self): + textbooks = [_textbook(i) for i in range(101)] + self.set_textbooks(textbooks) + assert len(self.get(self.course_staff, page_size=3).data["results"]) == 3 + assert len(self.get(self.course_staff, page_size=100).data["results"]) == 100 + capped = self.get(self.course_staff, page_size=500) + assert len(capped.data["results"]) == 100 + assert capped.data["num_pages"] == 2 + + def test_a_page_past_the_end_is_not_found(self): + self.set_textbooks([_textbook(1)]) + response = self.get(self.course_staff, page=99) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + + +@ddt.ddt +class CourseTextbooksIncompleteRecordTest(TextbooksTestBase): + """Stored textbooks missing a field are served instead of failing the page.""" + + def _new_and_legacy(self, textbooks): + """Store ``textbooks``; return the new response and the legacy one, letting legacy fail with a 500.""" + self.set_textbooks(textbooks) + client = Client(raise_request_exception=False) + assert client.login(username=self.course_staff.username, password=PASSWORD) + legacy = client.get(self.legacy_url()) + return self.get(self.course_staff), legacy + + def test_a_textbook_without_an_id(self): + new, legacy = self._new_and_legacy([{"tab_title": "T", "chapters": [{"title": "c", "url": "/u"}]}]) + assert legacy.status_code == 500 + assert new.status_code == 200 + assert new.data["results"] == [{"id": None, "chapters": [{"title": "c", "url": "/u"}], "tab_title": "T"}] + + def test_a_single_file_textbook_without_chapters(self): + new, legacy = self._new_and_legacy([{"tab_title": "T", "url": "/static/book.pdf"}]) + assert legacy.status_code == 500 + assert new.data["results"] == [{"id": None, "chapters": [], "tab_title": "T"}] + + def test_a_textbook_without_a_tab_title(self): + new, legacy = self._new_and_legacy([{"id": "1T", "chapters": []}]) + assert legacy.status_code == 500 + assert new.data["results"] == [{"id": "1T", "chapters": [], "tab_title": None}] + + def test_a_null_chapter_list_is_returned_empty(self): + new, _ = self._new_and_legacy([{"id": "1T", "chapters": None, "tab_title": "T"}]) + assert new.status_code == 200 + assert new.data["results"] == [{"id": "1T", "chapters": [], "tab_title": "T"}] + + @ddt.data( + ({"title": "c"}, {"title": "c", "url": ""}), + ({"url": "/u"}, {"title": "", "url": "/u"}), + ({"title": None, "url": None}, {"title": "", "url": ""}), + ) + @ddt.unpack + def test_a_chapter_missing_a_field_is_returned_with_an_empty_string(self, stored, expected): + new, _ = self._new_and_legacy([{"id": "1T", "chapters": [stored], "tab_title": "T"}]) + assert new.status_code == 200 + assert new.data["results"] == [{"id": "1T", "chapters": [expected], "tab_title": "T"}] + + def test_one_incomplete_textbook_does_not_hide_the_others(self): + good = _textbook(1, 2) + new, legacy = self._new_and_legacy([good, {"tab_title": "Broken"}]) + assert legacy.status_code == 500 + assert new.data["results"] == [good, {"id": None, "chapters": [], "tab_title": "Broken"}] + self.set_textbooks([good]) + assert self.get_legacy(self.course_staff).data["textbooks"] == [new.data["results"][0]] + + def test_a_course_missing_from_the_modulestore_is_not_found(self): + with patch("xmodule.modulestore.mixed.MixedModuleStore.get_course", return_value=None): + response = self.get(self.course_staff) + client = Client(raise_request_exception=False) + assert client.login(username=self.course_staff.username, password=PASSWORD) + legacy = client.get(self.legacy_url()) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + assert legacy.status_code == 500 + + +class CourseTextbooksReadOnlyTest(TextbooksTestBase): + """Listing textbooks changes nothing, with course-authoring policy disabled.""" + + def test_get_and_head_write_nothing_and_send_no_signal(self): + self.set_textbooks([_textbook(1)]) + client = logged_in_client(self.course_staff) + for method in ("get", "head"): + with CaptureQueriesContext(connection) as queries, SignalRecorder() as signals: + response = getattr(client, method)(self.url()) + assert response.status_code == 200 + assert writes(queries.captured_queries) == [], method + assert signals.unexpected() == [], method + + def test_the_recorder_sees_a_signal_when_one_is_sent(self): + """The harness above would notice a signal: a model save inside it is recorded.""" + with CaptureQueriesContext(connection) as queries, SignalRecorder() as signals: + UserFactory(password=PASSWORD) + assert writes(queries.captured_queries) != [] + assert any(signal is model_signals.post_save for _, signal in signals.unexpected()) + + def test_a_jwt_request_writes_nothing(self): + self.set_textbooks([_textbook(1)]) + header = f"JWT {create_jwt_for_user(self.course_staff)}" + with CaptureQueriesContext(connection) as queries: + response = APIClient().get(self.url(), HTTP_AUTHORIZATION=header) + assert response.status_code == 200 + assert writes(queries.captured_queries) == [] + + +@ddt.ddt +class AuthorizationCacheInitialisationTest(TextbooksTestBase): + """ + The one write either route can make: the authorization engine's cache-version row. + + Role checks read that singleton row through ``get_or_create``, so the first + request against a database without it creates it. Every later request only + reads it. + """ + + @ddt.data("legacy_url", "url") + def test_only_the_first_request_creates_the_cache_version_row(self, url_method): + self.set_textbooks([_textbook(1)]) + PolicyCacheControl.objects.all().delete() + client = logged_in_client(self.course_staff) + with CaptureQueriesContext(connection) as first: + assert client.get(getattr(self, url_method)()).status_code == 200 + first_writes = writes(first.captured_queries) + assert len(first_writes) == 1 + assert first_writes[0].startswith('INSERT INTO "openedx_authz_policycachecontrol"') + with CaptureQueriesContext(connection) as second: + assert client.get(getattr(self, url_method)()).status_code == 200 + assert writes(second.captured_queries) == [] + + +class CourseTextbooksReadOnlyAuthzEnabledTest(CourseAuthoringAuthzTestMixin, TextbooksTestBase): + """Listing textbooks changes nothing when course-authoring policy decides.""" + + def test_get_and_head_write_nothing_and_send_no_signal(self): + self.set_textbooks([_textbook(1)]) + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, COURSE_STAFF.external_key, self.course.id) + client = logged_in_client(user) + for method in ("get", "head"): + with CaptureQueriesContext(connection) as queries, SignalRecorder() as signals: + response = getattr(client, method)(self.url()) + assert response.status_code == 200 + assert writes(queries.captured_queries) == [], method + assert signals.unexpected() == [], method + + +@ddt.ddt +class CourseTextbooksQueryCountTest(TextbooksTestBase): + """The new route issues no more queries than the legacy one.""" + + @ddt.data((0, 15, 13), (3, 15, 13)) + @ddt.unpack + def test_query_count(self, textbook_count, legacy_queries, new_queries): + self.set_textbooks([_textbook(i, 2) for i in range(textbook_count)]) + client = logged_in_client(self.course_staff) + # One request to each route first, so per-process caches are warm for both. + client.get(self.legacy_url()) + client.get(self.url()) + with self.assertNumQueries(legacy_queries, using="default"): + assert client.get(self.legacy_url()).status_code == 200 + with self.assertNumQueries(new_queries, using="default"): + assert client.get(self.url()).status_code == 200 + + +class CourseTextbooksUrlTest(TextbooksTestBase): + """The address, its name, and how malformed requests are answered.""" + + def test_reverse_literal(self): + key = "course-v1:edX+Demo+2026" + assert reverse(URL_NAME, kwargs={"course_key": key}) == f"/api/authoring/v2/courses/{key}/textbooks/" + + def test_each_address_resolves_to_its_own_view(self): + assert resolve(self.url()).func.cls is CourseTextbooksViewSet + assert resolve(self.legacy_url()).func.view_class is CourseTextbooksView + assert self.legacy_url() == f"/api/contentstore/v1/textbooks/{self.course.id}" + + def test_malformed_and_deprecated_keys_do_not_route(self): + client = logged_in_client(self.global_staff) + for key in ("not-a-key", "edX/DemoX/Demo"): + response = client.get(f"/api/authoring/v2/courses/{key}/textbooks/") + assert response.status_code == 404, key + # The service's HTML not-found page answers here: no JSON catch-all + # exists under this mount. + assert response["Content-Type"].startswith("text/html"), key + + def test_the_legacy_route_still_serves_a_deprecated_key(self): + assert resolve("/api/contentstore/v1/textbooks/edX/DemoX/Demo").func.view_class is CourseTextbooksView + + def test_a_slashless_address_redirects(self): + client = logged_in_client(self.global_staff) + response = client.get(self.url().rstrip("/")) + assert response.status_code == 301 + assert response["Location"] == self.url() + + def test_writes_are_not_allowed(self): + client = logged_in_client(self.global_staff) + for method in ("post", "put", "patch", "delete"): + response = getattr(client, method)(self.url()) + assert_error_envelope(response, expected_status=405) + assert response["Content-Type"] == "application/json" + + def test_an_html_only_client_is_answered_not_acceptable_in_json(self): + client = logged_in_client(self.global_staff) + response = client.get(self.url(), HTTP_ACCEPT="text/html") + assert_error_envelope(response, expected_status=406) + assert response["Content-Type"] == "application/json" + + +TEXTBOOKS_PATH = "/api/authoring/v2/courses/{course_key}/textbooks/" +LEGACY_TEXTBOOKS_PATH = "/api/contentstore/v1/textbooks/{course_id}" + + +class CourseTextbooksSchemaTest(CourseTestCase): + """What the published schema says about the endpoint.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.schema = generate_cms_schema() + + def _operation(self): + return self.schema["paths"][TEXTBOOKS_PATH]["get"] + + def test_the_path_is_published_at_its_full_address(self): + assert set(self.schema["paths"][TEXTBOOKS_PATH]) == {"get"} + assert self._operation()["tags"] == ["openedx-platform-sdk"] + + def test_the_success_body_is_the_seven_key_page_of_textbooks(self): + body = self._operation()["responses"]["200"]["content"]["application/json"]["schema"] + page = resolve_ref(self.schema, body) + assert set(page["properties"]) == PAGE_KEYS + assert set(page["required"]) == PAGE_KEYS + items = page["properties"]["results"]["items"] + assert items == {"$ref": "#/components/schemas/CourseTextbook"} + assert page["properties"]["next"]["nullable"] is True + + def test_missing_scalars_are_published_as_nullable(self): + textbook = self.schema["components"]["schemas"]["CourseTextbook"] + assert textbook["properties"]["id"]["nullable"] is True + assert textbook["properties"]["tab_title"]["nullable"] is True + assert textbook["properties"]["chapters"]["items"] == {"$ref": "#/components/schemas/TextbookChapter"} + assert "required" not in textbook + + def test_page_parameters_are_published(self): + names = {p["name"] for p in self._operation()["parameters"]} + assert names == {"course_key", "page", "page_size"} + + def test_every_error_references_the_shared_envelope(self): + responses = self._operation()["responses"] + assert set(responses) == {"200", "401", "403", "404"} + for code in ("401", "403", "404"): + assert responses[code]["content"]["application/json"]["schema"] == { + "$ref": "#/components/schemas/ErrorResponse" + }, code + + def test_the_legacy_operation_is_unchanged(self): + operation = self.schema["paths"][LEGACY_TEXTBOOKS_PATH]["get"] + assert operation["operationId"] == "v1_textbooks_retrieve" + assert "deprecated" not in operation + + +class CmsSchemaFilterTest(CourseTestCase): + """Which mounts the CMS schema admits.""" + + @staticmethod + def _admitted(path): + return bool(cms_api_filter([(path, path, "GET", object())])) + + def test_both_api_mounts_are_admitted(self): + assert self._admitted("/api/authoring/v2/courses/course-v1:a+b+c/textbooks/") + assert self._admitted("/api/contentstore/v1/textbooks/course-v1:a+b+c") + assert self._admitted("/api/courses/course-v1:a+b+c/bulk_enable_disable_discussions") + + def test_unrelated_mounts_are_refused(self): + assert not self._admitted("/api/user/v1/accounts/") + assert not self._admitted("/api/authoring/courses/") + assert not self._admitted("/authoring/v2/courses/") + + +class SpectacularSettingsTest(CourseTestCase): + """The schema settings of the environments that publish the CMS schema.""" + + SETTINGS_MODULES = ("cms/envs/devstack.py", "cms/envs/production.py") + + def _source(self, module): + return (Path(__file__).resolve().parents[7] / module).read_text() + + def test_no_prefix_is_trimmed_so_paths_are_emitted_in_full(self): + for module in self.SETTINGS_MODULES: + source = self._source(module) + assert "SCHEMA_PATH_PREFIX_TRIM" not in source, module + assert "'SCHEMA_PATH_PREFIX': r'/api/(contentstore|authoring)'," in source, module + + def test_no_server_advertises_a_single_mount_as_its_root(self): + for module in self.SETTINGS_MODULES: + assert "CMS-contentstore" not in self._source(module), module diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_video_settings.py b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_video_settings.py new file mode 100644 index 000000000000..41aef38e3e5a --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_video_settings.py @@ -0,0 +1,475 @@ +"""Tests for the v2 course video settings endpoint.""" + +import itertools +import json +from datetime import UTC, datetime, timedelta +from unittest.mock import patch + +import ddt +from ccx_keys.locator import CCXLocator +from django.db import connection +from django.db.models import signals as model_signals +from django.test import Client +from django.test.utils import CaptureQueriesContext, override_settings +from django.urls import resolve, reverse +from edx_rest_framework_extensions.testing import assert_error_envelope +from edx_toggles.toggles import WaffleSwitch +from edx_toggles.toggles.testutils import override_waffle_switch +from edxval.api import ( + create_or_update_transcript_preferences, + create_profile, + create_video, + get_3rd_party_transcription_plans, + get_transcript_credentials_state_for_org, + get_transcript_preferences, +) +from edxval.models import Video +from openedx_authz.constants.roles import COURSE_AUDITOR, COURSE_STAFF +from openedx_authz.models.engine import PolicyCacheControl +from rest_framework.test import APIClient + +from cms.djangoapps.contentstore.rest_api.v1.views.videos import CourseVideosView, VideoDownloadView, VideoUsageView +from cms.djangoapps.contentstore.rest_api.v2.views.tests.helpers import ( + PASSWORD, + SignalRecorder, + generate_cms_schema, + logged_in_client, + writes, +) +from cms.djangoapps.contentstore.rest_api.v2.views.video_settings import CourseVideoSettingsViewSet +from cms.djangoapps.contentstore.tests.utils import CourseTestCase +from cms.djangoapps.contentstore.utils import get_course_video_settings_context, reverse_course_url +from cms.djangoapps.contentstore.video_storage_handlers import send_video_status_update +from common.djangoapps.student.roles import CourseInstructorRole, CourseLimitedStaffRole, CourseStaffRole +from common.djangoapps.student.tests.factories import UserFactory +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin +from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user +from openedx.core.djangoapps.oauth_dispatch.tests.factories import AccessTokenFactory, ApplicationFactory +from xmodule.modulestore.tests.factories import CourseFactory + +URL_NAME = "authoring_v2:course_video_settings" +HANDLERS = "cms.djangoapps.contentstore.video_storage_handlers" +LEGACY_URL_NAME = "cms.djangoapps.contentstore:v1:course_videos" +MISSING_COURSE_KEY = "course-v1:NoSuch+Course+Run" +TRANSCRIPTS_FLAG = "openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled" +XPERT_FLAG = "openedx.core.djangoapps.video_config.toggles.XPERT_TRANSLATIONS_UI.is_enabled" +IMAGE_UPLOAD_SWITCH = WaffleSwitch( # pylint: disable=toggle-missing-annotation + "videos.video_image_upload_enabled", __name__ +) +SETTINGS_KEYS = { + "image_upload_url", "video_handler_url", "encodings_download_url", "default_video_image_url", + "concurrent_upload_limit", "video_supported_file_formats", "video_upload_max_file_size", + "video_image_settings", "is_video_transcript_enabled", "is_ai_translations_enabled", + "active_transcript_preferences", "transcript_credentials", "transcript_available_languages", + "video_transcript_settings", +} + + +def typed(value): + """Return ``value`` with every leaf paired with its JSON type, so ``1`` and ``1.0`` differ.""" + if isinstance(value, dict): + return {key: typed(item) for key, item in value.items()} + if isinstance(value, list): + return [typed(item) for item in value] + return (type(value).__name__, value) + + +def body(response): + """Return the decoded JSON body of ``response``.""" + return json.loads(response.content) + + +def legacy_as_new(legacy): + """ + Return the legacy body with each declared difference applied, asserting each one occurs. + + The new representation drops the always-empty ``pagination_context``, + publishes the maximum upload size as an integer, spells the transcript + download format key correctly, and never embeds the uploaded videos. + """ + expected = dict(legacy) + assert "previous_uploads" in expected + expected.pop("previous_uploads") + assert expected.pop("pagination_context") == {} + assert expected["video_upload_max_file_size"] == "5" + expected["video_upload_max_file_size"] = 5 + transcript_settings = dict(expected["video_transcript_settings"]) + assert transcript_settings["trancript_download_file_format"] == "srt" + transcript_settings["transcript_download_file_format"] = transcript_settings.pop("trancript_download_file_format") + expected["video_transcript_settings"] = transcript_settings + return expected + + +class VideoSettingsTestBase(CourseTestCase): + """A course, its callers, and helpers for creating its uploaded videos.""" + + def setUp(self): + super().setUp() + self.now = datetime.now(UTC) + self.other_course = CourseFactory.create(org="OtherX", course="Other", run="Run") + + self.no_role_user = UserFactory(password=PASSWORD) + self.course_staff = UserFactory(password=PASSWORD) + CourseStaffRole(self.course.id).add_users(self.course_staff) + self.course_instructor = UserFactory(password=PASSWORD) + CourseInstructorRole(self.course.id).add_users(self.course_instructor) + self.limited_staff = UserFactory(password=PASSWORD) + CourseLimitedStaffRole(self.course.id).add_users(self.limited_staff) + self.other_course_staff = UserFactory(password=PASSWORD) + CourseStaffRole(self.other_course.id).add_users(self.other_course_staff) + self.other_course_instructor = UserFactory(password=PASSWORD) + CourseInstructorRole(self.other_course.id).add_users(self.other_course_instructor) + self.global_staff = UserFactory(is_staff=True, password=PASSWORD) + self.superuser = UserFactory(is_superuser=True, password=PASSWORD) + # Role checks create the authorization engine's cache-version row on + # first use; create it here so request-level measurements exclude it. + PolicyCacheControl.get() + create_profile("desktop_mp4") + create_profile("mobile_low") + + def add_video(self, edx_video_id, status, hours_old=1, course=None, image=None, encodings=(), **fields): + """Create a VAL video attached to ``course`` (default: this course), ``hours_old`` hours ago.""" + course_id = str((course or self.course).id) + create_video({ + "edx_video_id": edx_video_id, + "client_video_id": f"{edx_video_id}.mp4", + "duration": 12.5, + "status": status, + "courses": [{course_id: image}] if image else [course_id], + "encoded_videos": list(encodings), + **fields, + }) + Video.objects.filter(edx_video_id=edx_video_id).update(created=self.now - timedelta(hours=hours_old)) + + def url(self, course_key=None): + return reverse(URL_NAME, kwargs={"course_key": str(course_key or self.course.id)}) + + def legacy_url(self, course_key=None): + return reverse(LEGACY_URL_NAME, kwargs={"course_id": str(course_key or self.course.id)}) + + def get(self, user, course_key=None, method="get", **params): + client = logged_in_client(user) if user is not None else APIClient() + return getattr(client, method)(self.url(course_key), params) + + def get_legacy(self, user, course_key=None): + client = logged_in_client(user) if user is not None else APIClient() + return client.get(self.legacy_url(course_key)) + + +@ddt.ddt +class CourseVideoSettingsAccessTest(VideoSettingsTestBase): + """Who may read a course's video settings.""" + + @ddt.data("course_staff", "course_instructor", "global_staff", "superuser") + def test_allowed_callers_are_served(self, caller): + for method in ("get", "head"): + response = self.get(getattr(self, caller), method=method) + assert response.status_code == 200, (caller, method) + assert set(self.get(getattr(self, caller)).data) == SETTINGS_KEYS + + def test_anonymous_is_refused_with_a_challenge(self): + response = self.get(None) + assert_error_envelope(response, expected_status=401, expected_type_slug="authn") + assert response["WWW-Authenticate"] == 'JWT realm="api"' + + @ddt.data("no_role_user", "limited_staff", "other_course_staff", "other_course_instructor") + def test_callers_without_studio_access_to_this_course_are_refused(self, caller): + for method in ("get", "head"): + response = self.get(getattr(self, caller), method=method) + assert response.status_code == 403, (caller, method) + assert_error_envelope(self.get(getattr(self, caller)), expected_status=403, expected_type_slug="authz") + assert self.get_legacy(getattr(self, caller)).status_code == 403 + + def test_a_missing_course_is_not_found_for_a_privileged_caller(self): + response = self.get(self.global_staff, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + assert response.data["detail"] == "The requested course does not exist." + + def test_a_missing_course_is_refused_not_revealed_to_a_caller_without_access(self): + response = self.get(self.no_role_user, course_key=MISSING_COURSE_KEY) + assert_error_envelope(response, expected_status=403, expected_type_slug="authz") + legacy = self.get_legacy(self.no_role_user, course_key=MISSING_COURSE_KEY) + assert legacy.status_code == 404 + assert legacy.data["error_code"] == "course_does_not_exist" + + def test_a_course_missing_from_the_modulestore_is_not_found(self): + client = Client(raise_request_exception=False) + assert client.login(username=self.course_staff.username, password=PASSWORD) + with patch("xmodule.modulestore.mixed.MixedModuleStore.get_course", return_value=None): + response = self.get(self.course_staff) + legacy = client.get(self.legacy_url()) + assert_error_envelope(response, expected_status=404, expected_type_slug="not-found") + assert legacy.status_code == 500 + + def test_a_ccx_course_is_refused_even_to_global_staff(self): + ccx_key = CCXLocator.from_course_locator(self.course.id, "1") + assert_error_envelope(self.get(self.global_staff, course_key=ccx_key), expected_status=403, + expected_type_slug="authz") + + def test_an_inactive_course_author_is_authenticated_then_refused_as_on_legacy(self): + inactive = UserFactory(is_active=False, password=PASSWORD) + CourseStaffRole(self.course.id).add_users(inactive) + assert_error_envelope(self.get(inactive), expected_status=403, expected_type_slug="authz") + assert self.get_legacy(inactive).status_code == 403 + + def test_a_jwt_caller_is_served(self): + header = f"JWT {create_jwt_for_user(self.course_staff)}" + assert APIClient().get(self.url(), HTTP_AUTHORIZATION=header).status_code == 200 + + def test_a_bearer_token_is_not_accepted_where_legacy_accepts_it(self): + token = AccessTokenFactory(user=self.course_staff, application=ApplicationFactory()).token + response = APIClient().get(self.url(), HTTP_AUTHORIZATION=f"Bearer {token}") + assert_error_envelope(response, expected_status=401, expected_type_slug="authn") + assert APIClient().get(self.legacy_url(), HTTP_AUTHORIZATION=f"Bearer {token}").status_code == 200 + + def test_uploaded_videos_are_never_embedded(self): + """Legacy's previous_uploads is not offered, whatever the query string asks for.""" + self.add_video("encoded", "file_complete") + for params in ({}, {"view": "full"}): + assert set(self.get(self.course_staff, **params).data) == SETTINGS_KEYS, params + + +class CourseVideoSettingsAuthzEnabledTest(CourseAuthoringAuthzTestMixin, VideoSettingsTestBase): + """Access when course-authoring policy is enabled: the same callers as legacy, and no others.""" + + def test_a_policy_auditor_is_refused_as_on_legacy(self): + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, COURSE_AUDITOR.external_key, self.course.id) + assert_error_envelope(self.get(user), expected_status=403, expected_type_slug="authz") + assert self.get_legacy(user).status_code == 403 + + def test_a_policy_staff_role_is_served_as_on_legacy(self): + """Studio read access resolves the course staff role through its policy assignment.""" + user = UserFactory(password=PASSWORD) + self.add_user_to_role_in_course(user, COURSE_STAFF.external_key, self.course.id) + assert self.get(user).status_code == 200 + assert self.get_legacy(user).status_code == 200 + + def test_a_legacy_course_role_is_still_served(self): + assert self.get(self.course_staff).status_code == 200 + assert self.get_legacy(self.course_staff).status_code == 200 + + +@ddt.ddt +class CourseVideoSettingsParityTest(VideoSettingsTestBase): + """The settings match legacy's, apart from the declared differences, under every flag.""" + + def setUp(self): + super().setUp() + create_or_update_transcript_preferences( + str(self.course.id), provider="Cielo24", cielo24_fidelity="PROFESSIONAL", + cielo24_turnaround="PRIORITY", preferred_languages=["en"], video_source_language="en", + ) + self.add_video("encoded", "file_complete") + + @ddt.data(*itertools.product((False, True), (False, True), (False, True))) + @ddt.unpack + def test_settings_match_legacy(self, transcripts, xpert, image_upload): + with patch(TRANSCRIPTS_FLAG, return_value=transcripts), patch(XPERT_FLAG, return_value=xpert), \ + override_waffle_switch(IMAGE_UPLOAD_SWITCH, image_upload): + full = body(self.get(self.course_staff)) + legacy = body(self.get_legacy(self.course_staff)) + + assert "pagination_context" not in full + assert "trancript_download_file_format" not in full["video_transcript_settings"] + assert typed(full) == typed(legacy_as_new(legacy)) + + assert full["is_video_transcript_enabled"] is transcripts + assert full["is_ai_translations_enabled"] is xpert + assert full["video_image_settings"]["video_image_upload_enabled"] is image_upload + transcript_settings = full["video_transcript_settings"] + if transcripts: + assert full["active_transcript_preferences"] == get_transcript_preferences(str(self.course.id)) + assert full["transcript_credentials"] == get_transcript_credentials_state_for_org(self.course.id.org) + assert transcript_settings["transcription_plans"] == get_3rd_party_transcription_plans() + assert transcript_settings["transcript_preferences_handler_url"] == reverse_course_url( + "transcript_preferences_handler", str(self.course.id) + ) + else: + assert full["active_transcript_preferences"] is None + assert full["transcript_credentials"] is None + assert transcript_settings["transcription_plans"] is None + assert transcript_settings["transcript_preferences_handler_url"] is None + + def test_the_settings_come_from_the_same_helper_as_legacy(self): + video_settings = get_course_video_settings_context(self.course) + assert set(video_settings) == SETTINGS_KEYS + assert video_settings["video_upload_max_file_size"] == 5 + + +class CourseVideoSettingsReadOnlyTest(VideoSettingsTestBase): + """Reading the settings changes nothing, even when an upload has gone stale.""" + + def setUp(self): + super().setUp() + self.add_video("stale", "upload", hours_old=30) + self.add_video("fresh", "upload", hours_old=2) + self.saved = [] + model_signals.post_save.connect(self._record_save, sender=Video) + self.addCleanup(model_signals.post_save.disconnect, self._record_save, sender=Video) + + def _record_save(self, sender, instance, **kwargs): # pylint: disable=unused-argument + self.saved.append(instance.edx_video_id) + + def _observe(self, client, url, method="get", **params): + """Make one request; return it with the writes, signals and status-update calls it caused.""" + send_update = patch(f"{HANDLERS}.send_video_status_update", wraps=send_video_status_update) + with CaptureQueriesContext(connection) as queries, SignalRecorder() as signals, send_update as sent: + response = getattr(client, method)(url, params) + return response, writes(queries.captured_queries), signals.unexpected(), sent.call_count + + def test_get_and_head_write_nothing(self): + client = logged_in_client(self.course_staff) + for method in ("get", "head"): + response, statements, signals, sent = self._observe(client, self.url(), method) + assert response.status_code == 200, method + assert statements == [], method + assert signals == [], method + assert sent == 0, method + assert not self.saved, method + assert Video.objects.get(edx_video_id="stale").status == "upload", method + + def test_a_jwt_request_writes_nothing(self): + header = f"JWT {create_jwt_for_user(self.course_staff)}" + with CaptureQueriesContext(connection) as queries: + response = APIClient().get(self.url(), HTTP_AUTHORIZATION=header) + assert response.status_code == 200 + assert writes(queries.captured_queries) == [] + assert Video.objects.get(edx_video_id="stale").status == "upload" + + def test_the_legacy_route_still_marks_the_stale_upload_failed(self): + """The same observation made on the legacy route sees its write, so the harness above can fail.""" + response, statements, signals, sent = self._observe(logged_in_client(self.course_staff), self.legacy_url()) + assert response.status_code == 200 + assert Video.objects.get(edx_video_id="stale").status == "upload_failed" + assert Video.objects.get(edx_video_id="fresh").status == "upload" + assert any(sql.startswith('UPDATE "edxval_video"') for sql in statements) + assert any(signal is model_signals.post_save for _, signal in signals) + assert sent == 1 + assert self.saved == ["stale"] + + @override_settings(ENABLE_VIDEO_UPLOAD_PIPELINE=True) + def test_the_studio_video_listing_still_marks_the_stale_upload_failed(self): + self.course.video_upload_pipeline = {"course_video_upload_token": "course-token"} + self.save_course() + response = self.client.get(reverse_course_url("videos_handler", self.course.id), HTTP_ACCEPT="application/json") + assert response.status_code == 200 + assert Video.objects.get(edx_video_id="stale").status == "upload_failed" + + +class CourseVideoSettingsQueryCountTest(VideoSettingsTestBase): + """The settings cost a fixed number of queries, however many videos the course has.""" + + @override_settings(ENABLE_VIDEO_UPLOAD_PIPELINE=True) + def test_the_query_count_does_not_grow_with_uploads(self): + client = logged_in_client(self.course_staff) + client.get(self.url()) + with self.assertNumQueries(13, using="default"): + assert client.get(self.url()).status_code == 200 + for index in range(3): + self.add_video(f"video-{index}", "file_complete", hours_old=index + 1) + with self.assertNumQueries(13, using="default"): + assert client.get(self.url()).status_code == 200 + + +class CourseVideoSettingsUrlTest(VideoSettingsTestBase): + """The address, its name, and how malformed requests are answered.""" + + def test_reverse_literal(self): + key = "course-v1:edX+Demo+2026" + assert reverse(URL_NAME, kwargs={"course_key": key}) == f"/api/authoring/v2/courses/{key}/video_settings/" + + def test_each_address_resolves_to_its_own_view(self): + assert resolve(self.url()).func.cls is CourseVideoSettingsViewSet + assert resolve(self.legacy_url()).func.view_class is CourseVideosView + assert self.legacy_url() == f"/api/contentstore/v1/videos/{self.course.id}" + + def test_malformed_and_deprecated_keys_do_not_route(self): + client = logged_in_client(self.global_staff) + for key in ("not-a-key", "edX/DemoX/Demo"): + response = client.get(f"/api/authoring/v2/courses/{key}/video_settings/") + assert response.status_code == 404, key + # The service's HTML not-found page answers here: no JSON catch-all + # exists under this mount. + assert response["Content-Type"].startswith("text/html"), key + + def test_the_legacy_route_still_serves_a_deprecated_key(self): + assert resolve("/api/contentstore/v1/videos/edX/DemoX/Demo").func.view_class is CourseVideosView + + def test_a_slashless_address_redirects(self): + response = logged_in_client(self.global_staff).get(self.url().rstrip("/")) + assert response.status_code == 301 + assert response["Location"] == self.url() + + def test_writes_are_not_allowed(self): + client = logged_in_client(self.global_staff) + for method in ("post", "put", "patch", "delete"): + response = getattr(client, method)(self.url()) + assert_error_envelope(response, expected_status=405) + assert response["Content-Type"] == "application/json" + + def test_an_html_only_client_is_answered_not_acceptable_in_json(self): + response = logged_in_client(self.global_staff).get(self.url(), HTTP_ACCEPT="text/html") + assert_error_envelope(response, expected_status=406) + assert response["Content-Type"] == "application/json" + + +VIDEO_SETTINGS_PATH = "/api/authoring/v2/courses/{course_key}/video_settings/" +LEGACY_VIDEOS_PATH = "/api/contentstore/v1/videos/{course_id}" + + +class CourseVideoSettingsSchemaTest(CourseTestCase): + """What the published schema says about the endpoint.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.schema = generate_cms_schema() + cls.components = cls.schema["components"]["schemas"] + + def _operation(self): + return self.schema["paths"][VIDEO_SETTINGS_PATH]["get"] + + def test_the_path_is_published_at_its_full_address(self): + assert set(self.schema["paths"][VIDEO_SETTINGS_PATH]) == {"get"} + assert self._operation()["tags"] == ["openedx-platform-sdk"] + + def test_the_success_body_is_the_settings(self): + success = self._operation()["responses"]["200"]["content"]["application/json"]["schema"] + assert success == {"$ref": "#/components/schemas/CourseVideoSettings"} + assert set(self.components["CourseVideoSettings"]["properties"]) == SETTINGS_KEYS + + def test_only_the_course_key_is_a_parameter(self): + assert [p["name"] for p in self._operation()["parameters"]] == ["course_key"] + + def test_field_types_match_what_is_returned(self): + settings = self.components["CourseVideoSettings"]["properties"] + assert settings["video_upload_max_file_size"]["type"] == "integer" + assert settings["transcript_credentials"]["nullable"] is True + transcript = self.components["VideoSettingsTranscript"]["properties"] + assert "transcript_download_file_format" in transcript + assert "trancript_download_file_format" not in transcript + + def test_every_error_references_the_shared_envelope(self): + responses = self._operation()["responses"] + assert set(responses) == {"200", "401", "403", "404"} + for code in ("401", "403", "404"): + assert responses[code]["content"]["application/json"]["schema"] == { + "$ref": "#/components/schemas/ErrorResponse" + }, code + + def test_the_legacy_video_operations_are_unchanged(self): + operation = self.schema["paths"][LEGACY_VIDEOS_PATH]["get"] + assert operation["operationId"] == "v1_videos_retrieve" + assert "deprecated" not in operation + for path, view in self._legacy_sibling_paths(): + for operation in self.schema["paths"][path].values(): + assert "deprecated" not in operation, (path, view) + + def _legacy_sibling_paths(self): + """Return the legacy usage and download paths published beside the legacy videos path.""" + paths = [(p, v) for p in self.schema["paths"] for v in (VideoUsageView, VideoDownloadView) + if p.startswith("/api/contentstore/v1/videos/{course_id}/") and + p.endswith("/usage" if v is VideoUsageView else "/download")] + assert len(paths) == 2 + return paths diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/textbooks.py b/cms/djangoapps/contentstore/rest_api/v2/views/textbooks.py new file mode 100644 index 000000000000..4f580852206e --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/views/textbooks.py @@ -0,0 +1,109 @@ +"""API views for course PDF textbooks.""" + +from drf_spectacular.utils import OpenApiResponse, extend_schema +from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication +from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser +from edx_rest_framework_extensions.errors import ErrorResponseSerializer +from edx_rest_framework_extensions.mixins import StandardizedErrorMixin +from edx_rest_framework_extensions.paginators import DefaultPagination, IterablePaginationMixin +from rest_framework import viewsets +from rest_framework.permissions import IsAuthenticated + +from cms.djangoapps.contentstore.rest_api.v2.serializers.textbooks import CourseTextbookSerializer +from cms.djangoapps.contentstore.rest_api.v2.textbooks_service import get_course_textbooks +from cms.djangoapps.contentstore.views.permissions import CanViewPagesAndResources + + +class CourseTextbookPagination(DefaultPagination): + """ + The standard page envelope, with a schema that describes all seven of its keys. + + ``DefaultPagination`` returns seven keys but inherits a response schema + that documents only four, so generated clients would drop ``num_pages``, + ``current_page`` and ``start``. + + Delete this subclass and use ``DefaultPagination`` directly once + edx-drf-extensions documents all seven keys in + ``DefaultPagination.get_paginated_response_schema``. + """ + + def get_paginated_response_schema(self, schema): + return { + "type": "object", + "required": ["count", "num_pages", "current_page", "start", "next", "previous", "results"], + "properties": { + "count": {"type": "integer", "description": "Total number of items."}, + "num_pages": {"type": "integer", "description": "Total number of pages."}, + "current_page": {"type": "integer", "description": "Number of this page, starting at 1."}, + "start": {"type": "integer", "description": "Zero-based index of the first item on this page."}, + "next": { + "type": "string", + "format": "uri", + "nullable": True, + "description": "Address of the next page, or null on the last page.", + }, + "previous": { + "type": "string", + "format": "uri", + "nullable": True, + "description": "Address of the previous page, or null on the first page.", + }, + "results": schema, + }, + } + + +@extend_schema(tags=["openedx-platform-sdk"]) +class CourseTextbooksViewSet(StandardizedErrorMixin, IterablePaginationMixin, viewsets.ViewSet): + """ + The PDF textbooks of one course, in the order the course stores them. + + Callable by anyone allowed to view the course's pages and resources: on + courses using policy-based authoring permissions that policy decides, + elsewhere Studio read access to the course is required. Permission is + checked before the course is looked up, so a caller without access is + refused whether or not the course exists. + + A plain ``ViewSet``: textbooks are a settings field of the course in the + modulestore, not database rows. + """ + + # Studio authors reach this endpoint from the authoring UI while their + # account is still inactive, which the default session class rejects. + authentication_classes = (JwtAuthentication, SessionAuthenticationAllowInactiveUser) + permission_classes = (IsAuthenticated, CanViewPagesAndResources) + serializer_class = CourseTextbookSerializer + pagination_class = CourseTextbookPagination + + def get_serializer(self, *args, **kwargs): + """Instantiate the textbook serializer declared on this viewset.""" + return self.serializer_class(*args, **kwargs) + + @extend_schema( + summary="List a course's PDF textbooks", + description=( + "Returns one page of the course's PDF textbooks, each with its chapters, in the " + "order the course stores them, which is also the order of the course's textbook tabs." + ), + responses={ + 200: OpenApiResponse( + response=CourseTextbookSerializer, + description="A page of the course's textbooks.", + ), + 401: OpenApiResponse( + response=ErrorResponseSerializer, + description="The requester is not authenticated.", + ), + 403: OpenApiResponse( + response=ErrorResponseSerializer, + description="The requester cannot view the pages and resources of this course.", + ), + 404: OpenApiResponse( + response=ErrorResponseSerializer, + description="The course does not exist, or the requested page is out of range.", + ), + }, + ) + def list(self, request, course_key): + """Return one page of the course's textbooks.""" + return self.paginate_iterable(request, get_course_textbooks(course_key)) diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/video_settings.py b/cms/djangoapps/contentstore/rest_api/v2/views/video_settings.py new file mode 100644 index 000000000000..b6d8dfef098d --- /dev/null +++ b/cms/djangoapps/contentstore/rest_api/v2/views/video_settings.py @@ -0,0 +1,72 @@ +"""API views for a course's video upload and transcript settings.""" + +from drf_spectacular.utils import OpenApiResponse, extend_schema +from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication +from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser +from edx_rest_framework_extensions.errors import ErrorResponseSerializer +from edx_rest_framework_extensions.mixins import StandardizedErrorMixin +from rest_framework import viewsets +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response + +from cms.djangoapps.contentstore.rest_api.v2.serializers.video_settings import CourseVideoSettingsSerializer +from cms.djangoapps.contentstore.rest_api.v2.video_settings_service import get_course_video_settings +from cms.djangoapps.contentstore.views.permissions import HasStudioReadAccess + + +@extend_schema(tags=["openedx-platform-sdk"]) +class CourseVideoSettingsViewSet(StandardizedErrorMixin, viewsets.ViewSet): + """ + The video upload and transcript settings of one course. + + Only the settings are returned. The course's uploaded videos, which legacy + v1 embedded as ``previous_uploads``, are listed by their own paginated + collection at ``/api/authoring/v1/courses/{course_key}/videos/``. + + Callable by anyone with Studio read access to the course. Permission is + checked before the course is looked up, so a caller without access is + refused whether or not the course exists. + + A plain ``ViewSet``: the settings come from the modulestore, Django + settings, feature toggles and the video pipeline's own API. + """ + + # Studio authors reach this endpoint from the authoring UI while their + # account is still inactive, which the default session class rejects. + authentication_classes = (JwtAuthentication, SessionAuthenticationAllowInactiveUser) + permission_classes = (IsAuthenticated, HasStudioReadAccess) + serializer_class = CourseVideoSettingsSerializer + + def get_serializer(self, *args, **kwargs): + """Instantiate the settings serializer declared on this viewset.""" + return self.serializer_class(*args, **kwargs) + + @extend_schema( + summary="Retrieve a course's video settings", + description=( + "Returns the course's video upload limits, thumbnail image limits, transcript settings " + "and the Studio addresses that act on them. The course's uploaded videos are listed by " + "`GET /api/authoring/v1/courses/{course_key}/videos/`." + ), + responses={ + 200: OpenApiResponse( + response=CourseVideoSettingsSerializer, + description="The course's video settings.", + ), + 401: OpenApiResponse( + response=ErrorResponseSerializer, + description="The requester is not authenticated.", + ), + 403: OpenApiResponse( + response=ErrorResponseSerializer, + description="The requester does not have Studio read access to this course.", + ), + 404: OpenApiResponse( + response=ErrorResponseSerializer, + description="The course does not exist.", + ), + }, + ) + def retrieve(self, request, course_key): + """Return the course's video settings.""" + return Response(CourseVideoSettingsSerializer(get_course_video_settings(course_key)).data) diff --git a/cms/djangoapps/contentstore/utils.py b/cms/djangoapps/contentstore/utils.py index b1b39308020d..345a8b6ec0ce 100644 --- a/cms/djangoapps/contentstore/utils.py +++ b/cms/djangoapps/contentstore/utils.py @@ -1750,6 +1750,32 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None): It is used for both DRF and django views. """ + from .video_storage_handlers import _get_index_videos + + course = course_block + if not course: + with modulestore().bulk_operations(course_key): + course = modulestore().get_course(course_key) + + video_settings_context = get_course_video_settings_context(course) + previous_uploads, pagination_context = _get_index_videos(course, pagination_conf) + course_video_context = { + 'context_course': course, + **video_settings_context, + 'previous_uploads': previous_uploads, + 'pagination_context': pagination_context, + } + return course_video_context + + +def get_course_video_settings_context(course): + """ + Return the video upload and transcript settings of ``course``. + + This is the context of ``get_course_videos_context`` without the course + itself and without the course's previous uploads. + """ + from edx_toggles.toggles import WaffleSwitch from edxval.api import ( get_3rd_party_transcription_plans, @@ -1763,7 +1789,7 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None): Transcript, # pylint: disable=wrong-import-order ) - from .video_storage_handlers import _get_default_video_image_url, _get_index_videos, get_all_transcript_languages + from .video_storage_handlers import _get_default_video_image_url, get_all_transcript_languages VIDEO_SUPPORTED_FILE_FORMATS = { '.mp4': 'video/mp4', @@ -1775,21 +1801,13 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None): 'videos.video_image_upload_enabled', __name__ ) - course = course_block - if not course: - with modulestore().bulk_operations(course_key): - course = modulestore().get_course(course_key) - is_video_transcript_enabled = VideoTranscriptEnabledFlag.feature_enabled(course.id) is_ai_translations_enabled = use_xpert_translations_component(course.id) - previous_uploads, pagination_context = _get_index_videos(course, pagination_conf) - course_video_context = { - 'context_course': course, + course_video_settings_context = { 'image_upload_url': reverse_course_url('video_images_handler', str(course.id)), 'video_handler_url': reverse_course_url('videos_handler', str(course.id)), 'encodings_download_url': reverse_course_url('video_encodings_download', str(course.id)), 'default_video_image_url': _get_default_video_image_url(), - 'previous_uploads': previous_uploads, 'concurrent_upload_limit': settings.VIDEO_UPLOAD_PIPELINE.get('CONCURRENT_UPLOAD_LIMIT', 0), 'video_supported_file_formats': list(VIDEO_SUPPORTED_FILE_FORMATS.keys()), 'video_upload_max_file_size': VIDEO_UPLOAD_MAX_FILE_SIZE_GB, @@ -1812,10 +1830,9 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None): 'transcript_delete_handler_url': reverse_course_url('transcript_delete_handler', str(course.id)), 'trancript_download_file_format': Transcript.SRT }, - 'pagination_context': pagination_context } if is_video_transcript_enabled: - course_video_context['video_transcript_settings'].update({ + course_video_settings_context['video_transcript_settings'].update({ 'transcript_preferences_handler_url': reverse_course_url( 'transcript_preferences_handler', str(course.id) @@ -1826,10 +1843,12 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None): ), 'transcription_plans': get_3rd_party_transcription_plans(), }) - course_video_context['active_transcript_preferences'] = get_transcript_preferences(str(course.id)) + course_video_settings_context['active_transcript_preferences'] = get_transcript_preferences(str(course.id)) # Cached state for transcript providers' credentials (org-specific) - course_video_context['transcript_credentials'] = get_transcript_credentials_state_for_org(course.id.org) - return course_video_context + course_video_settings_context['transcript_credentials'] = get_transcript_credentials_state_for_org( + course.id.org + ) + return course_video_settings_context def get_course_index_context(request, course_key, course_block=None): diff --git a/cms/djangoapps/contentstore/views/permissions.py b/cms/djangoapps/contentstore/views/permissions.py index 6d63c7a124f4..6450a3bfcc3e 100644 --- a/cms/djangoapps/contentstore/views/permissions.py +++ b/cms/djangoapps/contentstore/views/permissions.py @@ -2,9 +2,12 @@ Custom permissions for the content store views. """ +from openedx_authz.constants.permissions import COURSES_VIEW_PAGES_AND_RESOURCES from rest_framework.permissions import BasePermission -from common.djangoapps.student.auth import has_studio_write_access +from common.djangoapps.student.auth import has_studio_read_access, 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 validate_course_key @@ -20,3 +23,38 @@ def has_permission(self, request, view): course_key_string = view.kwargs.get("course_key_string") course_key = validate_course_key(course_key_string) return has_studio_write_access(request.user, course_key) + + +class CanViewPagesAndResources(BasePermission): + """ + Allow callers who may view the pages and resources of the course in the path. + + Reads the parsed ``course_key`` URL argument. On courses where + policy-based authoring permissions are enabled, only that policy is + consulted; elsewhere the caller needs Studio read access to the course. + """ + + def has_permission(self, request, view): + course_key = view.kwargs.get("course_key") + if course_key is None: + return False + return user_has_course_permission( + request.user, + COURSES_VIEW_PAGES_AND_RESOURCES.identifier, + course_key, + LegacyAuthoringPermission.READ, + ) + + +class HasStudioReadAccess(BasePermission): + """ + Allow callers with Studio read access to the course in the path. + + Reads the parsed ``course_key`` URL argument. + """ + + def has_permission(self, request, view): + course_key = view.kwargs.get("course_key") + if course_key is None: + return False + return has_studio_read_access(request.user, course_key) diff --git a/cms/urls.py b/cms/urls.py index 0e089329ba88..8c13d112610a 100644 --- a/cms/urls.py +++ b/cms/urls.py @@ -413,6 +413,7 @@ # for the deprecation window. urlpatterns += [ path('api/authoring/v1/', include('cms.djangoapps.contentstore.rest_api.v1.authoring_urls')), + path('api/authoring/v2/', include('cms.djangoapps.contentstore.rest_api.v2.authoring_urls')), ] # Content tagging