Purge groups and roles from DB (for real)

lib/galaxy/managers/groups.py

@@ -110,19 +110,22 @@ def delete(self, trans: ProvidesAppContext, group_id: int):
             trans.sa_session.commit()
 
     def purge(self, trans: ProvidesAppContext, group_id: int):
-        group = self._get_group(trans.sa_session, group_id)
+        sa_session = trans.sa_session
+        group = self._get_group(sa_session, group_id)
         if not group.deleted:
             raise RequestParameterInvalidException(
                 f"Group '{group.name}' has not been deleted, so it cannot be purged."
             )
         # Delete UserGroupAssociations
         for uga in group.users:
-            trans.sa_session.delete(uga)
+            sa_session.delete(uga)
         # Delete GroupRoleAssociations
         for gra in group.roles:
-            trans.sa_session.delete(gra)
-        with transaction(trans.sa_session):
-            trans.sa_session.commit()
+            sa_session.delete(gra)
+        # Delete the group
+        sa_session.delete(group)
+        with transaction(sa_session):
+            sa_session.commit()
 
     def undelete(self, trans: ProvidesAppContext, group_id: int):
         group = self._get_group(trans.sa_session, group_id)

lib/galaxy/managers/roles.py

@@ -14,9 +14,14 @@
     Session,
 )
 
-import galaxy.exceptions
 from galaxy import model
-from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.exceptions import (
+    Conflict,
+    InconsistentDatabase,
+    InternalServerError,
+    ObjectNotFound,
+    RequestParameterInvalidException,
+)
 from galaxy.managers import base
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import Role
@@ -54,14 +59,14 @@ def get(self, trans: ProvidesUserContext, role_id: int) -> model.Role:
             stmt = select(self.model_class).where(self.model_class.id == role_id)
             role = self.session().execute(stmt).scalar_one()
         except sqlalchemy_exceptions.MultipleResultsFound:
-            raise galaxy.exceptions.InconsistentDatabase("Multiple roles found with the same id.")
+            raise InconsistentDatabase("Multiple roles found with the same id.")
         except sqlalchemy_exceptions.NoResultFound:
-            raise galaxy.exceptions.RequestParameterInvalidException("No accessible role found with the id provided.")
+            raise ObjectNotFound("No accessible role found with the id provided.")
         except Exception as e:
-            raise galaxy.exceptions.InternalServerError(f"Error loading from the database.{unicodify(e)}")
+            raise InternalServerError(f"Error loading from the database.{unicodify(e)}")
 
         if not (trans.user_is_admin or trans.app.security_agent.ok_to_display(trans.user, role)):
-            raise galaxy.exceptions.RequestParameterInvalidException("No accessible role found with the id provided.")
+            raise ObjectNotFound("No accessible role found with the id provided.")
 
         return role
 
@@ -81,7 +86,7 @@ def create_role(self, trans: ProvidesUserContext, role_definition_model: RoleDef
 
         stmt = select(Role).where(Role.name == name).limit(1)
         if trans.sa_session.scalars(stmt).first():
-            raise RequestParameterInvalidException(f"A role with that name already exists [{name}]")
+            raise Conflict(f"A role with that name already exists [{name}]")
 
         role_type = Role.types.ADMIN  # TODO: allow non-admins to create roles
 
@@ -117,36 +122,37 @@ def purge(self, trans: ProvidesUserContext, role: model.Role) -> model.Role:
         # - DefaultHistoryPermissions where role_id == Role.id
         # - GroupRoleAssociations where role_id == Role.id
         # - DatasetPermissionss where role_id == Role.id
+        sa_session = trans.sa_session
         if not role.deleted:
-            raise galaxy.exceptions.RequestParameterInvalidException(
-                f"Role '{role.name}' has not been deleted, so it cannot be purged."
-            )
+            raise RequestParameterInvalidException(f"Role '{role.name}' has not been deleted, so it cannot be purged.")
         # Delete UserRoleAssociations
         for ura in role.users:
-            user = trans.sa_session.query(trans.app.model.User).get(ura.user_id)
+            user = sa_session.query(trans.app.model.User).get(ura.user_id)
             # Delete DefaultUserPermissions for associated users
             for dup in user.default_permissions:
                 if role == dup.role:
-                    trans.sa_session.delete(dup)
+                    sa_session.delete(dup)
             # Delete DefaultHistoryPermissions for associated users
             for history in user.histories:
                 for dhp in history.default_permissions:
                     if role == dhp.role:
-                        trans.sa_session.delete(dhp)
-            trans.sa_session.delete(ura)
+                        sa_session.delete(dhp)
+            sa_session.delete(ura)
         # Delete GroupRoleAssociations
         for gra in role.groups:
-            trans.sa_session.delete(gra)
+            sa_session.delete(gra)
         # Delete DatasetPermissionss
         for dp in role.dataset_actions:
-            trans.sa_session.delete(dp)
-        with transaction(trans.sa_session):
-            trans.sa_session.commit()
+            sa_session.delete(dp)
+        # Delete the role
+        sa_session.delete(role)
+        with transaction(sa_session):
+            sa_session.commit()
         return role
 
     def undelete(self, trans: ProvidesUserContext, role: model.Role) -> model.Role:
         if not role.deleted:
-            raise galaxy.exceptions.RequestParameterInvalidException(
+            raise RequestParameterInvalidException(
                 f"Role '{role.name}' has not been deleted, so it cannot be undeleted."
             )
         role.deleted = False

lib/galaxy_test/api/test_groups.py

@@ -72,10 +72,10 @@ def test_show_unknown_raises_400(self):
         self._assert_status_code_is(response, 400)
 
     def test_update(self):
-        group = self.test_create_valid(group_name="group-test")
+        group = self.test_create_valid(group_name=f"group-test-{self.dataset_populator.get_random_name()}")
 
         group_id = group["id"]
-        updated_name = "group-test-updated"
+        updated_name = f"group-test-updated-{self.dataset_populator.get_random_name()}"
         update_payload = {
             "name": updated_name,
         }
@@ -101,6 +101,55 @@ def test_update_duplicating_name_raises_409(self):
         update_response = self._put(f"groups/{group_b_id}", data=update_payload, admin=True, json=True)
         self._assert_status_code_is(update_response, 409)
 
+    def test_delete(self):
+        group = self.test_create_valid()
+        group_id = group["id"]
+        delete_response = self._delete(f"groups/{group_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+
+    def test_delete_duplicating_name_raises_409(self):
+        group = self.test_create_valid()
+        group_id = group["id"]
+        group_name = group["name"]
+
+        delete_response = self._delete(f"groups/{group_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+
+        # Create a new group with the same name as the deleted one is not allowed
+        payload = self._build_valid_group_payload(group_name)
+        response = self._post("groups", payload, admin=True, json=True)
+        self._assert_status_code_is(response, 409)
+
+    def test_purge(self):
+        group = self.test_create_valid()
+        group_id = group["id"]
+
+        # Delete and purge the group
+        delete_response = self._delete(f"groups/{group_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+        purge_response = self._post(f"groups/{group_id}/purge", admin=True)
+        self._assert_status_code_is_ok(purge_response)
+
+        # The group is deleted and purged, so it cannot be found
+        response = self._get(f"groups/{group_id}", admin=True)
+        self._assert_status_code_is(response, 404)
+
+    def test_purge_can_reuse_name(self):
+        group = self.test_create_valid()
+        group_id = group["id"]
+        group_name = group["name"]
+
+        # Delete and purge the group
+        delete_response = self._delete(f"groups/{group_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+        purge_response = self._post(f"groups/{group_id}/purge", admin=True)
+        self._assert_status_code_is_ok(purge_response)
+
+        # Create a new group with the same name as the deleted one is allowed
+        payload = self._build_valid_group_payload(group_name)
+        response = self._post("groups", payload, admin=True, json=True)
+        self._assert_status_code_is(response, 200)
+
     def _assert_valid_group(self, group, assert_id=None):
         self._assert_has_keys(group, "id", "name", "model_class", "url")
         if assert_id is not None:

lib/galaxy_test/api/test_roles.py

@@ -103,15 +103,7 @@ def test_create_invalid_params(self):
     def test_create_valid(self):
         name = self.dataset_populator.get_random_name()
         description = "A test role."
-        payload = {
-            "name": name,
-            "description": description,
-            "user_ids": [self.dataset_populator.user_id()],
-        }
-        response = self._post("roles", payload, admin=True, json=True)
-        assert_status_code_is(response, 200)
-        role = response.json()
-        self.check_role_dict(role)
+        role = self._create_role(name=name, description=description)
 
         assert role["name"] == name
         assert role["description"] == description
@@ -133,11 +125,11 @@ def test_show_error_codes(self):
         response = self._get("roles/badroleid")
         assert_status_code_is(response, 400)
 
-        # Trying to access roles are errors - should probably be 403 not 400 though?
+        # Trying to access others roles raise (not found) error
         with self._different_user():
             different_user_role_id = self.dataset_populator.user_private_role_id()
         response = self._get(f"roles/{different_user_role_id}")
-        assert_status_code_is(response, 400)
+        assert_status_code_is(response, 404)
 
     @requires_admin
     def test_create_only_admin(self):
@@ -147,6 +139,77 @@ def test_create_only_admin(self):
         assert response_err["err_code"] == 403006
         assert "administrator" in response_err["err_msg"]
 
+    @requires_admin
+    def test_delete(self):
+        role = self._create_role()
+        role_id = role["id"]
+        response = self._delete(f"roles/{role_id}", admin=True)
+        assert_status_code_is(response, 200)
+
+    @requires_admin
+    def test_delete_duplicating_name_raises_409(self):
+        role = self._create_role()
+        role_id = role["id"]
+        role_name = role["name"]
+
+        delete_response = self._delete(f"roles/{role_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+
+        # Create a new role with the same name as the deleted one is not allowed
+        payload = self._build_valid_role_payload(role_name)
+        response = self._post("roles", payload, admin=True, json=True)
+        self._assert_status_code_is(response, 409)
+
+    @requires_admin
+    def test_purge(self):
+        role = self._create_role()
+        role_id = role["id"]
+
+        # Delete and purge the role
+        delete_response = self._delete(f"roles/{role_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+        purge_response = self._post(f"roles/{role_id}/purge", admin=True)
+        self._assert_status_code_is_ok(purge_response)
+
+        # The role is deleted and purged, so it cannot be found
+        response = self._get(f"roles/{role_id}", admin=True)
+        self._assert_status_code_is(response, 404)
+
+    @requires_admin
+    def test_purge_can_reuse_name(self):
+        role = self._create_role()
+        role_id = role["id"]
+        role_name = role["name"]
+
+        # Delete and purge the role
+        delete_response = self._delete(f"roles/{role_id}", admin=True)
+        self._assert_status_code_is_ok(delete_response)
+        purge_response = self._post(f"roles/{role_id}/purge", admin=True)
+        self._assert_status_code_is_ok(purge_response)
+
+        # Create a new role with the same name as the deleted one is allowed
+        payload = self._build_valid_role_payload(role_name)
+        response = self._post("roles", payload, admin=True, json=True)
+        self._assert_status_code_is(response, 200)
+
+    def _create_role(self, name: Optional[str] = None, description: Optional[str] = None) -> Dict[str, Any]:
+        payload = self._build_valid_role_payload(name=name, description=description)
+        response = self._post("roles", payload, admin=True, json=True)
+        assert_status_code_is(response, 200)
+        role = response.json()
+        self.check_role_dict(role)
+        return role
+
+    def _build_valid_role_payload(self, name: Optional[str] = None, description: Optional[str] = None):
+        name = name or self.dataset_populator.get_random_name()
+        description = description or f"A test role with name: {name}."
+        payload = {
+            "name": name,
+            "description": description,
+            "user_ids": [self.dataset_populator.user_id()],
+        }
+        return payload
+
     @staticmethod
     def check_role_dict(role_dict: Dict[str, Any], assert_id: Optional[str] = None) -> None:
         assert_has_keys(role_dict, "id", "name", "model_class", "url")