diff --git a/lib/galaxy/managers/groups.py b/lib/galaxy/managers/groups.py new file mode 100644 index 00000000000..8162bf3ebfe --- /dev/null +++ b/lib/galaxy/managers/groups.py @@ -0,0 +1,116 @@ +from typing import ( + Any, + Dict, + List, +) + +from sqlalchemy import false + +from galaxy import model +from galaxy.app import StructuredApp +from galaxy.exceptions import ( + Conflict, + ObjectAttributeMissingException, + ObjectNotFound, +) +from galaxy.managers.base import decode_id +from galaxy.managers.context import ProvidesAppContext +from galaxy.schema.fields import EncodedDatabaseIdField +from galaxy.web import url_for + + +class GroupsManager: + """Interface/service object shared by controllers for interacting with groups.""" + + def __init__(self, app: StructuredApp) -> None: + self._app = app + + def index(self, trans: ProvidesAppContext): + """ + Displays a collection (list) of groups. + """ + rval = [] + for group in trans.sa_session.query(model.Group).filter(model.Group.table.c.deleted == false()): + item = group.to_dict(value_mapper={'id': trans.security.encode_id}) + encoded_id = trans.security.encode_id(group.id) + item['url'] = url_for('group', id=encoded_id) + rval.append(item) + return rval + + def create(self, trans: ProvidesAppContext, payload: Dict[str, Any]): + """ + Creates a new group. + """ + name = payload.get('name', None) + if name is None: + raise ObjectAttributeMissingException("Missing required name") + self._check_duplicated_group_name(trans, name) + + group = model.Group(name=name) + trans.sa_session.add(group) + encoded_user_ids = payload.get('user_ids', []) + users = self._get_users_by_encoded_ids(trans, encoded_user_ids) + encoded_role_ids = payload.get('role_ids', []) + roles = self._get_roles_by_encoded_ids(trans, encoded_role_ids) + trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users) + trans.sa_session.flush() + + encoded_id = trans.security.encode_id(group.id) + item = group.to_dict(view='element', value_mapper={'id': trans.security.encode_id}) + item['url'] = url_for('group', id=encoded_id) + return [item] + + def show(self, trans: ProvidesAppContext, encoded_id: EncodedDatabaseIdField): + """ + Displays information about a group. + """ + group = self._get_group(trans, encoded_id) + item = group.to_dict(view='element', value_mapper={'id': trans.security.encode_id}) + item['url'] = url_for('group', id=encoded_id) + item['users_url'] = url_for('group_users', group_id=encoded_id) + item['roles_url'] = url_for('group_roles', group_id=encoded_id) + return item + + def update(self, trans: ProvidesAppContext, encoded_id: EncodedDatabaseIdField, payload: Dict[str, Any]): + """ + Modifies a group. + """ + group = self._get_group(trans, encoded_id) + name = payload.get('name', None) + if name: + self._check_duplicated_group_name(trans, name) + group.name = name + trans.sa_session.add(group) + encoded_user_ids = payload.get('user_ids', []) + users = self._get_users_by_encoded_ids(trans, encoded_user_ids) + encoded_role_ids = payload.get('role_ids', []) + roles = self._get_roles_by_encoded_ids(trans, encoded_role_ids) + trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users, delete_existing_assocs=False) + trans.sa_session.flush() + + def _decode_id(self, encoded_id: EncodedDatabaseIdField) -> int: + return decode_id(self._app, encoded_id) + + def _decode_ids(self, encoded_ids: List[EncodedDatabaseIdField]) -> List[int]: + return [self._decode_id(encoded_id) for encoded_id in encoded_ids] + + def _check_duplicated_group_name(self, trans: ProvidesAppContext, group_name: str) -> None: + if trans.sa_session.query(model.Group).filter(model.Group.table.c.name == group_name).first(): + raise Conflict(f"A group with name '{group_name}' already exists") + + def _get_group(self, trans: ProvidesAppContext, encoded_id: EncodedDatabaseIdField) -> model.Group: + decoded_group_id = self._decode_id(encoded_id) + group = trans.sa_session.query(model.Group).get(decoded_group_id) + if group is None: + raise ObjectNotFound(f"Group with id {encoded_id} was not found.") + return group + + def _get_users_by_encoded_ids(self, trans: ProvidesAppContext, encoded_user_ids: List[EncodedDatabaseIdField]) -> List[model.User]: + decoded_user_ids = self._decode_ids(encoded_user_ids) + users = trans.sa_session.query(model.User).filter(model.User.table.c.id.in_(decoded_user_ids)).all() + return users + + def _get_roles_by_encoded_ids(self, trans: ProvidesAppContext, encoded_role_ids: List[EncodedDatabaseIdField]) -> List[model.Role]: + decoded_role_ids = self._decode_ids(encoded_role_ids) + roles = trans.sa_session.query(model.Role).filter(model.Role.table.c.id.in_(decoded_role_ids)).all() + return roles diff --git a/lib/galaxy/webapps/galaxy/api/groups.py b/lib/galaxy/webapps/galaxy/api/groups.py index a4ff3c90dd5..9433dab6ff6 100644 --- a/lib/galaxy/webapps/galaxy/api/groups.py +++ b/lib/galaxy/webapps/galaxy/api/groups.py @@ -2,128 +2,62 @@ API operations on Group objects. """ import logging +from typing import ( + Any, + Dict, +) -from sqlalchemy import false - -from galaxy import web -from galaxy.webapps.base.controller import BaseAPIController, url_for +from galaxy.app import StructuredApp +from galaxy.managers.context import ProvidesAppContext +from galaxy.managers.groups import GroupsManager +from galaxy.schema.fields import EncodedDatabaseIdField +from galaxy.web import ( + expose_api, + require_admin, +) +from galaxy.webapps.base.controller import BaseAPIController log = logging.getLogger(__name__) class GroupAPIController(BaseAPIController): - @web.require_admin - @web.legacy_expose_api - def index(self, trans, **kwd): + def __init__(self, app: StructuredApp): + super().__init__(app) + self.manager = GroupsManager(app) + + @expose_api + @require_admin + def index(self, trans: ProvidesAppContext, **kwd): """ GET /api/groups Displays a collection (list) of groups. """ - rval = [] - for group in trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.table.c.deleted == false()): - if trans.user_is_admin: - item = group.to_dict(value_mapper={'id': trans.security.encode_id}) - encoded_id = trans.security.encode_id(group.id) - item['url'] = url_for('group', id=encoded_id) - rval.append(item) - return rval + return self.manager.index(trans) - @web.legacy_expose_api - def create(self, trans, payload, **kwd): + @expose_api + @require_admin + def create(self, trans: ProvidesAppContext, payload: Dict[str, Any], **kwd): """ POST /api/groups Creates a new group. """ - log.info("groups payload%s\n" % (payload)) - if not trans.user_is_admin: - trans.response.status = 403 - return "You are not authorized to create a new group." - name = payload.get('name', None) - if not name: - trans.response.status = 400 - return "Enter a valid name" - if trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.table.c.name == name).first(): - trans.response.status = 400 - return "A group with that name already exists" + return self.manager.create(trans, payload) - group = trans.app.model.Group(name=name) - trans.sa_session.add(group) - user_ids = payload.get('user_ids', []) - for i in user_ids: - log.info("user_id: %s\n" % (i)) - log.info("{} {}\n".format(i, trans.security.decode_id(i))) - users = [trans.sa_session.query(trans.model.User).get(trans.security.decode_id(i)) for i in user_ids] - role_ids = payload.get('role_ids', []) - roles = [trans.sa_session.query(trans.model.Role).get(trans.security.decode_id(i)) for i in role_ids] - trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users) - """ - # Create the UserGroupAssociations - for user in users: - trans.app.security_agent.associate_user_group( user, group ) - # Create the GroupRoleAssociations - for role in roles: - trans.app.security_agent.associate_group_role( group, role ) - """ - trans.sa_session.flush() - encoded_id = trans.security.encode_id(group.id) - item = group.to_dict(view='element', value_mapper={'id': trans.security.encode_id}) - item['url'] = url_for('group', id=encoded_id) - return [item] - - @web.require_admin - @web.legacy_expose_api - def show(self, trans, id, **kwd): + @expose_api + @require_admin + def show(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, **kwd): """ GET /api/groups/{encoded_group_id} Displays information about a group. """ - group_id = id - try: - decoded_group_id = trans.security.decode_id(group_id) - except TypeError: - trans.response.status = 400 - return "Malformed group id ( %s ) specified, unable to decode." % str(group_id) - try: - group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id) - except Exception: - group = None - if not group: - trans.response.status = 400 - return "Invalid group id ( %s ) specified." % str(group_id) - item = group.to_dict(view='element', value_mapper={'id': trans.security.encode_id}) - item['url'] = url_for('group', id=group_id) - item['users_url'] = url_for('group_users', group_id=group_id) - item['roles_url'] = url_for('group_roles', group_id=group_id) - return item + return self.manager.show(trans, id) - @web.require_admin - @web.legacy_expose_api - def update(self, trans, id, payload, **kwd): + @expose_api + @require_admin + def update(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, payload: Dict[str, Any], **kwd): """ PUT /api/groups/{encoded_group_id} Modifies a group. """ - group_id = id - try: - decoded_group_id = trans.security.decode_id(group_id) - except TypeError: - trans.response.status = 400 - return "Malformed group id ( %s ) specified, unable to decode." % str(group_id) - try: - group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id) - except Exception: - group = None - if not group: - trans.response.status = 400 - return "Invalid group id ( %s ) specified." % str(group_id) - name = payload.get('name', None) - if name: - group.name = name - trans.sa_session.add(group) - user_ids = payload.get('user_ids', []) - users = [trans.sa_session.query(trans.model.User).get(trans.security.decode_id(i)) for i in user_ids] - role_ids = payload.get('role_ids', []) - roles = [trans.sa_session.query(trans.model.Role).get(trans.security.decode_id(i)) for i in role_ids] - trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users, delete_existing_assocs=False) - trans.sa_session.flush() + self.manager.update(trans, id, payload) diff --git a/lib/galaxy_test/api/test_groups.py b/lib/galaxy_test/api/test_groups.py new file mode 100644 index 00000000000..66606be3dc6 --- /dev/null +++ b/lib/galaxy_test/api/test_groups.py @@ -0,0 +1,117 @@ +import json + +from galaxy_test.base.populators import DatasetPopulator +from ._framework import ApiTestCase + + +class GroupsApiTestCase(ApiTestCase): + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def test_create_valid(self, group_name: str = None): + payload = self._build_valid_group_payload(group_name) + response = self._post("groups", payload, admin=True, json=True) + self._assert_status_code_is(response, 200) + group = response.json()[0] # POST /api/groups returns a list + self._assert_valid_group(group) + return group + + def test_create_only_admin(self): + response = self._post("groups", json=True) + self._assert_status_code_is(response, 403) + + def test_create_invalid_params_raises_400(self): + payload = self._build_valid_group_payload() + payload["name"] = None + response = self._post("groups", payload, admin=True, json=True) + self._assert_status_code_is(response, 400) + + def test_create_duplicated_name_raises_409(self): + payload = self._build_valid_group_payload() + response = self._post("groups", payload, admin=True, json=True) + self._assert_status_code_is(response, 200) + + response = self._post("groups", payload, admin=True, json=True) + self._assert_status_code_is(response, 409) + + def test_index(self): + self.test_create_valid() + response = self._get("groups", admin=True) + self._assert_status_code_is(response, 200) + groups = response.json() + assert isinstance(groups, list) + assert len(groups) > 0 + for group in groups: + self._assert_valid_group(group) + + def test_index_only_admin(self): + response = self._get("groups") + self._assert_status_code_is(response, 403) + + def test_show(self): + group = self.test_create_valid() + group_id = group["id"] + response = self._get(f"groups/{group_id}", admin=True) + self._assert_status_code_is(response, 200) + response_group = response.json() + self._assert_valid_group(response_group) + self._assert_has_keys(response_group, "users_url", "roles_url") + + def test_show_only_admin(self): + group = self.test_create_valid() + group_id = group["id"] + response = self._get(f"groups/{group_id}") + self._assert_status_code_is(response, 403) + + def test_show_unknown_raises_400(self): + group_id = "invalid-group-id" + response = self._get(f"groups/{group_id}", admin=True) + self._assert_status_code_is(response, 400) + + def test_update(self): + group = self.test_create_valid(group_name="group-test") + + group_id = group["id"] + updated_name = "group-test-updated" + update_payload = json.dumps({ + "name": updated_name, + }) + update_response = self._put(f"groups/{group_id}", data=update_payload, admin=True) + self._assert_status_code_is_ok(update_response) + + def test_update_only_admin(self): + group = self.test_create_valid() + group_id = group["id"] + response = self._put(f"groups/{group_id}") + self._assert_status_code_is(response, 403) + + def test_update_duplicating_name_raises_409(self): + group_a = self.test_create_valid() + group_b = self.test_create_valid() + + # Update group_b with the same name as group_a + group_b_id = group_b["id"] + updated_name = group_a["name"] + update_payload = json.dumps({ + "name": updated_name, + }) + update_response = self._put(f"groups/{group_b_id}", data=update_payload, admin=True) + self._assert_status_code_is(update_response, 409) + + 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: + assert group["id"] == assert_id + + def _build_valid_group_payload(self, name: str = None): + name = name or self.dataset_populator.get_random_name() + user_id = self.dataset_populator.user_id() + role_id = self.dataset_populator.user_private_role_id() + payload = { + "name": name, + "user_ids": [user_id], + "role_ids": [role_id], + } + return payload