tmp

lib/galaxy/model/security.py

@@ -1458,77 +1458,91 @@ def get_showable_folders(
                 self.get_showable_folders(user, roles, folder, actions_to_check, showable_folders=showable_folders)
         return showable_folders
 
-    #    def set_user_group_and_role_associations(
+
+    #def set_user_group_and_role_associations(
     #        self, user_id: int, group_ids: Optional[List[int]] = None, role_ids: Optional[List[int]] = None
     #    ) -> None:
     #        """ Set user groups and user roles, replacing current associations."""
     #            self._set_user_groups(user_id, group_ids or [])
     #            self._set_user_roles(user_id, role_ids or [])
     #            self.sa_session.commit()
     #
+
+
+
+
     def set_group_user_and_role_associations(
-        # TODO set group type
         self,
-        group,
+        group: Group,
         *,
         user_ids: Optional[List[int]] = None,
         role_ids: Optional[List[int]] = None,
     ) -> None:
         """Set group users and group roles, replacing current associations."""
         self._ensure_model_instance_has_id(group)
-        self._set_group_users(group.id, user_ids or [])
-        self._set_group_roles(group.id, role_ids or [])
+        set_group_users(group.id, user_ids or [], self.sa_session)
+        set_group_roles(group.id, role_ids or [], self.sa_session)
 
-    #
-    #    def set_role_user_and_group_associations(
-    #        self, role_id: int, user_ids: Optional[List[int]] = None, group_ids: Optional[List[int]] = None
-    #    ) -> None:
-    #        """ Set role users and role groups, replacing current associations."""
-    #            self._set_group_users(role_id, user_ids or [])
-    #            self._set_group_roles(role_id, grour_ids or [])
-    #            self.sa_session.commit()
-    #
-    #    def _set_user_groups(self, user, groups):
-    #        delete_stmt = delete(UserGroupAssociation).where(UserGroupAssociation.user_id == user.id)
-    #        insert_values = [{"user_id": user.id, "group_id": group_id} for group_id in groups]
-    #        self._set_associations(UserGroupAssociation, delete_stmt, insert_values)
+        #self._set_group_users(group.id, user_ids or [])
+        #self._set_group_roles(group.id, role_ids or [])
+
+
+    def set_role_user_and_group_associations(
+        self,
+        role: Role,
+        *,
+        user_ids: Optional[List[int]] = None,
+        group_ids: Optional[List[int]] = None,
+    ) -> None:
+        """ Set role users and role groups, replacing current associations."""
+        self._ensure_model_instance_has_id(role)
+        self._set_role_users(role.id, user_ids or [])
+        self._set_role_groups(role.id, group_ids or [])
+
+    #def _set_user_groups(self, user, groups):
+    #    delete_stmt = delete(UserGroupAssociation).where(UserGroupAssociation.user_id == user.id)
+    #    insert_values = [{"user_id": user.id, "group_id": group_id} for group_id in groups]
+    #    self._set_associations(UserGroupAssociation, delete_stmt, insert_values)
+
+    #    def _set_user_roles(self, user, roles):
+    #        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.user_id == user.id)
+    #        insert_values = [{"user_id": user.id, "role_id": role_id} for role_id in roles]
+    #        self._set_associations(UserRoleAssociation, delete_stmt, insert_values)
     #
 
-    def _ensure_model_instance_has_id(self, model_instance):
-        # If model_instance is new, it may have not been assigned a database id yet, which is required
-        # for creating association records. Flush if that's the case.
-        if model_instance.id is None:
-            self.sa_session.flush([model_instance])
 
     def _set_group_users(self, group_id, users):
         delete_stmt = delete(UserGroupAssociation).where(UserGroupAssociation.group_id == group_id)
         insert_values = [{"group_id": group_id, "user_id": user_id} for user_id in users]
         self._set_associations(UserGroupAssociation, delete_stmt, insert_values)
 
-    #    def _set_user_roles(self, user, roles):
-    #        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.user_id == user.id)
-    #        insert_values = [{"user_id": user.id, "role_id": role_id} for role_id in roles]
-    #        self._set_associations(UserRoleAssociation, delete_stmt, insert_values)
-    #
-    #    def _set_role_users(self, role, users):
-    #        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.role_id == role.id)
-    #        insert_values = [{"role_id": role.id, "user_id": user_id} for user_id in users]
-    #        self._set_associations(UserRoleAssociation, delete_stmt, insert_values)
-    #
     def _set_group_roles(self, group_id, roles):
         delete_stmt = delete(GroupRoleAssociation).where(GroupRoleAssociation.group_id == group_id)
         insert_values = [{"group_id": group_id, "role_id": role_id} for role_id in roles]
         self._set_associations(GroupRoleAssociation, delete_stmt, insert_values)
 
-    #    def _set_role_groups(self, role, groups):
-    #        delete_stmt = delete(GroupRoleAssociation).where(GroupRoleAssociation.role_id == role.id)
-    #        insert_values = [{"role_id": role.id, "group_id": group_id} for group_id in groups]
-    #        self._set_associations(GroupRoleAssociation, delete_stmt, insert_values)
+    def _set_role_users(self, role_id, users):
+        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.role_id == role_id)
+        insert_values = [{"role_id": role_id, "user_id": user_id} for user_id in users]
+        self._set_associations(UserRoleAssociation, delete_stmt, insert_values)
+
+    def _set_role_groups(self, role_id, groups):
+        delete_stmt = delete(GroupRoleAssociation).where(GroupRoleAssociation.role_id == role_id)
+        insert_values = [{"role_id": role_id, "group_id": group_id} for group_id in groups]
+        self._set_associations(GroupRoleAssociation, delete_stmt, insert_values)
+
+    def _ensure_model_instance_has_id(self, model_instance):
+        # If model_instance is new, it may have not been assigned a database id yet, which is required
+        # for creating association records. Flush if that's the case.
+        if model_instance.id is None:
+            self.sa_session.flush([model_instance])
 
     def _set_associations(self, assoc_model, delete_stmt, insert_values):
-        # Ensure parent model has a database-assigned id
-        if assoc_model.id is None:
-            self.sa_session.flush(assoc_model)
+        # TODO remove this
+        ## Ensure parent model has a database-assigned id
+        #if assoc_model.id is None:
+        #    self.sa_session.flush(assoc_model)
+
         # Delete current associations
         self.sa_session.execute(delete_stmt)
         # Create new associations
@@ -1564,24 +1578,6 @@ def set_entity_user_associations(self, users=None, roles=None, groups=None, dele
             for group in groups:
                 self.associate_components(user=user, group=group)
 
-    def set_entity_role_associations(self, roles=None, users=None, groups=None, delete_existing_assocs=True):
-        users = users or []
-        roles = roles or []
-        groups = groups or []
-        for role in roles:
-            if delete_existing_assocs:
-                flush_needed = False
-                for a in role.users + role.groups:
-                    self.sa_session.delete(a)
-                    flush_needed = True
-                if flush_needed:
-                    with transaction(self.sa_session):
-                        self.sa_session.commit()
-            for user in users:
-                self.associate_components(user=user, role=role)
-            for group in groups:
-                self.associate_components(group=group, role=role)
-
     def get_component_associations(self, **kwd):
         assert len(kwd) == 2, "You must specify exactly 2 Galaxy security components to check for associations."
         if "dataset" in kwd:
@@ -1755,3 +1751,23 @@ def _walk_action_roles(permissions, query_action):
                 yield action, roles
         elif action == query_action.action and roles:
             yield action, roles
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

lib/galaxy/webapps/galaxy/controllers/admin.py

@@ -834,7 +834,13 @@ def manage_users_and_groups_for_role(self, trans, payload=None, **kwd):
                                 trans.sa_session.delete(dhp)
                     with transaction(trans.sa_session):
                         trans.sa_session.commit()
-            trans.app.security_agent.set_entity_role_associations(roles=[role], users=in_users, groups=in_groups)
+
+
+            trans.app.security_agent.set_role_user_and_group_associations(
+                role, user_ids=user_ids, group_ids=group_ids
+
+            with transaction(trans.sa_session):
+                trans.sa_session.commit()
             trans.sa_session.refresh(role)
             return {
                 "message": f"Role '{role.name}' has been updated with {len(in_users)} associated users and {len(in_groups)} associated groups."
@@ -923,7 +929,6 @@ def manage_users_and_roles_for_group(self, trans, payload=None, **kwd):
                 )
                 with transaction(trans.sa_session):
                     trans.sa_session.commit()
-
                 trans.sa_session.refresh(group)
                 return {
                     "message": f"Group '{group.name}' has been updated with {len(user_ids)} associated users and {len(role_ids)} associated roles."

test/unit/data/model/conftest.py

@@ -149,6 +149,16 @@ def f(**kwd):
     return f
 
 
+@pytest.fixture
+def make_group(session):
+    def f(**kwd):
+        model = m.Group(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
 @pytest.fixture
 def make_hda(session, make_history):
     def f(**kwd):

test/unit/data/model/db/test_security.py

@@ -0,0 +1,21 @@
+from galaxy.model.security import GalaxyRBACAgent
+
+
+def test_set_group_user_and_role_associations(make_user, make_role, make_group):
+    users = [make_user() for _ in range(3)]
+    roles = [make_role() for _ in range(3)]
+
+    user_ids = [users[0].id, users[1].id]  # first and second user
+    role_ids = [role.id for role in roles]   # all roles
+
+    group = make_group()
+    assert len(group.users) == 0
+    assert len(group.roles) == 0
+
+    db.group.set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+    assert len(group.users) == 2
+    assert len(group.roles) == 3
+    # also verify ids
+
+