Merge pull request #14118 from jmchilton/modernize_exception_handling

Slightly modernized and standardized API error handling.

lib/galaxy/webapps/base/controller.py

@@ -199,10 +199,8 @@ def get_object(self, trans, id, class_name, check_ownership=False, check_accessi
                 deleted=deleted,
             )
 
-        except exceptions.ItemDeletionException as e:
-            raise HTTPBadRequest(detail=f"Invalid {class_name} id ( {str(id)} ) specified: {util.unicodify(e)}")
-        except exceptions.MessageException as e:
-            raise HTTPBadRequest(detail=e.err_msg)
+        except exceptions.MessageException:
+            raise
         except Exception as e:
             log.exception("Exception in get_object check for %s %s.", class_name, str(id))
             raise HTTPInternalServerError(comment=util.unicodify(e))

lib/galaxy/webapps/galaxy/api/users.py

@@ -25,6 +25,7 @@
 from galaxy.exceptions import ObjectInvalid
 from galaxy.managers import (
     api_keys,
+    base as managers_base,
     users,
 )
 from galaxy.managers.context import ProvidesUserContext
@@ -175,36 +176,51 @@ def index(self, trans: ProvidesUserContext, deleted="False", f_email=None, f_nam
         return rval
 
     @expose_api_anonymous
-    def show(self, trans: ProvidesUserContext, id, deleted="False", **kwd):
+    def show(self, trans: ProvidesUserContext, id, **kwd):
         """
         GET /api/users/{encoded_id}
         GET /api/users/deleted/{encoded_id}
         GET /api/users/current
         Displays information about a user.
         """
+        user = self._get_user_full(trans, id, **kwd)
+        if user is not None:
+            return self.user_serializer.serialize_to_view(user, view="detailed")
+        else:
+            return self.anon_user_api_value(trans)
+
+    def _get_user_full(self, trans, user_id, **kwd):
+        """Return referenced user or None if anonymous user is referenced."""
+        deleted = kwd.get("deleted", "False")
         deleted = util.string_as_bool(deleted)
         try:
             # user is requesting data about themselves
-            if id == "current":
+            if user_id == "current":
                 # ...and is anonymous - return usage and quota (if any)
                 if not trans.user:
-                    item = self.anon_user_api_value(trans)
-                    return item
+                    return None
 
                 # ...and is logged in - return full
                 else:
                     user = trans.user
             else:
-                user = self.get_user(trans, id, deleted=deleted)
+                return managers_base.get_object(
+                    trans,
+                    user_id,
+                    "User",
+                    deleted=deleted,
+                )
             # check that the user is requesting themselves (and they aren't del'd) unless admin
             if not trans.user_is_admin:
-                assert trans.user == user
-                assert not user.deleted
-        except exceptions.ItemDeletionException:
+                if trans.user != user or user.deleted:
+                    raise exceptions.InsufficientPermissionsException(
+                        "You are not allowed to perform action on that user", id=user_id
+                    )
+            return user
+        except exceptions.MessageException:
             raise
         except Exception:
-            raise exceptions.RequestParameterInvalidException("Invalid user id specified", id=id)
-        return self.user_serializer.serialize_to_view(user, view="detailed")
+            raise exceptions.RequestParameterInvalidException("Invalid user id specified", id=user_id)
 
     @expose_api
     def create(self, trans: GalaxyWebTransaction, payload: dict, **kwd):
@@ -253,14 +269,7 @@ def update(self, trans: ProvidesUserContext, id: str, payload: dict, **kwd):
             the serialized item after any changes
         """
         current_user = trans.user
-        user_to_update = self.user_manager.by_id(self.decode_id(id))
-
-        # only allow updating other users if they're admin
-        editing_someone_else = current_user != user_to_update
-        is_admin = self.user_manager.is_admin(current_user)
-        if editing_someone_else and not is_admin:
-            raise exceptions.InsufficientPermissionsException("You are not allowed to update that user", id=id)
-
+        user_to_update = self._get_user_full(trans, id, **kwd)
         self.user_deserializer.deserialize(user_to_update, payload, user=current_user, trans=trans)
         return self.user_serializer.serialize_to_view(user_to_update, view="detailed")
 

lib/galaxy_test/api/test_libraries.py

@@ -1,12 +1,10 @@
 import unittest
 
-import pytest
-import requests
-
 from galaxy.model.unittest_utils.store_fixtures import (
     one_ld_library_model_store_dict,
     TEST_LIBRARY_NAME,
 )
+from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import (
     DatasetCollectionPopulator,
     DatasetPopulator,
@@ -208,10 +206,9 @@ def test_show_private_dataset_permissions(self):
             "ForCreateDatasets", wait=True
         )
         with self._different_user():
-            with pytest.raises(requests.exceptions.HTTPError) as exc_info:
-                self.library_populator.show_ld(library["id"], library_dataset["id"])
-            # TODO: this should really be 403 and a proper JSON exception.
-            self._assert_status_code_is(exc_info.value.response, 400)
+            response = self.library_populator.show_ld_raw(library["id"], library_dataset["id"])
+            api_asserts.assert_status_code_is(response, 403)
+            api_asserts.assert_error_code_is(response, 403002)
 
     def test_create_dataset(self):
         library, library_dataset = self.library_populator.new_library_dataset_in_private_library(

lib/galaxy_test/api/test_users.py

@@ -60,7 +60,7 @@ def test_update(self):
 
             # not them
             update_response = self.__update(not_the_user, username=new_name)
-            self._assert_status_code_is(update_response, 403)
+            self._assert_status_code_is(update_response, 400)
 
             # non-existent
             no_user_id = "5d7db0757a2eb7ef"

lib/galaxy_test/base/populators.py

@@ -2279,8 +2279,12 @@ def raw_library_contents_create(self, library_id, payload, files=None):
         url_rel = f"libraries/{library_id}/contents"
         return self.galaxy_interactor.post(url_rel, payload, files=files)
 
-    def show_ld(self, library_id, library_dataset_id):
+    def show_ld_raw(self, library_id: str, library_dataset_id: str) -> Response:
         response = self.galaxy_interactor.get(f"libraries/{library_id}/contents/{library_dataset_id}")
+        return response
+
+    def show_ld(self, library_id: str, library_dataset_id: str) -> Dict[str, Any]:
+        response = self.show_ld_raw(library_id, library_dataset_id)
         response.raise_for_status()
         return response.json()