Skip to content

Commit

Permalink
Support course copy for Canvas pages
Browse files Browse the repository at this point in the history
  • Loading branch information
marcospri committed Oct 20, 2023
1 parent cc7a4c8 commit 3ac313c
Show file tree
Hide file tree
Showing 2 changed files with 224 additions and 15 deletions.
95 changes: 87 additions & 8 deletions lms/views/api/canvas/pages.py
Original file line number Diff line number Diff line change
@@ -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<course_id>[^\/]*)\/page_id\/(?P<page_id>[^\/]*)"
)


class PageNotFoundInCourse(FileNotFoundInCourse):
...


@view_defaults(permission=Permissions.API, renderer="json")
class PagesAPIViews:
def __init__(self, request):
Expand Down Expand Up @@ -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.
Expand All @@ -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,
},
),
Expand All @@ -74,15 +144,24 @@ 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
),
"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
144 changes: 137 additions & 7 deletions tests/unit/lms/views/api/canvas/pages_test.py
Original file line number Diff line number Diff line change
@@ -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(
Expand Down Expand Up @@ -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,
Expand All @@ -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()
Expand All @@ -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
)
Expand Down

0 comments on commit 3ac313c

Please sign in to comment.