diff --git a/lms/views/api/canvas/pages.py b/lms/views/api/canvas/pages.py index 12b5436541..ab6dcc025d 100644 --- a/lms/views/api/canvas/pages.py +++ b/lms/views/api/canvas/pages.py @@ -1,19 +1,27 @@ +import logging import re from pyramid.view import view_config, view_defaults from lms.security import Permissions from lms.services.canvas import CanvasService +from lms.services.exceptions import CanvasAPIError, FileNotFoundInCourse from lms.validation.authentication import BearerTokenSchema from lms.views import helpers +LOG = logging.getLogger(__name__) + # A regex for parsing the COURSE_ID and PAGE_ID parts out of one of our custom -# canvas://file/course/COURSE_ID/page_id/PAGE_ID URLs. +# canvas://page/course/COURSE_ID/page_id/PAGE_ID URLs. DOCUMENT_URL_REGEX = re.compile( r"canvas:\/\/page\/course\/(?P[^\/]*)\/page_id\/(?P[^\/]*)" ) +class PageNotFoundInCourse(FileNotFoundInCourse): + ... + + @view_defaults(permission=Permissions.API, renderer="json") class PagesAPIViews: def __init__(self, request): @@ -43,11 +51,72 @@ def list_pages(self): @view_config(request_method="GET", route_name="canvas_api.pages.via_url") def via_url(self): - application_instance = self.request.lti_user.application_instance + course_copy_plugin = self.request.product.plugin.course_copy + current_course = self.request.find_service(name="course").get_by_context_id( + self.request.lti_user.lti.course_id, raise_on_missing=True + ) + current_course_id = str( + current_course.extra["canvas"]["custom_canvas_course_id"] + ) assignment = self.request.find_service(name="assignment").get_assignment( - application_instance.tool_consumer_instance_guid, + self.request.lti_user.application_instance.tool_consumer_instance_guid, self.request.lti_user.lti.assignment_id, ) + document_url = assignment.document_url + document_course_id, document_page_id = self._parse_document_url(document_url) + + effective_page_id = None + if current_course_id == document_course_id: + # Not in a course copy scenario, use the IDs from the document_url + effective_page_id = document_page_id + LOG.debug("Via URL for page in the same course. %s", document_url) + + mapped_page_id = current_course.get_mapped_page_id(document_page_id) + if not effective_page_id and mapped_page_id != document_page_id: + effective_page_id = mapped_page_id + LOG.debug( + "Via URL for page already mapped for course copy. Document: %s, course: %s, mapped page_id: %s", + document_url, + current_course_id, + mapped_page_id, + ) + + if not effective_page_id: + found_page = course_copy_plugin.find_matching_page_in_course( + document_page_id, current_course_id + ) + if not found_page: + # We couldn't fix course copy, there might be something else going on + # or maybe teacher never launched before a student. + LOG.debug( + "Via URL for page, couldn't find page in the new course. Document: %s, course: %s.", + document_url, + current_course_id, + ) + raise PageNotFoundInCourse( + "canvas_page_not_found_in_course", document_page_id + ) + + # Store a mapping so we don't have to re-search next time. + current_course.set_mapped_page_id(document_page_id, found_page.lms_id) + effective_page_id = found_page.lms_id + LOG.debug( + "Via URL for page, found page in the new course. Document: %s, course: %s, new page id: %s", + document_url, + current_course_id, + found_page.lms_id, + ) + + # Try to access the page + # We don't need the result of this exact call but we accomplishes two things here: + # We can check that we indeed have access to this page, if we don't we try to fix any course copy related issues. + # We make sure that we have a recent Oauth2 token to make a request later in the proxying endpoint. + try: + _ = self.canvas.api.pages.page(current_course_id, effective_page_id) + except CanvasAPIError as err: + raise PageNotFoundInCourse( + "canvas_page_not_found_in_course", effective_page_id + ) from err # We build a token to authorize the view that fetches the actual # canvas pages content as the user making this request. @@ -60,7 +129,8 @@ def via_url(self): self.request.route_url( "canvas_api.pages.proxy", _query={ - "document_url": assignment.document_url, + "course_id": current_course_id, + "page_id": effective_page_id, "authorization": auth_token, }, ), @@ -74,11 +144,12 @@ def via_url(self): ) def proxy(self): """Proxy the contents of a canvas page.""" - document_url_match = DOCUMENT_URL_REGEX.search( - self.request.params["document_url"] + course_id, page_id = ( + self.request.params["course_id"], + self.request.params["page_id"], ) - course_id = document_url_match["course_id"] - page = self.canvas.api.pages.page(course_id, document_url_match["page_id"]) + + page = self.canvas.api.pages.page(course_id, page_id) return { "canonical_url": page.canonical_url( self.request.lti_user.application_instance.lms_host(), course_id @@ -86,3 +157,11 @@ def proxy(self): "title": page.title, "body": page.body, } + + @staticmethod + def _parse_document_url(document_url): + document_url_match = DOCUMENT_URL_REGEX.search(document_url) + course_id = document_url_match["course_id"] + page_id = document_url_match["page_id"] + + return course_id, page_id diff --git a/tests/unit/lms/views/api/canvas/pages_test.py b/tests/unit/lms/views/api/canvas/pages_test.py index 5d319621da..c214132ba3 100644 --- a/tests/unit/lms/views/api/canvas/pages_test.py +++ b/tests/unit/lms/views/api/canvas/pages_test.py @@ -1,11 +1,12 @@ import random -from unittest.mock import sentinel +from unittest.mock import Mock, sentinel import pytest from h_matchers import Any from lms.services.canvas_api._pages import CanvasPage -from lms.views.api.canvas.pages import PagesAPIViews +from lms.services.exceptions import CanvasAPIError +from lms.views.api.canvas.pages import PageNotFoundInCourse, PagesAPIViews @pytest.mark.usefixtures( @@ -34,6 +35,7 @@ def test_list_pages(self, canvas_service, pyramid_request, pages): ) canvas_service.api.pages.list.assert_called_once_with(course_id) + @pytest.mark.usefixtures("course_copy_plugin") def test_via_url( self, helpers, @@ -42,8 +44,14 @@ def test_via_url( assignment_service, lti_user, BearerTokenSchema, + course_service, ): - assignment_service.get_assignment.return_value.document_url = "DOCUMENT_URL" + assignment_service.get_assignment.return_value.document_url = ( + "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + ) + course_service.get_by_context_id.return_value.extra = { + "canvas": {"custom_canvas_course_id": "COURSE_ID"} + } BearerTokenSchema.return_value.authorization_param.return_value = "TOKEN" response = PagesAPIViews(pyramid_request).via_url() @@ -59,17 +67,139 @@ def test_via_url( pyramid_request, Any.url.with_path("/api/canvas/pages/proxy").with_query( { - "document_url": "DOCUMENT_URL", + "course_id": "COURSE_ID", + "page_id": "PAGE_ID", + "authorization": "TOKEN", + } + ), + ) + assert response == {"via_url": helpers.via_url.return_value} + + @pytest.mark.usefixtures("course_copy_plugin") + def test_via_url_copied_with_mapped_id( + self, + helpers, + pyramid_request, + assignment_service, + BearerTokenSchema, + course_service, + ): + assignment_service.get_assignment.return_value.document_url = ( + "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + ) + course_service.get_by_context_id.return_value.extra = { + "canvas": {"custom_canvas_course_id": "OTHER_COURSE_ID"} + } + course_service.get_by_context_id.return_value.get_mapped_page_id.return_value = ( + "OTHER_PAGE_ID" + ) + BearerTokenSchema.return_value.authorization_param.return_value = "TOKEN" + + response = PagesAPIViews(pyramid_request).via_url() + + helpers.via_url.assert_called_once_with( + pyramid_request, + Any.url.with_path("/api/canvas/pages/proxy").with_query( + { + "course_id": "OTHER_COURSE_ID", + "page_id": "OTHER_PAGE_ID", + "authorization": "TOKEN", + } + ), + ) + assert response == {"via_url": helpers.via_url.return_value} + + def test_via_url_copied_no_page_found( + self, + pyramid_request, + assignment_service, + course_service, + course_copy_plugin, + ): + assignment_service.get_assignment.return_value.document_url = ( + "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + ) + course_service.get_by_context_id.return_value.extra = { + "canvas": {"custom_canvas_course_id": "OTHER_COURSE_ID"} + } + course_service.get_by_context_id.return_value.get_mapped_page_id.return_value = ( + None + ) + course_copy_plugin.find_matching_page_in_course.return_value = None + + with pytest.raises(PageNotFoundInCourse): + PagesAPIViews(pyramid_request).via_url() + + course_copy_plugin.find_matching_page_in_course.assert_called_once_with( + "PAGE_ID", "OTHER_COURSE_ID" + ) + + def test_via_url_copied_found_page( + self, + pyramid_request, + assignment_service, + course_service, + course_copy_plugin, + BearerTokenSchema, + helpers, + ): + assignment_service.get_assignment.return_value.document_url = ( + "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + ) + course_service.get_by_context_id.return_value.extra = { + "canvas": {"custom_canvas_course_id": "OTHER_COURSE_ID"} + } + course_service.get_by_context_id.return_value.get_mapped_page_id.return_value = ( + None + ) + BearerTokenSchema.return_value.authorization_param.return_value = "TOKEN" + + course_copy_plugin.find_matching_page_in_course.return_value = Mock( + lms_id="OTHER_PAGE_ID" + ) + + response = PagesAPIViews(pyramid_request).via_url() + + course_copy_plugin.find_matching_page_in_course.assert_called_once_with( + "PAGE_ID", "OTHER_COURSE_ID" + ) + course_service.get_by_context_id.return_value.set_mapped_page_id( + "PAGE_ID", "OTHER_PAGE_ID" + ) + + helpers.via_url.assert_called_once_with( + pyramid_request, + Any.url.with_path("/api/canvas/pages/proxy").with_query( + { + "course_id": "OTHER_COURSE_ID", + "page_id": "OTHER_PAGE_ID", "authorization": "TOKEN", } ), ) assert response == {"via_url": helpers.via_url.return_value} + @pytest.mark.usefixtures("course_copy_plugin") + def test_via_url_copied_cant_access_page( + self, pyramid_request, assignment_service, canvas_service, course_service + ): + assignment_service.get_assignment.return_value.document_url = ( + "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + ) + course_service.get_by_context_id.return_value.extra = { + "canvas": {"custom_canvas_course_id": "OTHER_COURSE_ID"} + } + course_service.get_by_context_id.return_value.get_mapped_page_id.return_value = ( + "OTHER_PAGE_ID" + ) + canvas_service.api.pages.page.side_effect = CanvasAPIError + + with pytest.raises(PageNotFoundInCourse): + PagesAPIViews(pyramid_request).via_url() + def test_proxy(self, canvas_service, pyramid_request, application_instance): - pyramid_request.params[ - "document_url" - ] = "canvas://page/course/COURSE_ID/page_id/PAGE_ID" + pyramid_request.params["course_id"] = "COURSE_ID" + pyramid_request.params["page_id"] = "PAGE_ID" canvas_service.api.pages.page.return_value = CanvasPage( id=1, title=sentinel.title, updated_at="updated", body=sentinel.body )