Change Canvas Studio admin token refresh to work more like other refreshes

lms/routes.py

@@ -148,6 +148,10 @@ def includeme(config):  # pylint:disable=too-many-statements
     config.add_route(
         "canvas_studio_api.oauth.refresh", "/api/canvas_studio/oauth/refresh"
     )
+    config.add_route(
+        "canvas_studio_api.oauth.refresh_admin",
+        "/api/canvas_studio/oauth/refresh_admin",
+    )
     config.add_route("canvas_studio_api.media.list", "/api/canvas_studio/media")
     config.add_route(
         "canvas_studio_api.collections.media.list",

lms/services/canvas_studio.py

@@ -10,7 +10,11 @@
 from lms.models.oauth2_token import Service
 from lms.models.user import User
 from lms.services.aes import AESService
-from lms.services.exceptions import ExternalRequestError, OAuth2TokenError
+from lms.services.exceptions import (
+    ExternalRequestError,
+    OAuth2TokenError,
+    SerializableError,
+)
 from lms.services.oauth_http import OAuthHTTPService
 from lms.services.oauth_http import factory as oauth_http_factory
 from lms.validation._base import RequestsResponseSchema
@@ -150,6 +154,21 @@ def refresh_access_token(self):
             auth=(self._client_id, self._client_secret),
         )
 
+    def refresh_admin_access_token(self):
+        """Refresh the existing admin access token for Canvas Studio API calls."""
+
+        try:
+            self._admin_oauth_http.refresh_access_token(
+                self._token_url(),
+                self.redirect_uri(),
+                auth=(self._client_id, self._client_secret),
+            )
+        except ExternalRequestError as refresh_err:
+            raise SerializableError(
+                error_code="canvas_studio_admin_token_refresh_failed",
+                message="Canvas Studio admin token refresh failed.",
+            ) from refresh_err
+
     def authorization_url(self, state: str) -> str:
         """
         Construct the authorization endpoint URL for Canvas Studio.
@@ -341,9 +360,7 @@ def _api_request(
         response = self._bare_api_request(path, as_admin=as_admin)
         return schema_cls(response).parse()
 
-    def _admin_api_request(
-        self, path: str, allow_redirects=True, allow_refresh=True
-    ) -> requests.Response:
+    def _admin_api_request(self, path: str, allow_redirects=True) -> requests.Response:
         """
         Make a request to the Canvas Studio API using the admin user identity.
 
@@ -374,36 +391,11 @@ def _admin_api_request(
                     ) from err
                 raise
 
-            # If we already performed a refresh in response to the original
-            # request, `allow_refresh` will be False and we abort.
-            if not allow_refresh:
-                raise
-
-            # For admin requests, we have to do the refresh here, because the
-            # refresh path involving the frontend is designed for refreshing
-            # the current LTI user's access token.
-            #
-            # Ideally it would be simpler if we could just encapsulate the
-            # entire refresh process in OAuthHTTPService.
-            try:
-                self._admin_oauth_http.refresh_access_token(
-                    self._token_url(),
-                    self.redirect_uri(),
-                    auth=(self._client_id, self._client_secret),
-                )
-            except ExternalRequestError as refresh_err:
-                raise ExternalRequestError(
-                    message="Canvas Studio admin token refresh failed. Ask the admin user to re-authenticate."
-                ) from refresh_err
-
-            # Retry the request with the new token.
-            return self._admin_api_request(
-                path,
-                allow_redirects=allow_redirects,
-                # If the request fails again, make sure we don't repeat the
-                # refresh to avoid getting stuck in a loop.
-                allow_refresh=False,
-            )
+            raise OAuth2TokenError(
+                refreshable=True,
+                refresh_route="canvas_studio_api.oauth.refresh_admin",
+                refresh_service=Service.CANVAS_STUDIO,
+            ) from err
 
     def _bare_api_request(
         self, path: str, as_admin=False, allow_redirects=True

lms/static/scripts/frontend_apps/components/LaunchErrorDialog.tsx

@@ -356,6 +356,21 @@ export default function LaunchErrorDialog({
           </p>
         </ErrorModal>
       );
+
+    case 'canvas_studio_admin_token_refresh_failed':
+      return (
+        <ErrorModal
+          {...defaultProps}
+          onRetry={undefined}
+          title="Unable to access Canvas Studio video"
+        >
+          <p>
+            Your Canvas LMS administrator needs to re-authorize the integration
+            between Hypothesis and Canvas Studio.
+          </p>
+        </ErrorModal>
+      );
+
     case 'blackboard_group_set_not_found':
       return (
         <ErrorModal

lms/static/scripts/frontend_apps/components/test/LaunchErrorDialog-test.js

@@ -105,6 +105,14 @@ describe('LaunchErrorDialog', () => {
       hasRetry: false,
       withError: true,
     },
+    {
+      errorState: 'canvas_studio_admin_token_refresh_failed',
+      expectedText:
+        'Your Canvas LMS administrator needs to re-authorize the integration between Hypothesis and Canvas Studio',
+      expectedTitle: 'Unable to access Canvas Studio video',
+      hasRetry: false,
+      withError: true,
+    },
     {
       errorState: 'd2l_file_not_found_in_course_instructor',
       expectedText:

lms/static/scripts/frontend_apps/errors.ts

@@ -22,6 +22,7 @@ export type LTILaunchServerErrorCode =
   | 'canvas_studio_download_unavailable'
   | 'canvas_studio_transcript_unavailable'
   | 'canvas_studio_media_not_found'
+  | 'canvas_studio_admin_token_refresh_failed'
   | 'd2l_file_not_found_in_course_instructor'
   | 'd2l_file_not_found_in_course_student'
   | 'd2l_group_set_empty'
@@ -169,6 +170,7 @@ export function isLTILaunchServerError(error: ErrorLike): error is APIError {
       'canvas_studio_download_unavailable',
       'canvas_studio_transcript_unavailable',
       'canvas_studio_media_not_found',
+      'canvas_studio_admin_token_refresh_failed',
       'vitalsource_user_not_found',
       'vitalsource_no_book_license',
       'moodle_page_not_found_in_course',

lms/views/api/refresh.py

@@ -25,6 +25,10 @@ def get_refreshed_token_from_canvas(self):
     def get_refreshed_token_from_canvas_studio(self):
         self.request.find_service(CanvasStudioService).refresh_access_token()
 
+    @view_config(route_name="canvas_studio_api.oauth.refresh_admin")
+    def get_refreshed_admin_token_from_canvas_studio(self):
+        self.request.find_service(CanvasStudioService).refresh_admin_access_token()
+
     @view_config(route_name="blackboard_api.oauth.refresh")
     def get_refreshed_token_from_blackboard(self):
         blackboard_api_client = self.request.find_service(name="blackboard_api_client")

tests/unit/lms/services/canvas_studio_test.py

@@ -8,7 +8,11 @@
 
 from lms.models.oauth2_token import Service
 from lms.services.canvas_studio import CanvasStudioService, factory
-from lms.services.exceptions import ExternalRequestError, OAuth2TokenError
+from lms.services.exceptions import (
+    ExternalRequestError,
+    OAuth2TokenError,
+    SerializableError,
+)
 from lms.services.oauth_http import OAuthHTTPService
 from tests import factories
 
@@ -36,6 +40,29 @@ def test_refresh_access_token(self, svc, oauth_http_service, client_secret):
             auth=("the_client_id", client_secret),
         )
 
+    def test_refresh_admin_access_token(
+        self, svc, admin_oauth_http_service, client_secret
+    ):
+        svc.refresh_admin_access_token()
+
+        admin_oauth_http_service.refresh_access_token.assert_called_with(
+            "https://hypothesis.instructuremedia.com/api/public/oauth/token",
+            "http://example.com/api/canvas_studio/oauth/callback",
+            auth=("the_client_id", client_secret),
+        )
+
+    def test_refresh_admin_access_token_error(self, svc, admin_oauth_http_service):
+        response = factories.requests.Response(status_code=403)
+        admin_oauth_http_service.refresh_access_token.side_effect = (
+            ExternalRequestError(response=response)
+        )
+
+        with pytest.raises(SerializableError) as exc_info:
+            svc.refresh_admin_access_token()
+
+        assert exc_info.value.error_code == "canvas_studio_admin_token_refresh_failed"
+        assert exc_info.value.message == "Canvas Studio admin token refresh failed."
+
     def test_authorization_url(self, svc):
         state = "the_callback_state"
         auth_url = svc.authorization_url(state)
@@ -237,60 +264,20 @@ def test_get_video_download_url_fails_if_admin_not_authenticated(
             == "The Canvas Studio admin needs to authenticate the Hypothesis integration"
         )
 
-    def test_admin_token_refreshed_if_needed(
-        self, admin_oauth_http_service, svc, client_secret
+    def test_get_video_download_url_when_admin_token_expired(
+        self, admin_oauth_http_service, svc
     ):
-        # Set up admin-authenticated OAuth request to fail due to expired token.
-        token_expired_response = factories.requests.Response(status_code=401)
-        original_get = admin_oauth_http_service.get.side_effect
-        admin_oauth_http_service.get.side_effect = ExternalRequestError(
-            response=token_expired_response
-        )
-
-        def refresh_ok(*_args, **_kwargs):
-            admin_oauth_http_service.get.side_effect = original_get
-
-        def refresh_fail(*_args, **_kwargs):
-            raise ExternalRequestError(message="refresh failed")
-
-        admin_oauth_http_service.refresh_access_token.side_effect = refresh_ok
-
-        # Perform a request that is admin-authenticated. This should trigger
-        # a refresh and then succeed as normal.
-        url = svc.get_video_download_url("42")
-
-        admin_oauth_http_service.refresh_access_token.assert_called_with(
-            "https://hypothesis.instructuremedia.com/api/public/oauth/token",
-            "http://example.com/api/canvas_studio/oauth/callback",
-            auth=("the_client_id", client_secret),
-        )
-        assert url == "https://videos.cdn.com/video.mp4?signature=abc"
-
-        # Set up the initial request to fail again, due to an expired token,
-        # but this time make the refresh fail.
+        response = factories.requests.Response(status_code=401)
         admin_oauth_http_service.get.side_effect = ExternalRequestError(
-            response=token_expired_response
+            response=response
         )
-        admin_oauth_http_service.refresh_access_token.side_effect = refresh_fail
 
-        with pytest.raises(ExternalRequestError) as exc_info:
+        with pytest.raises(OAuth2TokenError) as exc_info:
             svc.get_video_download_url("42")
 
-        assert (
-            exc_info.value.message
-            == "Canvas Studio admin token refresh failed. Ask the admin user to re-authenticate."
-        )
-
-        # Set up the initial request to fail again, due to an expired token,
-        # but this time make subsequent requests fail even though the refresh
-        # apparently succeeded.
-        admin_oauth_http_service.get.side_effect = ExternalRequestError(
-            response=token_expired_response
-        )
-        admin_oauth_http_service.refresh_access_token.side_effect = None
-
-        with pytest.raises(Exception) as exc_info:
-            svc.get_video_download_url("42")
+        assert exc_info.value.refreshable is True
+        assert exc_info.value.refresh_route == "canvas_studio_api.oauth.refresh_admin"
+        assert exc_info.value.refresh_service == Service.CANVAS_STUDIO
 
     def test_get_transcript_url_returns_url_if_published(self, svc):
         transcript_url = svc.get_transcript_url("42")

tests/unit/lms/views/api/refresh_test.py

@@ -25,6 +25,13 @@ def test_get_refreshed_token_from_canvas_studio(self, canvas_studio_service, vie
 
         canvas_studio_service.refresh_access_token.assert_called_once_with()
 
+    def test_get_refreshed_admin_token_from_canvas_studio(
+        self, canvas_studio_service, views
+    ):
+        views.get_refreshed_admin_token_from_canvas_studio()
+
+        canvas_studio_service.refresh_admin_access_token.assert_called_once_with()
+
     def test_get_refreshed_token_from_d2l(self, d2l_api_client, views):
         views.get_refreshed_token_from_d2l()