diff --git a/lib/galaxy/managers/groups.py b/lib/galaxy/managers/groups.py index 81f17469b71..33de69670cf 100644 --- a/lib/galaxy/managers/groups.py +++ b/lib/galaxy/managers/groups.py @@ -110,19 +110,22 @@ class GroupsManager: 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) diff --git a/lib/galaxy/managers/roles.py b/lib/galaxy/managers/roles.py index a24f6d456ff..89bf69815e2 100644 --- a/lib/galaxy/managers/roles.py +++ b/lib/galaxy/managers/roles.py @@ -14,9 +14,14 @@ from sqlalchemy.orm import ( 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 @@ class RoleManager(base.ModelManager[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 @@ class RoleManager(base.ModelManager[model.Role]): 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 @@ class RoleManager(base.ModelManager[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 diff --git a/lib/galaxy_test/api/test_groups.py b/lib/galaxy_test/api/test_groups.py index 6e561497a2c..e644392b615 100644 --- a/lib/galaxy_test/api/test_groups.py +++ b/lib/galaxy_test/api/test_groups.py @@ -72,10 +72,10 @@ class TestGroupsApi(ApiTestCase): 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 @@ class TestGroupsApi(ApiTestCase): 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: diff --git a/lib/galaxy_test/api/test_roles.py b/lib/galaxy_test/api/test_roles.py index 8e8fed4a64d..cee1805ed72 100644 --- a/lib/galaxy_test/api/test_roles.py +++ b/lib/galaxy_test/api/test_roles.py @@ -103,15 +103,7 @@ class TestRolesApi(ApiTestCase): 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 @@ class TestRolesApi(ApiTestCase): 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 @@ class TestRolesApi(ApiTestCase): 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")