opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #17411 from davelopez/purge_groups_and_roles_from_db

Browse Source 
Purge `groups` and `roles` from DB (for real)
This commit is contained in:
Marius van den Beek
2024-02-02 16:39:21 +01:00
committed by GitHub
parent beea9fa832 b9e40f8fa6
commit bbfc0425c4
4 changed files with 158 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/groups.py
+8 -5
View File
@@ -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)
lib/galaxy/managers/roles.py
+25 -19
View File
@@ -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
lib/galaxy_test/api/test_groups.py
+51 -2
View File
@@ -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:
lib/galaxy_test/api/test_roles.py
+74 -11
View File
@@ -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")