Merge pull request #18777 from jdavcs/dev_assoc_handling2

Backend handling of setting user-role, user-group, and group-role associations

lib/galaxy/app.py

@@ -655,7 +655,7 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication, Inst
         # Load security policy.
         self.security_agent = self.model.security_agent
         self.host_security_agent = galaxy.model.security.HostAgent(
-            model=self.security_agent.model, permitted_actions=self.security_agent.permitted_actions
+            self.security_agent.sa_session, permitted_actions=self.security_agent.permitted_actions
         )
 
         # We need the datatype registry for running certain tasks that modify HDAs, and to build the registry we need

lib/galaxy/managers/groups.py

@@ -13,8 +13,6 @@ from galaxy.exceptions import (
 from galaxy.managers.context import ProvidesAppContext
 from galaxy.model import Group
 from galaxy.model.base import transaction
-from galaxy.model.db.role import get_roles_by_ids
-from galaxy.model.db.user import get_users_by_ids
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.fields import Security
 from galaxy.schema.groups import (
@@ -54,13 +52,11 @@ class GroupsManager:
 
         group = model.Group(name=name)
         sa_session.add(group)
-        user_ids = payload.user_ids
-        users = get_users_by_ids(sa_session, user_ids)
-        role_ids = payload.role_ids
-        roles = get_roles_by_ids(sa_session, role_ids)
-        trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users)
-        with transaction(sa_session):
-            sa_session.commit()
+
+        trans.app.security_agent.set_group_user_and_role_associations(
+            group, user_ids=payload.user_ids, role_ids=payload.role_ids
+        )
+        sa_session.commit()
 
         encoded_id = Security.security.encode_id(group.id)
         item = group.to_dict(view="element")
@@ -88,23 +84,12 @@ class GroupsManager:
         if name := payload.name:
             self._check_duplicated_group_name(sa_session, name)
             group.name = name
-            sa_session.add(group)
-
-        users = None
-        if payload.user_ids is not None:
-            users = get_users_by_ids(sa_session, payload.user_ids)
-
-        roles = None
-        if payload.role_ids is not None:
-            roles = get_roles_by_ids(sa_session, payload.role_ids)
-
-        self._app.security_agent.set_entity_group_associations(
-            groups=[group], roles=roles, users=users, delete_existing_assocs=False
-        )
-
-        with transaction(sa_session):
             sa_session.commit()
 
+        self._app.security_agent.set_group_user_and_role_associations(
+            group, user_ids=payload.user_ids, role_ids=payload.role_ids
+        )
+
         encoded_id = Security.security.encode_id(group.id)
         item = group.to_dict(view="element")
         item["url"] = self._url_for(trans, "show_group", group_id=encoded_id)

lib/galaxy/model/__init__.py

@@ -849,14 +849,6 @@ class User(Base, Dictifiable, RepresentById):
     all_notifications: Mapped[List["UserNotificationAssociation"]] = relationship(
         back_populates="user", cascade_backrefs=False
     )
-    non_private_roles: Mapped[List["UserRoleAssociation"]] = relationship(
-        viewonly=True,
-        primaryjoin=(
-            lambda: (User.id == UserRoleAssociation.user_id)
-            & (UserRoleAssociation.role_id == Role.id)
-            & not_(Role.name == User.email)
-        ),
-    )
 
     preferences: AssociationProxy[Any]
 
@@ -2967,10 +2959,11 @@ class Group(Base, Dictifiable, RepresentById):
 
 class UserGroupAssociation(Base, RepresentById):
     __tablename__ = "user_group_association"
+    __table_args__ = (UniqueConstraint("user_id", "group_id"),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id"), index=True, nullable=True)
-    group_id: Mapped[int] = mapped_column(ForeignKey("galaxy_group.id"), index=True, nullable=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id"), index=True)
+    group_id: Mapped[int] = mapped_column(ForeignKey("galaxy_group.id"), index=True)
     create_time: Mapped[datetime] = mapped_column(default=now, nullable=True)
     update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True)
     user: Mapped["User"] = relationship(back_populates="groups")
@@ -3685,10 +3678,11 @@ class HistoryUserShareAssociation(Base, UserShareAssociation):
 
 class UserRoleAssociation(Base, RepresentById):
     __tablename__ = "user_role_association"
+    __table_args__ = (UniqueConstraint("user_id", "role_id"),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id"), index=True, nullable=True)
-    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), index=True, nullable=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id"), index=True)
+    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), index=True)
     create_time: Mapped[datetime] = mapped_column(default=now, nullable=True)
     update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True)
 
@@ -3703,10 +3697,11 @@ class UserRoleAssociation(Base, RepresentById):
 
 class GroupRoleAssociation(Base, RepresentById):
     __tablename__ = "group_role_association"
+    __table_args__ = (UniqueConstraint("group_id", "role_id"),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    group_id: Mapped[int] = mapped_column(ForeignKey("galaxy_group.id"), index=True, nullable=True)
-    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), index=True, nullable=True)
+    group_id: Mapped[int] = mapped_column(ForeignKey("galaxy_group.id"), index=True)
+    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), index=True)
     create_time: Mapped[datetime] = mapped_column(default=now, nullable=True)
     update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True)
     group: Mapped["Group"] = relationship(back_populates="roles")

lib/galaxy/model/mapping.py

@@ -97,7 +97,7 @@ def _build_model_mapping(engine, map_install_models, thread_local_log) -> Galaxy
         model_modules.append(tool_shed_install)
 
     model_mapping = GalaxyModelMapping(model_modules, engine)
-    model_mapping.security_agent = GalaxyRBACAgent(model_mapping)
+    model_mapping.security_agent = GalaxyRBACAgent(model_mapping.session)
     model_mapping.thread_local_log = thread_local_log
     return model_mapping
 

lib/galaxy/model/migrations/alembic/versions_gxy/13fe10b8e35b_add_not_null_constraints_to_user_group_.py

@@ -0,0 +1,42 @@
+"""Add not-null constraints to user_group_association
+
+Revision ID: 13fe10b8e35b
+Revises: 56ddf316dbd0
+Create Date: 2024-09-09 21:26:26.032842
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import UserGroupAssociationNullFix
+from galaxy.model.migrations.util import (
+    alter_column,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "13fe10b8e35b"
+down_revision = "56ddf316dbd0"
+branch_labels = None
+depends_on = None
+
+table_name = "user_group_association"
+
+
+def upgrade():
+    with transaction():
+        _remove_records_with_nulls()
+        alter_column(table_name, "user_id", nullable=False)
+        alter_column(table_name, "group_id", nullable=False)
+
+
+def downgrade():
+    with transaction():
+        alter_column(table_name, "user_id", nullable=True)
+        alter_column(table_name, "group_id", nullable=True)
+
+
+def _remove_records_with_nulls():
+    """Remove associations having null as user_id or group_id"""
+    connection = op.get_bind()
+    UserGroupAssociationNullFix(connection).run()

lib/galaxy/model/migrations/alembic/versions_gxy/1fdd615f2cdb_add_not_null_constraints_to_user_role_.py

@@ -0,0 +1,42 @@
+"""Add not-null constraints to user_role_association
+
+Revision ID: 1fdd615f2cdb
+Revises: 349dd9d9aac9
+Create Date: 2024-09-09 21:28:11.987054
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import UserRoleAssociationNullFix
+from galaxy.model.migrations.util import (
+    alter_column,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "1fdd615f2cdb"
+down_revision = "349dd9d9aac9"
+branch_labels = None
+depends_on = None
+
+table_name = "user_role_association"
+
+
+def upgrade():
+    with transaction():
+        _remove_records_with_nulls()
+        alter_column(table_name, "user_id", nullable=False)
+        alter_column(table_name, "role_id", nullable=False)
+
+
+def downgrade():
+    with transaction():
+        alter_column(table_name, "user_id", nullable=True)
+        alter_column(table_name, "role_id", nullable=True)
+
+
+def _remove_records_with_nulls():
+    """Remove associations having null as user_id or role_id"""
+    connection = op.get_bind()
+    UserRoleAssociationNullFix(connection).run()

lib/galaxy/model/migrations/alembic/versions_gxy/25b092f7938b_add_not_null_constraints_to_group_role_.py

@@ -0,0 +1,42 @@
+"""Add not-null constraints to group_role_association
+
+Revision ID: 25b092f7938b
+Revises: 9ef6431f3a4e
+Create Date: 2024-09-09 16:17:26.652865
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import GroupRoleAssociationNullFix
+from galaxy.model.migrations.util import (
+    alter_column,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "25b092f7938b"
+down_revision = "9ef6431f3a4e"
+branch_labels = None
+depends_on = None
+
+table_name = "group_role_association"
+
+
+def upgrade():
+    with transaction():
+        _remove_records_with_nulls()
+        alter_column(table_name, "group_id", nullable=True)
+        alter_column(table_name, "role_id", nullable=False)
+
+
+def downgrade():
+    with transaction():
+        alter_column(table_name, "group_id", nullable=True)
+        alter_column(table_name, "role_id", nullable=True)
+
+
+def _remove_records_with_nulls():
+    """Remove associations having null as group_id or role_id"""
+    connection = op.get_bind()
+    GroupRoleAssociationNullFix(connection).run()

lib/galaxy/model/migrations/alembic/versions_gxy/349dd9d9aac9_add_unique_constraint_to_user_role_assoc.py

@@ -0,0 +1,45 @@
+"""Add unique constraint to user_role_association
+
+Revision ID: 349dd9d9aac9
+Revises: 1cf595475b58
+Create Date: 2024-09-09 16:14:58.278850
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import UserRoleAssociationDuplicateFix
+from galaxy.model.migrations.util import (
+    create_unique_constraint,
+    drop_constraint,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "349dd9d9aac9"
+down_revision = "1cf595475b58"
+branch_labels = None
+depends_on = None
+
+table_name = "user_role_association"
+constraint_column_names = ["user_id", "role_id"]
+unique_constraint_name = (
+    "user_role_association_user_id_key"  # This is what the model's naming convention will generate.
+)
+
+
+def upgrade():
+    with transaction():
+        _remove_duplicate_records()
+        create_unique_constraint(unique_constraint_name, table_name, constraint_column_names)
+
+
+def downgrade():
+    with transaction():
+        drop_constraint(unique_constraint_name, table_name)
+
+
+def _remove_duplicate_records():
+    """Remove duplicate associations"""
+    connection = op.get_bind()
+    UserRoleAssociationDuplicateFix(connection).run()

lib/galaxy/model/migrations/alembic/versions_gxy/56ddf316dbd0_add_unique_constraint_to_user_group_.py

@@ -0,0 +1,45 @@
+"""Add unique constraint to user_group_association
+
+Revision ID: 56ddf316dbd0
+Revises: 1fdd615f2cdb
+Create Date: 2024-09-09 16:10:37.081834
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import UserGroupAssociationDuplicateFix
+from galaxy.model.migrations.util import (
+    create_unique_constraint,
+    drop_constraint,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "56ddf316dbd0"
+down_revision = "1fdd615f2cdb"
+branch_labels = None
+depends_on = None
+
+table_name = "user_group_association"
+constraint_column_names = ["user_id", "group_id"]
+unique_constraint_name = (
+    "user_group_association_user_id_key"  # This is what the model's naming convention will generate.
+)
+
+
+def upgrade():
+    with transaction():
+        _remove_duplicate_records()
+        create_unique_constraint(unique_constraint_name, table_name, constraint_column_names)
+
+
+def downgrade():
+    with transaction():
+        drop_constraint(unique_constraint_name, table_name)
+
+
+def _remove_duplicate_records():
+    """Remove duplicate associations"""
+    connection = op.get_bind()
+    UserGroupAssociationDuplicateFix(connection).run()

lib/galaxy/model/migrations/alembic/versions_gxy/9ef6431f3a4e_add_unique_constraint_to_group_role_.py

@@ -0,0 +1,45 @@
+"""Add unique constraint to group_role_association
+
+Revision ID: 9ef6431f3a4e
+Revises: 13fe10b8e35b
+Create Date: 2024-09-09 15:01:20.426534
+
+"""
+
+from alembic import op
+
+from galaxy.model.migrations.data_fixes.association_table_fixer import GroupRoleAssociationDuplicateFix
+from galaxy.model.migrations.util import (
+    create_unique_constraint,
+    drop_constraint,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "9ef6431f3a4e"
+down_revision = "13fe10b8e35b"
+branch_labels = None
+depends_on = None
+
+table_name = "group_role_association"
+constraint_column_names = ["group_id", "role_id"]
+unique_constraint_name = (
+    "group_role_association_group_id_key"  # This is what the model's naming convention will generate.
+)
+
+
+def upgrade():
+    with transaction():
+        _remove_duplicate_records()
+        create_unique_constraint(unique_constraint_name, table_name, constraint_column_names)
+
+
+def downgrade():
+    with transaction():
+        drop_constraint(unique_constraint_name, table_name)
+
+
+def _remove_duplicate_records():
+    """Remove duplicate associations"""
+    connection = op.get_bind()
+    GroupRoleAssociationDuplicateFix(connection).run()

lib/galaxy/model/migrations/data_fixes/association_table_fixer.py

@@ -0,0 +1,200 @@
+from abc import (
+    ABC,
+    abstractmethod,
+)
+
+from sqlalchemy import (
+    delete,
+    func,
+    null,
+    or_,
+    select,
+)
+
+from galaxy.model import (
+    GroupRoleAssociation,
+    UserGroupAssociation,
+    UserRoleAssociation,
+)
+
+
+class AssociationNullFix(ABC):
+
+    def __init__(self, connection):
+        self.connection = connection
+        self.assoc_model = self.association_model()
+        self.assoc_name = self.assoc_model.__tablename__
+        self.where_clause = self.build_where_clause()
+
+    def run(self):
+        invalid_assocs = self.count_associations_with_nulls()
+        if invalid_assocs:
+            self.delete_associations_with_nulls()
+
+    def count_associations_with_nulls(
+        self,
+    ):
+        """
+        Retrieve association records where one or both associated item ids are null.
+        """
+        select_stmt = select(func.count()).where(self.where_clause)
+        return self.connection.scalar(select_stmt)
+
+    def delete_associations_with_nulls(self):
+        """
+        Delete association records where one or both associated item ids are null.
+        """
+        delete_stmt = delete(self.assoc_model).where(self.where_clause)
+        self.connection.execute(delete_stmt)
+
+    @abstractmethod
+    def association_model(self):
+        """Return model class"""
+
+    @abstractmethod
+    def build_where_clause(self):
+        """Build where clause for filtering records containing nulls instead of associated item ids"""
+
+
+class UserGroupAssociationNullFix(AssociationNullFix):
+
+    def association_model(self):
+        return UserGroupAssociation
+
+    def build_where_clause(self):
+        return or_(UserGroupAssociation.user_id == null(), UserGroupAssociation.group_id == null())
+
+
+class UserRoleAssociationNullFix(AssociationNullFix):
+
+    def association_model(self):
+        return UserRoleAssociation
+
+    def build_where_clause(self):
+        return or_(UserRoleAssociation.user_id == null(), UserRoleAssociation.role_id == null())
+
+
+class GroupRoleAssociationNullFix(AssociationNullFix):
+
+    def association_model(self):
+        return GroupRoleAssociation
+
+    def build_where_clause(self):
+        return or_(GroupRoleAssociation.group_id == null(), GroupRoleAssociation.role_id == null())
+
+
+class AssociationDuplicateFix(ABC):
+
+    def __init__(self, connection):
+        self.connection = connection
+        self.assoc_model = self.association_model()
+        self.assoc_name = self.assoc_model.__tablename__
+
+    def run(self):
+        duplicate_assocs = self.select_duplicate_associations()
+        if duplicate_assocs:
+            self.delete_duplicate_associations(duplicate_assocs)
+
+    def select_duplicate_associations(self):
+        """Retrieve duplicate association records."""
+        select_stmt = self.build_duplicate_tuples_statement()
+        return self.connection.execute(select_stmt).all()
+
+    @abstractmethod
+    def association_model(self):
+        """Return model class"""
+
+    @abstractmethod
+    def build_duplicate_tuples_statement(self):
+        """
+        Build select statement returning a list of tuples (item1_id, item2_id) that have counts > 1
+        """
+
+    @abstractmethod
+    def build_duplicate_ids_statement(self, item1_id, item2_id):
+        """
+        Build select statement returning a list of ids for duplicate records retrieved via build_duplicate_tuples_statement().
+        """
+
+    def delete_duplicate_associations(self, records):
+        """
+        Delete duplicate association records retaining oldest record in each group of duplicates.
+        """
+        to_delete = []
+        for item1_id, item2_id in records:
+            to_delete += self._get_duplicates_to_delete(item1_id, item2_id)
+        for id in to_delete:
+            delete_stmt = delete(self.assoc_model).where(self.assoc_model.id == id)
+            self.connection.execute(delete_stmt)
+
+    def _get_duplicates_to_delete(self, item1_id, item2_id):
+        stmt = self.build_duplicate_ids_statement(item1_id, item2_id)
+        duplicates = self.connection.scalars(stmt).all()
+        # IMPORTANT: we slice to skip the first item ([1:]), which is the oldest record and SHOULD NOT BE DELETED.
+        return duplicates[1:]
+
+
+class UserGroupAssociationDuplicateFix(AssociationDuplicateFix):
+
+    def association_model(self):
+        return UserGroupAssociation
+
+    def build_duplicate_tuples_statement(self):
+        stmt = (
+            select(UserGroupAssociation.user_id, UserGroupAssociation.group_id)
+            .group_by(UserGroupAssociation.user_id, UserGroupAssociation.group_id)
+            .having(func.count() > 1)
+        )
+        return stmt
+
+    def build_duplicate_ids_statement(self, user_id, group_id):
+        stmt = (
+            select(UserGroupAssociation.id)
+            .where(UserGroupAssociation.user_id == user_id, UserGroupAssociation.group_id == group_id)
+            .order_by(UserGroupAssociation.update_time)
+        )
+        return stmt
+
+
+class UserRoleAssociationDuplicateFix(AssociationDuplicateFix):
+
+    def association_model(self):
+        return UserRoleAssociation
+
+    def build_duplicate_tuples_statement(self):
+        stmt = (
+            select(UserRoleAssociation.user_id, UserRoleAssociation.role_id)
+            .group_by(UserRoleAssociation.user_id, UserRoleAssociation.role_id)
+            .having(func.count() > 1)
+        )
+        return stmt
+
+    def build_duplicate_ids_statement(self, user_id, role_id):
+        stmt = (
+            select(UserRoleAssociation.id)
+            .where(UserRoleAssociation.user_id == user_id, UserRoleAssociation.role_id == role_id)
+            .order_by(UserRoleAssociation.update_time)
+        )
+        return stmt
+
+
+class GroupRoleAssociationDuplicateFix(AssociationDuplicateFix):
+
+    def association_model(self):
+        return GroupRoleAssociation
+
+    def build_duplicate_tuples_statement(self):
+        stmt = (
+            select(GroupRoleAssociation.group_id, GroupRoleAssociation.role_id)
+            .group_by(GroupRoleAssociation.group_id, GroupRoleAssociation.role_id)
+            .having(func.count() > 1)
+        )
+        return stmt
+
+    def build_duplicate_ids_statement(self, group_id, role_id):
+        stmt = (
+            select(GroupRoleAssociation.id)
+            .where(GroupRoleAssociation.group_id == group_id, GroupRoleAssociation.role_id == role_id)
+            .order_by(GroupRoleAssociation.update_time)
+        )
+        return stmt

lib/galaxy/model/security.py

@@ -1,33 +1,48 @@
 import logging
 import socket
+import sqlite3
 from datetime import (
     datetime,
     timedelta,
 )
-from typing import List
+from typing import (
+    List,
+    Optional,
+)
 
 from sqlalchemy import (
     and_,
+    delete,
     false,
     func,
+    insert,
     not_,
     or_,
     select,
+    text,
 )
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload
-from sqlalchemy.sql import text
 
 import galaxy.model
+from galaxy.exceptions import RequestParameterInvalidException
 from galaxy.model import (
     Dataset,
+    DatasetCollection,
     DatasetPermissions,
+    DefaultHistoryPermissions,
+    DefaultUserPermissions,
     Group,
     GroupRoleAssociation,
     HistoryDatasetAssociationDisplayAtAuthorization,
     Library,
     LibraryDataset,
+    LibraryDatasetCollectionAssociation,
     LibraryDatasetDatasetAssociation,
+    LibraryDatasetDatasetAssociationPermissions,
     LibraryDatasetPermissions,
+    LibraryFolder,
+    LibraryFolderPermissions,
     LibraryPermissions,
     Role,
     User,
@@ -51,23 +66,18 @@ log = logging.getLogger(__name__)
 
 
 class GalaxyRBACAgent(RBACAgent):
-    def __init__(self, model, permitted_actions=None):
-        self.model = model
+    def __init__(self, sa_session, permitted_actions=None):
+        self.sa_session = sa_session
         if permitted_actions:
             self.permitted_actions = permitted_actions
         # List of "library_item" objects and their associated permissions and info template objects
         self.library_item_assocs = (
-            (self.model.Library, self.model.LibraryPermissions),
-            (self.model.LibraryFolder, self.model.LibraryFolderPermissions),
-            (self.model.LibraryDataset, self.model.LibraryDatasetPermissions),
-            (self.model.LibraryDatasetDatasetAssociation, self.model.LibraryDatasetDatasetAssociationPermissions),
+            (Library, LibraryPermissions),
+            (LibraryFolder, LibraryFolderPermissions),
+            (LibraryDataset, LibraryDatasetPermissions),
+            (LibraryDatasetDatasetAssociation, LibraryDatasetDatasetAssociationPermissions),
         )
 
-    @property
-    def sa_session(self):
-        """Returns a SQLAlchemy session"""
-        return self.model.context
-
     def sort_by_attr(self, seq, attr):
         """
         Sort the sequence of objects by object's attribute
@@ -139,11 +149,11 @@ class GalaxyRBACAgent(RBACAgent):
         else:
             limit = None
         total_count = None
-        if isinstance(item, self.model.Library) and self.library_is_public(item):
+        if isinstance(item, Library) and self.library_is_public(item):
             is_public_item = True
-        elif isinstance(item, self.model.Dataset) and self.dataset_is_public(item):
+        elif isinstance(item, Dataset) and self.dataset_is_public(item):
             is_public_item = True
-        elif isinstance(item, self.model.LibraryFolder):
+        elif isinstance(item, LibraryFolder):
             is_public_item = True
         else:
             is_public_item = False
@@ -238,8 +248,8 @@ class GalaxyRBACAgent(RBACAgent):
         """
         admin_controller = cntrller in ["library_admin"]
         roles = set()
-        if (isinstance(item, self.model.Library) and self.library_is_public(item)) or (
-            isinstance(item, self.model.Dataset) and self.dataset_is_public(item)
+        if (isinstance(item, Library) and self.library_is_public(item)) or (
+            isinstance(item, Dataset) and self.dataset_is_public(item)
         ):
             return self.get_all_roles(trans, cntrller)
         # If item has roles associated with the access permission, we need to start with them.
@@ -272,13 +282,13 @@ class GalaxyRBACAgent(RBACAgent):
         """
         role_type = role.type
         if user:
-            if role_type == self.model.Role.types.PRIVATE:
+            if role_type == Role.types.PRIVATE:
                 return role == self.get_private_user_role(user)
-            if role_type == self.model.Role.types.SHARING:
+            if role_type == Role.types.SHARING:
                 return role in self.get_sharing_roles(user)
             # If role_type is neither private nor sharing, it's ok to display
             return True
-        return role_type != self.model.Role.types.PRIVATE and role_type != self.model.Role.types.SHARING
+        return role_type != Role.types.PRIVATE and role_type != Role.types.SHARING
 
     def allow_action(self, roles, action, item):
         """
@@ -329,7 +339,7 @@ class GalaxyRBACAgent(RBACAgent):
         ret_permissions = {}
         if len(permission_items) > 0:
             # SM: NB: LibraryDatasets became Datasets for some odd reason.
-            if isinstance(permission_items[0], trans.model.LibraryDataset):
+            if isinstance(permission_items[0], LibraryDataset):
                 ids = [item.library_dataset_id for item in permission_items]
                 stmt = select(LibraryDatasetPermissions).where(
                     and_(
@@ -348,7 +358,7 @@ class GalaxyRBACAgent(RBACAgent):
                     ret_permissions[item.library_dataset_id] = []
                 for permission in permissions:
                     ret_permissions[permission.library_dataset_id].append(permission)
-            elif isinstance(permission_items[0], trans.model.Dataset):
+            elif isinstance(permission_items[0], Dataset):
                 ids = [item.id for item in permission_items]
 
                 stmt = select(DatasetPermissions).where(
@@ -499,7 +509,7 @@ class GalaxyRBACAgent(RBACAgent):
     def item_permission_map_for_add(self, trans, user_roles, libitems):
         return self.allow_action_on_libitems(trans, user_roles, self.permitted_actions.LIBRARY_ADD, libitems)
 
-    def can_access_dataset(self, user_roles, dataset: galaxy.model.Dataset):
+    def can_access_dataset(self, user_roles, dataset: Dataset):
         # SM: dataset_is_public will access dataset.actions, which is a
         # backref that causes a query to be made to DatasetPermissions
         retval = self.dataset_is_public(dataset) or self.allow_action(
@@ -518,7 +528,7 @@ class GalaxyRBACAgent(RBACAgent):
 
         return True
 
-    def can_access_collection(self, user_roles: List[galaxy.model.Role], collection: galaxy.model.DatasetCollection):
+    def can_access_collection(self, user_roles: List[Role], collection: DatasetCollection):
         action_tuples = collection.dataset_action_tuples
         if not self.can_access_datasets(user_roles, action_tuples):
             return False
@@ -599,21 +609,21 @@ class GalaxyRBACAgent(RBACAgent):
         return False
 
     def can_access_library_item(self, roles, item, user):
-        if isinstance(item, self.model.Library):
+        if isinstance(item, Library):
             return self.can_access_library(roles, item)
-        elif isinstance(item, self.model.LibraryFolder):
+        elif isinstance(item, LibraryFolder):
             return (
                 self.can_access_library(roles, item.parent_library) and self.check_folder_contents(user, roles, item)[0]
             )
-        elif isinstance(item, self.model.LibraryDataset):
+        elif isinstance(item, LibraryDataset):
             return self.can_access_library(roles, item.folder.parent_library) and self.can_access_dataset(
                 roles, item.library_dataset_dataset_association.dataset
             )
-        elif isinstance(item, self.model.LibraryDatasetDatasetAssociation):
+        elif isinstance(item, LibraryDatasetDatasetAssociation):
             return self.can_access_library(
                 roles, item.library_dataset.folder.parent_library
             ) and self.can_access_dataset(roles, item.dataset)
-        elif isinstance(item, self.model.LibraryDatasetCollectionAssociation):
+        elif isinstance(item, LibraryDatasetCollectionAssociation):
             return self.can_access_library(roles, item.folder.parent_library)
         else:
             log.warning(f"Unknown library item type: {type(item)}")
@@ -658,7 +668,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         datasets = datasets or []
         perms = {}
         for dataset in datasets:
-            if not isinstance(dataset, self.model.Dataset):
+            if not isinstance(dataset, Dataset):
                 dataset = dataset.dataset
             these_perms = {}
             # initialize blank perms
@@ -700,43 +710,29 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     perms[action].extend([_ for _ in role_ids if _ not in perms[action]])
         return perms
 
-    def associate_components(self, **kwd):
-        if "user" in kwd:
-            if "group" in kwd:
-                return self.associate_user_group(kwd["user"], kwd["group"])
-            elif "role" in kwd:
-                return self.associate_user_role(kwd["user"], kwd["role"])
-        elif "role" in kwd:
-            if "group" in kwd:
-                return self.associate_group_role(kwd["group"], kwd["role"])
-        if "action" in kwd:
-            if "dataset" in kwd and "role" in kwd:
-                return self.associate_action_dataset_role(kwd["action"], kwd["dataset"], kwd["role"])
-        raise Exception(f"No valid method of associating provided components: {kwd}")
-
     def associate_user_group(self, user, group):
-        assoc = self.model.UserGroupAssociation(user, group)
+        assoc = UserGroupAssociation(user, group)
         self.sa_session.add(assoc)
         with transaction(self.sa_session):
             self.sa_session.commit()
         return assoc
 
     def associate_user_role(self, user, role):
-        assoc = self.model.UserRoleAssociation(user, role)
+        assoc = UserRoleAssociation(user, role)
         self.sa_session.add(assoc)
         with transaction(self.sa_session):
             self.sa_session.commit()
         return assoc
 
     def associate_group_role(self, group, role):
-        assoc = self.model.GroupRoleAssociation(group, role)
+        assoc = GroupRoleAssociation(group, role)
         self.sa_session.add(assoc)
         with transaction(self.sa_session):
             self.sa_session.commit()
         return assoc
 
     def associate_action_dataset_role(self, action, dataset, role):
-        assoc = self.model.DatasetPermissions(action, dataset, role)
+        assoc = DatasetPermissions(action, dataset, role)
         self.sa_session.add(assoc)
         with transaction(self.sa_session):
             self.sa_session.commit()
@@ -767,14 +763,14 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         return role
 
     def get_role(self, name, type=None):
-        type = type or self.model.Role.types.SYSTEM
+        type = type or Role.types.SYSTEM
         # will raise exception if not found
         stmt = select(Role).where(and_(Role.name == name, Role.type == type))
         return self.sa_session.execute(stmt).scalar_one()
 
     def create_role(self, name, description, in_users, in_groups, create_group_for_role=False, type=None):
-        type = type or self.model.Role.types.SYSTEM
-        role = self.model.Role(name=name, description=description, type=type)
+        type = type or Role.types.SYSTEM
+        role = Role(name=name, description=description, type=type)
         self.sa_session.add(role)
         # Create the UserRoleAssociations
         for user in [self.sa_session.get(User, x) for x in in_users]:
@@ -784,7 +780,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
             self.associate_group_role(group, role)
         if create_group_for_role:
             # Create the group
-            group = self.model.Group(name=name)
+            group = Group(name=name)
             self.sa_session.add(group)
             # Associate the group with the role
             self.associate_group_role(group, role)
@@ -831,7 +827,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         for action, roles in permissions.items():
             if isinstance(action, Action):
                 action = action.action
-            for dup in [self.model.DefaultUserPermissions(user, action, role) for role in roles]:
+            for dup in [DefaultUserPermissions(user, action, role) for role in roles]:
                 self.sa_session.add(dup)
                 flush_needed = True
         if flush_needed:
@@ -871,7 +867,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         for action, roles in permissions.items():
             if isinstance(action, Action):
                 action = action.action
-            for dhp in [self.model.DefaultHistoryPermissions(history, action, role) for role in roles]:
+            for dhp in [DefaultHistoryPermissions(history, action, role) for role in roles]:
                 self.sa_session.add(dhp)
                 flush_needed = True
         if flush_needed:
@@ -922,7 +918,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
             for _, roles in _walk_action_roles(permissions, self.permitted_actions.DATASET_ACCESS):
                 dataset_access_roles.extend(roles)
 
-            if len(dataset_access_roles) != 1 or dataset_access_roles[0].type != self.model.Role.types.PRIVATE:
+            if len(dataset_access_roles) != 1 or dataset_access_roles[0].type != Role.types.PRIVATE:
                 return galaxy.model.CANNOT_SHARE_PRIVATE_DATASET_MESSAGE
 
         flush_needed = False
@@ -940,7 +936,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     role_id = role.id
                 else:
                     role_id = role
-                dp = self.model.DatasetPermissions(action, dataset, role_id=role_id)
+                dp = DatasetPermissions(action, dataset, role_id=role_id)
                 self.sa_session.add(dp)
                 flush_needed = True
         if flush_needed and flush:
@@ -970,7 +966,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     self.sa_session.delete(dp)
                     flush_needed = True
             # Add the new specific permission on the dataset
-            for dp in [self.model.DatasetPermissions(action, dataset, role) for role in roles]:
+            for dp in [DatasetPermissions(action, dataset, role) for role in roles]:
                 self.sa_session.add(dp)
                 flush_needed = True
         if flush_needed:
@@ -993,9 +989,9 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         return permissions
 
     def copy_dataset_permissions(self, src, dst, flush=True):
-        if not isinstance(src, self.model.Dataset):
+        if not isinstance(src, Dataset):
             src = src.dataset
-        if not isinstance(dst, self.model.Dataset):
+        if not isinstance(dst, Dataset):
             dst = dst.dataset
         self.set_all_dataset_permissions(dst, self.get_permissions(src), flush=flush)
 
@@ -1004,7 +1000,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         intersect = None
         users = users or []
         for user in users:
-            roles = [ura.role for ura in user.roles if ura.role.type == self.model.Role.types.SHARING]
+            roles = [ura.role for ura in user.roles if ura.role.type == Role.types.SHARING]
             if intersect is None:
                 intersect = roles
             else:
@@ -1021,14 +1017,12 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     sharing_role = role
                     break
         if sharing_role is None:
-            sharing_role = self.model.Role(
-                name=f"Sharing role for: {', '.join(u.email for u in users)}", type=self.model.Role.types.SHARING
-            )
+            sharing_role = Role(name=f"Sharing role for: {', '.join(u.email for u in users)}", type=Role.types.SHARING)
             self.sa_session.add(sharing_role)
             with transaction(self.sa_session):
                 self.sa_session.commit()
             for user in users:
-                self.associate_components(user=user, role=sharing_role)
+                self.associate_user_role(user, sharing_role)
         self.set_dataset_permission(dataset, {self.permitted_actions.DATASET_ACCESS: [sharing_role]})
 
     def set_all_library_permissions(self, trans, library_item, permissions=None):
@@ -1047,7 +1041,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     for role_assoc in [permission_class(action, library_item, role) for role in roles]:
                         self.sa_session.add(role_assoc)
                         flush_needed = True
-                    if isinstance(library_item, self.model.LibraryDatasetDatasetAssociation):
+                    if isinstance(library_item, LibraryDatasetDatasetAssociation):
                         # Permission setting related to DATASET_MANAGE_PERMISSIONS was broken for a period of time,
                         # so it is possible that some Datasets have no roles associated with the DATASET_MANAGE_PERMISSIONS
                         # permission.  In this case, we'll reset this permission to the library_item user's private role.
@@ -1086,14 +1080,12 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     self.sa_session.delete(item_permission)
                     flush_needed = True
             # Add the new specific permission on the library item
-            if isinstance(library_item, self.model.LibraryDataset):
-                for item_permission in [
-                    self.model.LibraryDatasetPermissions(action, library_item, role) for role in roles
-                ]:
+            if isinstance(library_item, LibraryDataset):
+                for item_permission in [LibraryDatasetPermissions(action, library_item, role) for role in roles]:
                     self.sa_session.add(item_permission)
                     flush_needed = True
-            elif isinstance(library_item, self.model.LibraryPermissions):
-                for item_permission in [self.model.LibraryPermissions(action, library_item, role) for role in roles]:
+            elif isinstance(library_item, LibraryPermissions):
+                for item_permission in [LibraryPermissions(action, library_item, role) for role in roles]:
                     self.sa_session.add(item_permission)
                     flush_needed = True
         if flush_needed:
@@ -1151,7 +1143,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
             if not dataset.purged and not self.dataset_is_public(dataset):
                 self.make_dataset_public(dataset)
 
-    def dataset_is_public(self, dataset: galaxy.model.Dataset):
+    def dataset_is_public(self, dataset: Dataset):
         """
         A dataset is considered public if there are no "access" actions
         associated with it.  Any other actions ( 'manage permissions',
@@ -1194,7 +1186,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
             return False
         else:
             access_role = access_roles[0]
-            return access_role.type == self.model.Role.types.PRIVATE
+            return access_role.type == Role.types.PRIVATE
 
     def datasets_are_public(self, trans, datasets):
         """
@@ -1294,7 +1286,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     # permission on this dataset, or the dataset is not accessible.
                     # Since we have more than 1 role, none of them can be private.
                     for role in in_roles:
-                        if role.type == self.model.Role.types.PRIVATE:
+                        if role.type == Role.types.PRIVATE:
                             private_role_found = True
                             break
                 if len(in_roles) == 1:
@@ -1358,7 +1350,7 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                     f"Invalid class ({target_library_item.__class__}) specified for target_library_item ({target_library_item.__class__.__name__})"
                 )
             # Make sure user's private role is included
-            private_role = self.model.security_agent.get_private_user_role(user)
+            private_role = self.get_private_user_role(user)
             for action in self.permitted_actions.values():
                 if not found_permission_class.filter_by(role_id=private_role.id, action=action.action).first():
                     lp = found_permission_class(action.action, target_library_item, private_role)
@@ -1407,9 +1399,9 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         for action in actions_to_check:
             if self.allow_action(roles, action, library_item):
                 return True, hidden_folder_ids
-        if isinstance(library_item, self.model.Library):
+        if isinstance(library_item, Library):
             return self.show_library_item(user, roles, library_item.root_folder, actions_to_check, hidden_folder_ids="")
-        if isinstance(library_item, self.model.LibraryFolder):
+        if isinstance(library_item, LibraryFolder):
             for folder in library_item.active_folders:
                 can_show, hidden_folder_ids = self.show_library_item(
                     user, roles, folder, actions_to_check, hidden_folder_ids=hidden_folder_ids
@@ -1433,11 +1425,11 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
         """
         hidden_folder_ids = hidden_folder_ids or []
         showable_folders = showable_folders or []
-        if isinstance(library_item, self.model.Library):
+        if isinstance(library_item, Library):
             return self.get_showable_folders(
                 user, roles, library_item.root_folder, actions_to_check, showable_folders=[]
             )
-        if isinstance(library_item, self.model.LibraryFolder):
+        if isinstance(library_item, LibraryFolder):
             if library_item.id not in hidden_folder_ids:
                 for action in actions_to_check:
                     if self.allow_action(roles, action, library_item):
@@ -1447,62 +1439,171 @@ WHERE history.user_id != :user_id and history_dataset_association.dataset_id = :
                 self.get_showable_folders(user, roles, folder, actions_to_check, showable_folders=showable_folders)
         return showable_folders
 
-    def set_entity_user_associations(self, users=None, roles=None, groups=None, delete_existing_assocs=True):
-        users = users or []
-        roles = roles or []
-        groups = groups or []
-        for user in users:
-            if delete_existing_assocs:
-                flush_needed = False
-                for a in user.non_private_roles + user.groups:
-                    self.sa_session.delete(a)
-                    flush_needed = True
-                if flush_needed:
-                    with transaction(self.sa_session):
-                        self.sa_session.commit()
-            self.sa_session.refresh(user)
-            for role in roles:
-                # Make sure we are not creating an additional association with a PRIVATE role
-                if role not in [x.role for x in user.roles]:
-                    self.associate_components(user=user, role=role)
-            for group in groups:
-                self.associate_components(user=user, group=group)
+    def set_user_group_and_role_associations(
+        self,
+        user: User,
+        *,
+        group_ids: Optional[List[int]] = None,
+        role_ids: Optional[List[int]] = None,
+    ) -> None:
+        """
+        Set user groups and user roles, replacing current associations.
 
-    def set_entity_group_associations(self, groups=None, users=None, roles=None, delete_existing_assocs=True):
-        users = users or []
-        roles = roles or []
-        groups = groups or []
-        for group in groups:
-            if delete_existing_assocs:
-                flush_needed = False
-                for a in group.roles + group.users:
-                    self.sa_session.delete(a)
-                    flush_needed = True
-                if flush_needed:
-                    with transaction(self.sa_session):
-                        self.sa_session.commit()
-            for role in roles:
-                self.associate_components(group=group, role=role)
-            for user in users:
-                self.associate_components(group=group, user=user)
+        Associations are set only if a list of new associations is provided.
+        If the provided list is empty, existing associations will be removed.
+        If the provided value is None, existing associations will not be updated.
+        """
+        self._persist_new_model(user)
+        if group_ids is not None:
+            self._set_user_groups(user, group_ids)
+        if role_ids is not None:
+            self._set_user_roles(user, role_ids)
+        # Commit only if both user groups and user roles have been set.
+        self.sa_session.commit()
 
-    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 set_group_user_and_role_associations(
+        self,
+        group: Group,
+        *,
+        user_ids: Optional[List[int]] = None,
+        role_ids: Optional[List[int]] = None,
+    ) -> None:
+        """
+        Set group users and group roles, replacing current associations.
+
+        Associations are set only if a list of new associations is provided.
+        If the provided list is empty, existing associations will be removed.
+        If the provided value is None, existing associations will not be updated.
+        """
+        self._persist_new_model(group)
+        if user_ids is not None:
+            self._set_group_users(group, user_ids)
+        if role_ids is not None:
+            self._set_group_roles(group, role_ids)
+        # Commit only if both group users and group roles have been set.
+        self.sa_session.commit()
+
+    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.
+
+        Associations are set only if a list of new associations is provided.
+        If the provided list is empty, existing associations will be removed.
+        If the provided value is None, existing associations will not be updated.
+        """
+        self._persist_new_model(role)
+        if user_ids is not None:
+            self._set_role_users(role, user_ids)
+        if group_ids is not None:
+            self._set_role_groups(role, group_ids)
+        # Commit only if both role users and role groups have been set.
+        self.sa_session.commit()
+
+    def _set_user_groups(self, user, group_ids):
+        delete_stmt = delete(UserGroupAssociation).where(UserGroupAssociation.user_id == user.id)
+        insert_values = [{"user_id": user.id, "group_id": group_id} for group_id in group_ids]
+        self._set_associations(user, UserGroupAssociation, delete_stmt, insert_values)
+
+    def _set_user_roles(self, user, role_ids):
+        # Do not include user's private role association in delete statement.
+        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.user_id == user.id)
+        private_role = get_private_user_role(user, self.sa_session)
+        if not private_role:
+            log.warning("User %s does not have a private role assigned", user)
+        else:
+            delete_stmt = delete_stmt.where(UserRoleAssociation.role_id != private_role.id)
+        role_ids = self._filter_private_roles(role_ids)
+        # breakpoint()
+
+        insert_values = [{"user_id": user.id, "role_id": role_id} for role_id in role_ids]
+        self._set_associations(user, UserRoleAssociation, delete_stmt, insert_values)
+
+    def _filter_private_roles(self, role_ids):
+        """Filter out IDs of private roles: those should not be assignable via UI"""
+        stmt = select(Role.id).where(Role.id.in_(role_ids)).where(Role.type == Role.types.PRIVATE)
+        private_role_ids = self.sa_session.scalars(stmt).all()
+        # We could simply select only private roles; however, that would get rid of potential duplicates
+        # and invalid role_ids; which would hide any bugs that should be caught in the _set_associations() method.
+        return [role_id for role_id in role_ids if role_id not in private_role_ids]
+
+    def _set_group_users(self, group, user_ids):
+        delete_stmt = delete(UserGroupAssociation).where(UserGroupAssociation.group_id == group.id)
+        insert_values = [{"group_id": group.id, "user_id": user_id} for user_id in user_ids]
+        self._set_associations(group, UserGroupAssociation, delete_stmt, insert_values)
+
+    def _set_group_roles(self, group, role_ids):
+        delete_stmt = delete(GroupRoleAssociation).where(GroupRoleAssociation.group_id == group.id)
+        insert_values = [{"group_id": group.id, "role_id": role_id} for role_id in role_ids]
+        self._set_associations(group, GroupRoleAssociation, delete_stmt, insert_values)
+
+    def _set_role_users(self, role, user_ids):
+        # Do not set users if the role is private
+        # Even though we do not expect to be handling a private role here, the following code is
+        # a safeguard against deleting a user-role-association record for a private role.
+        if role.type == Role.types.PRIVATE:
+            return
+
+        # First, check previously associated users to:
+        # - delete DefaultUserPermissions for users that are being removed from this role;
+        # - delete DefaultHistoryPermissions for histories associated with users that are being removed from this role.
+        for ura in role.users:
+            if ura.user_id not in user_ids:  # If a user will be removed from this role, then:
+                user = self.sa_session.get(User, ura.user_id)
+                # Delete DefaultUserPermissions for this user
+                for dup in user.default_permissions:
+                    if role == dup.role:
+                        self.sa_session.delete(dup)
+                # Delete DefaultHistoryPermissions for histories associated with this user
+                for history in user.histories:
+                    for dhp in history.default_permissions:
+                        if role == dhp.role:
+                            self.sa_session.delete(dhp)
+
+        delete_stmt = delete(UserRoleAssociation).where(UserRoleAssociation.role_id == role.id)
+        insert_values = [{"role_id": role.id, "user_id": user_id} for user_id in user_ids]
+        self._set_associations(role, UserRoleAssociation, delete_stmt, insert_values)
+
+    def _set_role_groups(self, role, group_ids):
+        delete_stmt = delete(GroupRoleAssociation).where(GroupRoleAssociation.role_id == role.id)
+        insert_values = [{"role_id": role.id, "group_id": group_id} for group_id in group_ids]
+        self._set_associations(role, GroupRoleAssociation, delete_stmt, insert_values)
+
+    def _persist_new_model(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, parent_model, assoc_model, delete_stmt, insert_values):
+        """
+        Delete current associations for assoc_model, then insert new associations if values are provided.
+        """
+        # Ensure sqlite respects foreign key constraints.
+        if self.sa_session.bind.dialect.name == "sqlite":
+            self.sa_session.execute(text("PRAGMA foreign_keys = ON;"))
+        self.sa_session.execute(delete_stmt)
+        if not insert_values:
+            return
+        try:
+            self.sa_session.execute(insert(assoc_model), insert_values)
+        except IntegrityError as ie:
+            self.sa_session.rollback()
+            if is_unique_constraint_violation(ie):
+                msg = f"Attempting to create a duplicate {assoc_model} record ({insert_values})"
+                log.exception(msg)
+                raise RequestParameterInvalidException()
+            elif is_foreign_key_violation(ie):
+                msg = f"Attempting to create an invalid {assoc_model} record ({insert_values})"
+                log.exception(msg)
+                raise RequestParameterInvalidException()
+            else:
+                raise
 
     def get_component_associations(self, **kwd):
         assert len(kwd) == 2, "You must specify exactly 2 Galaxy security components to check for associations."
@@ -1594,16 +1695,11 @@ class HostAgent(RBACAgent):
         ucsc_archaea=("lowepub.cse.ucsc.edu",),
     )
 
-    def __init__(self, model, permitted_actions=None):
-        self.model = model
+    def __init__(self, sa_session, permitted_actions=None):
+        self.sa_session = sa_session
         if permitted_actions:
             self.permitted_actions = permitted_actions
 
-    @property
-    def sa_session(self):
-        """Returns a SQLAlchemy session"""
-        return self.model.context
-
     def allow_action(self, addr, action, **kwd):
         if "dataset" in kwd and action == self.permitted_actions.DATASET_ACCESS:
             hda = kwd["dataset"]
@@ -1664,7 +1760,7 @@ class HostAgent(RBACAgent):
         if hdadaa:
             hdadaa.update_time = datetime.utcnow()
         else:
-            hdadaa = self.model.HistoryDatasetAssociationDisplayAtAuthorization(hda=hda, user=user, site=site)
+            hdadaa = HistoryDatasetAssociationDisplayAtAuthorization(hda=hda, user=user, site=site)
         self.sa_session.add(hdadaa)
         with transaction(self.sa_session):
             self.sa_session.commit()
@@ -1677,3 +1773,31 @@ def _walk_action_roles(permissions, query_action):
                 yield action, roles
         elif action == query_action.action and roles:
             yield action, roles
+
+
+def is_unique_constraint_violation(error):
+    # A more elegant way to handle sqlite iw this:
+    #   if hasattr(error.orig, "sqlite_errorname"):
+    #       return error.orig.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE"
+    # However, that's only possible with Python 3.11+
+    # https://docs.python.org/3/library/sqlite3.html#sqlite3.Error.sqlite_errorcode
+    if isinstance(error.orig, sqlite3.IntegrityError):
+        return error.orig.args[0].startswith("UNIQUE constraint failed")
+    else:
+        # If this is a PostgreSQL unique constraint, then error.orig is an instance of psycopg2.errors.UniqueViolation
+        # and should have an attribute `pgcode` = 23505.
+        return int(getattr(error.orig, "pgcode", -1)) == 23505
+
+
+def is_foreign_key_violation(error):
+    # A more elegant way to handle sqlite iw this:
+    #   if hasattr(error.orig, "sqlite_errorname"):
+    #       return error.orig.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE"
+    # However, that's only possible with Python 3.11+
+    # https://docs.python.org/3/library/sqlite3.html#sqlite3.Error.sqlite_errorcode
+    if isinstance(error.orig, sqlite3.IntegrityError):
+        return error.orig.args[0] == "FOREIGN KEY constraint failed"
+    else:
+        # If this is a PostgreSQL foreign key error, then error.orig is an instance of psycopg2.errors.ForeignKeyViolation
+        # and should have an attribute `pgcode` = 23503.
+        return int(getattr(error.orig, "pgcode", -1)) == 23503

lib/galaxy/schema/groups.py

@@ -73,5 +73,18 @@ class GroupCreatePayload(Model):
 
 
 @partial_model()
-class GroupUpdatePayload(GroupCreatePayload):
-    pass
+class GroupUpdatePayload(Model):
+    """Payload schema for updating a group."""
+
+    name: str = Field(
+        ...,
+        title="name of the group",
+    )
+    user_ids: Optional[List[DecodedDatabaseIdField]] = Field(
+        None,
+        title="user IDs",
+    )
+    role_ids: Optional[List[DecodedDatabaseIdField]] = Field(
+        None,
+        title="role IDs",
+    )

lib/galaxy/security/__init__.py

@@ -95,9 +95,6 @@ class RBACAgent:
     def can_manage_library_item(self, roles, item):
         raise Exception("Unimplemented Method")
 
-    def associate_components(self, **kwd):
-        raise Exception(f"No valid method of associating provided components: {kwd}")
-
     def create_private_user_role(self, user):
         raise Exception("Unimplemented Method")
 

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

@@ -13,7 +13,10 @@ from galaxy import (
     util,
     web,
 )
-from galaxy.exceptions import ActionInputError
+from galaxy.exceptions import (
+    ActionInputError,
+    RequestParameterInvalidException,
+)
 from galaxy.managers.quotas import QuotaManager
 from galaxy.model.base import transaction
 from galaxy.model.index_filter_util import (
@@ -807,35 +810,17 @@ class AdminGalaxy(controller.JSAppLauncher):
                 ],
             }
         else:
-            in_users = [
-                trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_users"))
-            ]
-            in_groups = [
-                trans.sa_session.query(trans.app.model.Group).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_groups"))
-            ]
-            if None in in_users or None in in_groups:
+            user_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_users"))]
+            group_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_groups"))]
+            try:
+                trans.app.security_agent.set_role_user_and_group_associations(
+                    role, user_ids=user_ids, group_ids=group_ids
+                )
+                return {
+                    "message": f"Role '{role.name}' has been updated with {len(user_ids)} associated users and {len(group_ids)} associated groups."
+                }
+            except RequestParameterInvalidException:
                 return self.message_exception(trans, "One or more invalid user/group id has been provided.")
-            for ura in role.users:
-                user = trans.sa_session.query(trans.app.model.User).get(ura.user_id)
-                if user not in in_users:
-                    # Delete DefaultUserPermissions for previously associated users that have been removed from the role
-                    for dup in user.default_permissions:
-                        if role == dup.role:
-                            trans.sa_session.delete(dup)
-                    # Delete DefaultHistoryPermissions for previously associated users that have been removed from the role
-                    for history in user.histories:
-                        for dhp in history.default_permissions:
-                            if role == dhp.role:
-                                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.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."
-            }
 
     @web.legacy_expose_api
     @web.require_admin
@@ -912,21 +897,17 @@ class AdminGalaxy(controller.JSAppLauncher):
                 ],
             }
         else:
-            in_users = [
-                trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_users"))
-            ]
-            in_roles = [
-                trans.sa_session.query(trans.app.model.Role).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_roles"))
-            ]
-            if None in in_users or None in in_roles:
+            user_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_users"))]
+            role_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_roles"))]
+            try:
+                trans.app.security_agent.set_group_user_and_role_associations(
+                    group, user_ids=user_ids, role_ids=role_ids
+                )
+                return {
+                    "message": f"Group '{group.name}' has been updated with {len(user_ids)} associated users and {len(role_ids)} associated roles."
+                }
+            except RequestParameterInvalidException:
                 return self.message_exception(trans, "One or more invalid user/role id has been provided.")
-            trans.app.security_agent.set_entity_group_associations(groups=[group], users=in_users, roles=in_roles)
-            trans.sa_session.refresh(group)
-            return {
-                "message": f"Group '{group.name}' has been updated with {len(in_users)} associated users and {len(in_roles)} associated roles."
-            }
 
     @web.legacy_expose_api
     @web.require_admin
@@ -1099,28 +1080,18 @@ class AdminGalaxy(controller.JSAppLauncher):
                 ],
             }
         else:
-            in_roles = [
-                trans.sa_session.query(trans.app.model.Role).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_roles"))
-            ]
-            in_groups = [
-                trans.sa_session.query(trans.app.model.Group).get(trans.security.decode_id(x))
-                for x in util.listify(payload.get("in_groups"))
-            ]
-            if None in in_groups or None in in_roles:
+            role_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_roles"))]
+            group_ids = [trans.security.decode_id(id) for id in util.listify(payload.get("in_groups"))]
+            try:
+                trans.app.security_agent.set_user_group_and_role_associations(
+                    user, group_ids=group_ids, role_ids=role_ids
+                )
+                return {
+                    "message": f"User '{user.email}' has been updated with {len(role_ids)} associated roles and {len(group_ids)} associated groups (private roles are not displayed)."
+                }
+            except RequestParameterInvalidException:
                 return self.message_exception(trans, "One or more invalid role/group id has been provided.")
 
-            # make sure the user is not dis-associating himself from his private role
-            private_role = trans.app.security_agent.get_private_user_role(user)
-            if private_role not in in_roles:
-                in_roles.append(private_role)
-
-            trans.app.security_agent.set_entity_user_associations(users=[user], roles=in_roles, groups=in_groups)
-            trans.sa_session.refresh(user)
-            return {
-                "message": f"User '{user.email}' has been updated with {len(in_roles) - 1} associated roles and {len(in_groups)} associated groups (private roles are not displayed)."
-            }
-
 
 # ---- Utility methods -------------------------------------------------------
 

lib/galaxy_test/api/test_groups.py

@@ -107,7 +107,9 @@ class TestGroupsApi(ApiTestCase):
             another_user_id = self.dataset_populator.user_id()
             another_role_id = self.dataset_populator.user_private_role_id()
         assert another_user_id is not None
-        update_response = self._put(f"groups/{group_id}", data={"user_ids": [another_user_id]}, admin=True, json=True)
+        update_response = self._put(
+            f"groups/{group_id}", data={"user_ids": [user_id, another_user_id]}, admin=True, json=True
+        )
         self._assert_status_code_is_ok(update_response)
 
         # Check if the user was added
@@ -119,7 +121,9 @@ class TestGroupsApi(ApiTestCase):
         )
 
         # Add another role to the group
-        update_response = self._put(f"groups/{group_id}", data={"role_ids": [another_role_id]}, admin=True, json=True)
+        update_response = self._put(
+            f"groups/{group_id}", data={"role_ids": [user_private_role_id, another_role_id]}, admin=True, json=True
+        )
         self._assert_status_code_is_ok(update_response)
 
         # Check if the role was added

test/unit/app/managers/test_NotificationManager.py

@@ -524,8 +524,9 @@ class TestNotificationRecipientResolver(NotificationsBaseTestCase):
         sa_session = self.trans.sa_session
         group = Group(name=name)
         sa_session.add(group)
-        self.trans.app.security_agent.set_entity_group_associations(groups=[group], roles=roles, users=users)
-        sa_session.flush()
+        user_ids = [user.id for user in users]
+        role_ids = [role.id for role in roles]
+        self.trans.app.security_agent.set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
         return group
 
     def _create_test_role(self, name: str, users: List[User], groups: List[Group]):

test/unit/data/model/conftest.py

@@ -119,6 +119,19 @@ def make_default_history_permissions(session, make_history, make_role):
     return f
 
 
+@pytest.fixture
+def make_default_user_permissions(session, make_user, make_role):
+    def f(**kwd):
+        kwd["user"] = kwd.get("user") or make_user()
+        kwd["action"] = kwd.get("action") or random_str()
+        kwd["role"] = kwd.get("role") or make_role()
+        model = m.DefaultUserPermissions(**kwd)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
 @pytest.fixture
 def make_event(session):
     def f(**kwd):
@@ -151,6 +164,26 @@ def make_galaxy_session_to_history_association(session, make_history, make_galax
     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_group_role_association(session):
+    def f(group, role):
+        model = m.GroupRoleAssociation(group, role)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
 @pytest.fixture
 def make_hda(session, make_history):
     def f(**kwd):
@@ -397,6 +430,16 @@ def make_user_item_rating_association(session):
     return f
 
 
+@pytest.fixture
+def make_user_group_association(session):
+    def f(user, group):
+        model = m.UserGroupAssociation(user, group)
+        write_to_db(session, model)
+        return model
+
+    return f
+
+
 @pytest.fixture
 def make_user_role_association(session):
     def f(user, role):

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

@@ -6,8 +6,8 @@ from collections import (
 MockTransaction = namedtuple("MockTransaction", "user")
 
 
-def verify_items(items, expected_items):
+def have_same_elements(items, expected_items):
     """
     Assert that items and expected_items contain the same elements.
     """
-    assert Counter(items) == Counter(expected_items)
+    return Counter(items) == Counter(expected_items)

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

@@ -8,6 +8,7 @@ from sqlalchemy import (
     create_engine,
     text,
 )
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from galaxy import model as m
@@ -35,7 +36,11 @@ def engine(db_url: str) -> "Engine":
 
 @pytest.fixture
 def session(engine: "Engine") -> Session:
-    return Session(engine)
+    session = Session(engine)
+    # For sqlite, we need to explicitly enale foreign key constraints.
+    if engine.name == "sqlite":
+        session.execute(text("PRAGMA foreign_keys = ON;"))
+    return session
 
 
 @pytest.fixture(autouse=True, scope="module")
@@ -58,12 +63,35 @@ def init_datatypes() -> None:
 
 
 @pytest.fixture(autouse=True)
-def clear_database(engine: "Engine") -> "Generator":
+def clear_database(engine: "Engine", session) -> "Generator":
     """Delete all rows from all tables. Called after each test."""
     yield
-    with engine.begin() as conn:
-        for table in m.mapper_registry.metadata.tables:
-            # Unless db is sqlite, disable foreign key constraints to delete out of order
-            if engine.name != "sqlite":
-                conn.execute(text(f"ALTER TABLE {table} DISABLE TRIGGER ALL"))
-            conn.execute(text(f"DELETE FROM {table}"))
+
+    # If a test left an open transaction, rollback to prevent database locking.
+    if session.in_transaction():
+        session.rollback()
+
+    with engine.connect() as conn:
+        if engine.name == "sqlite":
+            conn.execute(text("PRAGMA foreign_keys = OFF;"))
+            for table in m.mapper_registry.metadata.tables:
+                conn.execute(text(f"DELETE FROM {table}"))
+        else:
+            # For postgres, we can disable foreign key constraints with this statement:
+            #   conn.execute(text(f"ALTER TABLE {table} DISABLE TRIGGER ALL"))
+            # However, unless running as superuser, this will raise an error when trying
+            #   to disable a system trigger. Disabling USER triggers instead of ALL
+            #   won't work because the USER option excludes foreign key constraints.
+            # The following is an alternative: we do multiple passes until all tables have been cleared:
+            to_delete = list(m.mapper_registry.metadata.tables)
+            failed = []
+            while to_delete:
+                for table in to_delete:
+                    try:
+                        conn.execute(text(f"DELETE FROM {table}"))
+                    except IntegrityError:
+                        failed.append(table)
+                        conn.rollback()
+                to_delete, failed = failed, []
+
+        conn.commit()

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

@@ -5,7 +5,7 @@ from galaxy.model.db.library import (
     get_library_ids,
     get_library_permissions_by_role,
 )
-from . import verify_items
+from . import have_same_elements
 
 
 def test_get_library_ids(session, make_library, make_library_permissions):
@@ -18,7 +18,7 @@ def test_get_library_ids(session, make_library, make_library_permissions):
 
     ids = get_library_ids(session, "b").all()
     expected = [l2.id, l3.id]
-    verify_items(ids, expected)
+    have_same_elements(ids, expected)
 
 
 def test_get_library_permissions_by_role(session, make_role, make_library_permissions):
@@ -31,7 +31,7 @@ def test_get_library_permissions_by_role(session, make_role, make_library_permis
 
     lp_roles = [lp.role for lp in lps]
     expected = [r1, r2]
-    verify_items(lp_roles, expected)
+    have_same_elements(lp_roles, expected)
 
 
 def test_get_libraries_for_admins(session, make_library):
@@ -44,14 +44,14 @@ def test_get_libraries_for_admins(session, make_library):
 
     libs_deleted = get_libraries_for_admins(session, True).all()
     expected = [libs[0], libs[1]]
-    verify_items(libs_deleted, expected)
+    have_same_elements(libs_deleted, expected)
 
     libs_not_deleted = get_libraries_for_admins(session, False).all()
     expected = [libs[2], libs[3], libs[4]]
-    verify_items(libs_not_deleted, expected)
+    have_same_elements(libs_not_deleted, expected)
 
     libs_all = get_libraries_for_admins(session, None).all()
-    verify_items(libs_all, libs)
+    have_same_elements(libs_all, libs)
 
 
 def test_get_libraries_for_admins__ordering(session, make_library):
@@ -75,7 +75,7 @@ def test_get_libraries_for_non_admins(session, make_library):
     # Expected:  l1 (not deleted, not restricted), l2 (not deleted, restricted but accessible)
     # Not returned: l3 (not deleted but restricted), l4 (deleted)
     expected = [l1, l2]
-    verify_items(allowed, expected)
+    have_same_elements(allowed, expected)
 
 
 def test_get_libraries_for_admins_non_admins__ordering(session, make_library):

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

@@ -4,7 +4,7 @@ from galaxy.model.db.role import (
     get_private_user_role,
     get_roles_by_ids,
 )
-from . import verify_items
+from . import have_same_elements
 
 
 def test_get_npns_roles(session, make_role):
@@ -18,7 +18,7 @@ def test_get_npns_roles(session, make_role):
     # Expected: r4, r5
     # Not returned: r1: deleted, r2: private, r3: sharing
     expected = [r4, r5]
-    verify_items(roles, expected)
+    have_same_elements(roles, expected)
 
 
 def test_get_private_user_role(session, make_user, make_role, make_user_role_association):
@@ -41,4 +41,4 @@ def test_get_roles_by_ids(session, make_role):
 
     roles2 = get_roles_by_ids(session, ids)
     expected = [r1, r2, r3]
-    verify_items(roles2, expected)
+    have_same_elements(roles2, expected)

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

@@ -0,0 +1,941 @@
+import pytest
+
+from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.model import (
+    Group,
+    Role,
+    User,
+)
+from galaxy.model.security import GalaxyRBACAgent
+from . import have_same_elements
+
+
+@pytest.fixture
+def make_user_and_role(session, make_user, make_role, make_user_role_association):
+    """
+    Each user created in Galaxy is assumed to have a private role, such that role.name == user.email.
+    Since we are testing user/group/role associations here, to ensure the correct state of the test database,
+    we need to ensure that a user is never created without a corresponding private role.
+    Therefore, we use this fixture instead of make_user (which only creates a user).
+    """
+
+    def f(**kwd):
+        user = make_user()
+        private_role = make_role(name=user.email, type=Role.types.PRIVATE)
+        make_user_role_association(user, private_role)
+        return user, private_role
+
+    return f
+
+
+def test_private_user_role_assoc_not_affected_by_setting_user_roles(session, make_user_and_role):
+    # Create user with a private role
+    user, private_role = make_user_and_role()
+    assert user.email == private_role.name
+    verify_user_associations(user, [], [private_role])  # the only existing association is with the private role
+
+    # Update users's email so it's no longer the same as the private role's name.
+    user.email = user.email + "updated"
+    session.add(user)
+    session.commit()
+    assert user.email != private_role.name
+
+    # Delete user roles
+    GalaxyRBACAgent(session).set_user_group_and_role_associations(user, role_ids=[])
+    # association with private role is preserved
+    verify_user_associations(user, [], [private_role])
+
+
+def test_private_user_role_assoc_not_affected_by_setting_role_users(session, make_user_and_role):
+    # Create user with a private role
+    user, private_role = make_user_and_role()
+    assert user.email == private_role.name
+    verify_user_associations(user, [], [private_role])  # the only existing association is with the private role
+
+    # Update users's email
+    user.email = user.email + "updated"
+    session.add(user)
+    session.commit()
+    assert user.email != private_role.name
+
+    # Update role users
+    GalaxyRBACAgent(session).set_role_user_and_group_associations(private_role, user_ids=[])
+    # association of private role with user is preserved
+    verify_role_associations(private_role, [user], [])
+
+
+def test_cannot_assign_private_roles(session, make_user_and_role, make_role):
+    user, private_role1 = make_user_and_role()
+    _, private_role2 = make_user_and_role()
+    new_role = make_role()
+    verify_user_associations(user, [], [private_role1])  # the only existing association is with the private role
+
+    # Try to assign 2 more roles: regular role + another private role
+    GalaxyRBACAgent(session).set_user_group_and_role_associations(user, role_ids=[new_role.id, private_role2.id])
+    # Only regular role has been added: other private role ignored; original private role still assigned
+    verify_user_associations(user, [], [private_role1, new_role])
+
+
+class TestSetGroupUserAndRoleAssociations:
+
+    def test_add_associations_to_existing_group(self, session, make_user_and_role, make_role, make_group):
+        """
+        State: group exists in database, has no user and role associations.
+        Action: add new associations.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # users and roles for creating associations
+        users_to_add = [users[0], users[2], users[4]]
+        user_ids = [u.id for u in users_to_add]
+        roles_to_add = [roles[1], roles[3]]
+        role_ids = [r.id for r in roles_to_add]
+
+        # verify no preexisting associations
+        verify_group_associations(group, [], [])
+
+        # set associations
+        GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+        # verify new associations
+        verify_group_associations(group, users_to_add, roles_to_add)
+
+    def test_add_associations_to_new_group(self, session, make_user_and_role, make_role):
+        """
+        State: group does NOT exist in database, has no user and role associations.
+        Action: add new associations.
+        """
+        group = Group()
+        session.add(group)
+        assert group.id is None  # group does not exist in database
+        users = [make_user_and_role()[0] for _ in range(5)]  # type: ignore[unreachable]
+        roles = [make_role() for _ in range(5)]
+
+        # users and roles for creating associations
+        users_to_add = [users[0], users[2], users[4]]
+        user_ids = [u.id for u in users_to_add]
+        roles_to_add = [roles[1], roles[3]]
+        role_ids = [r.id for r in roles_to_add]
+
+        # set associations
+        GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+        # verify new associations
+        verify_group_associations(group, users_to_add, roles_to_add)
+
+    def test_update_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_group_role_association,
+    ):
+        """
+        State: group exists in database AND has user and role associations.
+        Action: update associations (add some/drop some).
+        Expect: old associations are REPLACED by new associations.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[0], users[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for user in users_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_group_role_association(group, role)
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+        # users and roles for creating new associations
+        new_users_to_add = [users[0], users[1], users[3]]
+        user_ids = [u.id for u in new_users_to_add]
+        new_roles_to_add = [roles[2]]
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        # set associations
+        GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+        # verify new associations
+        verify_group_associations(group, new_users_to_add, new_roles_to_add)
+
+    def test_drop_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_group_role_association,
+    ):
+        """
+        State: group exists in database AND has user and role associations.
+        Action: drop all associations.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[0], users[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for user in users_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_group_role_association(group, role)
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+        # drop associations
+        GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=[], role_ids=[])
+
+        # verify associations dropped
+        verify_group_associations(group, [], [])
+
+    def test_invalid_user(self, session, make_user_and_role, make_role, make_group):
+        """
+        State: group exists in database, has no user and role associations.
+        Action: try to add several associations, last one having an invalid user id.
+        Expect: no associations are added, appropriate error is raised.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+
+        # users for creating associations
+        user_ids = [users[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(group.users) == 0
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids)
+
+        # verify no change
+        assert len(group.users) == 0
+
+    def test_invalid_role(self, session, make_role, make_group):
+        """
+        state: group exists in database, has no user and role associations.
+        action: try to add several associations, last one having an invalid role id.
+        expect: no associations are added, appropriate error is raised.
+        """
+        group = make_group()
+        roles = [make_role() for _ in range(5)]
+
+        # roles for creating associations
+        role_ids = [roles[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(group.roles) == 0
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_group_user_and_role_associations(group, role_ids=role_ids)
+
+        # verify no change
+        assert len(group.roles) == 0
+
+    def test_duplicate_user(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_group_role_association,
+    ):
+        """
+        State: group exists in database and has user and role associations.
+        Action: try update user and role associations including a duplicate user
+        Expect: error raised, no change is made to group users and group roles.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[0], users[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for user in users_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_group_role_association(group, role)
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+        # users and roles for creating new associations
+        new_users_to_add = users + [users[0]]  # include a duplicate user
+        user_ids = [u.id for u in new_users_to_add]
+
+        new_roles_to_add = roles  # NO duplice roles
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+        # verify associations not updated
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+    def test_duplicate_role(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_group_role_association,
+    ):
+        """
+        State: group exists in database and has user and role associations.
+        Action: try update user and role associations including a duplicate role
+        Expect: error raised, no change is made to group users and group roles.
+        """
+        group = make_group()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[0], users[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for user in users_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_group_role_association(group, role)
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+        # users and roles for creating new associations
+        new_users_to_add = users  # NO duplicate users
+        user_ids = [u.id for u in new_users_to_add]
+
+        new_roles_to_add = roles + [roles[0]]  # include a duplicate role
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_group_user_and_role_associations(group, user_ids=user_ids, role_ids=role_ids)
+
+        # verify associations not updated
+        verify_group_associations(group, users_to_load, roles_to_load)
+
+
+class TestSetUserGroupAndRoleAssociations:
+    """
+    Note: a user should always have a private role which is not affected
+    by modifying a user's group associations or role associations.
+    """
+
+    def test_add_associations_to_existing_user(self, session, make_user_and_role, make_role, make_group):
+        """
+        State: user exists in database, has no group and only one private role association.
+        Action: add new associations.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # groups and roles for creating associations
+        groups_to_add = [groups[0], groups[2], groups[4]]
+        group_ids = [g.id for g in groups_to_add]
+        roles_to_add = [roles[1], roles[3]]
+        role_ids = [r.id for r in roles_to_add]
+
+        # verify preexisting associations
+        verify_user_associations(user, [], [private_role])
+
+        # set associations
+        GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids, role_ids=role_ids)
+
+        # verify new associations
+        verify_user_associations(user, groups_to_add, roles_to_add + [private_role])
+
+    def test_add_associations_to_new_user(self, session, make_role, make_group):
+        """
+        State: user does NOT exist in database, has no group and role associations.
+        Action: add new associations.
+        """
+        user = User(email="foo@foo.com", password="password")
+        # We are not creating a private role and a user-role association with that role because that would result in
+        # adding the user to the database before calling the method under test, whereas the test is intended to verify
+        # correct processing of a user that has NOT been saved to the database.
+
+        session.add(user)
+        assert user.id is None  # user does not exist in database
+        groups = [make_group() for _ in range(5)]  # type: ignore[unreachable]
+        roles = [make_role() for _ in range(5)]
+
+        # groups and roles for creating associations
+        groups_to_add = [groups[0], groups[2], groups[4]]
+        group_ids = [g.id for g in groups_to_add]
+        roles_to_add = [roles[1], roles[3]]
+        role_ids = [r.id for r in roles_to_add]
+
+        # set associations
+        GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids, role_ids=role_ids)
+
+        # verify new associations
+        verify_user_associations(user, groups_to_add, roles_to_add)
+
+    def test_update_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_user_role_association,
+    ):
+        """
+        State: user exists in database AND has group and role associations.
+        Action: update associations (add some/drop some).
+        Expect: old associations are REPLACED by new associations.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        groups_to_load = [groups[0], groups[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for group in groups_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_user_role_association(user, role)
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+        # groups and roles for creating new associations
+        new_groups_to_add = [groups[0], groups[1], groups[3]]
+        group_ids = [g.id for g in new_groups_to_add]
+        new_roles_to_add = [roles[2]]
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        # set associations
+        GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids, role_ids=role_ids)
+        # verify new associations
+        verify_user_associations(user, new_groups_to_add, new_roles_to_add + [private_role])
+
+    def test_drop_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_user_role_association,
+    ):
+        """
+        State: user exists in database AND has group and role associations.
+        Action: drop all associations.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        groups_to_load = [groups[0], groups[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for group in groups_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_user_role_association(user, role)
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+        # drop associations
+        GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=[], role_ids=[])
+
+        # verify associations dropped
+        verify_user_associations(user, [], [private_role])
+
+    def test_invalid_group(self, session, make_user_and_role, make_group):
+        """
+        State: user exists in database, has no group and only one private role association.
+        Action: try to add several associations, last one having an invalid group id.
+        Expect: no associations are added, appropriate error is raised.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+
+        # groups for creating associations
+        group_ids = [groups[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(user.groups) == 0
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids)
+
+        # verify no change
+        assert len(user.groups) == 0
+
+    def test_invalid_role(self, session, make_user_and_role, make_role):
+        """
+        State: user exists in database, has no group and only one private role association.
+        action: try to add several associations, last one having an invalid role id.
+        expect: no associations are added, appropriate error is raised.
+        """
+        user, private_role = make_user_and_role()
+        roles = [make_role() for _ in range(5)]
+
+        # roles for creating associations
+        role_ids = [roles[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(user.roles) == 1  # one is the private role association
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_user_group_and_role_associations(user, role_ids=role_ids)
+
+        # verify no change
+        assert len(user.roles) == 1  # one is the private role association
+
+    def test_duplicate_group(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_user_role_association,
+    ):
+        """
+        State: user exists in database and has group and role associations.
+        Action: try update group and role associations including a duplicate group
+        Expect: error raised, no change is made to user groups and user roles.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        groups_to_load = [groups[0], groups[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for group in groups_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_user_role_association(user, role)
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+        # groups and roles for creating new associations
+        new_groups_to_add = groups + [groups[0]]  # include a duplicate group
+        group_ids = [g.id for g in new_groups_to_add]
+
+        new_roles_to_add = roles  # NO duplicate roles
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids, role_ids=role_ids)
+
+        # verify associations not updated
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+    def test_duplicate_role(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_group_association,
+        make_user_role_association,
+    ):
+        """
+        State: user exists in database and has group and role associations.
+        Action: try update group and role associations including a duplicate role
+        Expect: error raised, no change is made to user groups and user roles.
+        """
+        user, private_role = make_user_and_role()
+        groups = [make_group() for _ in range(5)]
+        roles = [make_role() for _ in range(5)]
+
+        # load and verify existing associations
+        groups_to_load = [groups[0], groups[2]]
+        roles_to_load = [roles[1], roles[3]]
+        for group in groups_to_load:
+            make_user_group_association(user, group)
+        for role in roles_to_load:
+            make_user_role_association(user, role)
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+        # groups and roles for creating new associations
+        new_groups_to_add = groups  # NO duplicate groups
+        group_ids = [g.id for g in new_groups_to_add]
+
+        new_roles_to_add = roles + [roles[0]]  # include a duplicate role
+        role_ids = [r.id for r in new_roles_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+        assert not have_same_elements(roles_to_load, new_roles_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_user_group_and_role_associations(user, group_ids=group_ids, role_ids=role_ids)
+
+        # verify associations not updated
+        verify_user_associations(user, groups_to_load, roles_to_load + [private_role])
+
+
+class TestSetRoleUserAndGroupAssociations:
+    """
+    Note: a user should always have a private role which is not affected
+    by modifying a user's group associations or role associations.
+    """
+
+    def test_add_associations_to_existing_role(self, session, make_user_and_role, make_role, make_group):
+        """
+        State: role exists in database, has no group and no user associations.
+        Action: add new associations.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        groups = [make_group() for _ in range(5)]
+
+        # users and groups for creating associations
+        users_to_add = [users[0], users[2], users[4]]
+        user_ids = [u.id for u in users_to_add]
+        groups_to_add = [groups[0], groups[2], groups[4]]
+        group_ids = [g.id for g in groups_to_add]
+
+        # verify preexisting associations
+        verify_role_associations(role, [], [])
+
+        # set associations
+        GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids, group_ids=group_ids)
+
+        # verify new associations
+        verify_role_associations(role, users_to_add, groups_to_add)
+
+    def test_add_associations_to_new_role(self, session, make_user_and_role, make_group):
+        """
+        State: user does NOT exist in database, has no group and role associations.
+        Action: add new associations.
+        """
+        role = Role()
+        session.add(role)
+        assert role.id is None  # role does not exist in database
+        users = [make_user_and_role()[0] for _ in range(5)]  # type: ignore[unreachable]
+        groups = [make_group() for _ in range(5)]
+
+        # users and groups for creating associations
+        users_to_add = [users[0], users[2], users[4]]
+        user_ids = [u.id for u in users_to_add]
+        groups_to_add = [groups[0], groups[2], groups[4]]
+        group_ids = [g.id for g in groups_to_add]
+
+        # set associations
+        GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids, group_ids=group_ids)
+
+        # verify new associations
+        verify_role_associations(role, users_to_add, groups_to_add)
+
+    def test_update_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_user_role_association,
+        make_group_role_association,
+    ):
+        """
+        State: role exists in database AND has user and group associations.
+        Action: update associations (add some/drop some).
+        Expect: old associations are REPLACED by new associations.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        groups = [make_group() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[1], users[3]]
+        groups_to_load = [groups[0], groups[2]]
+        for user in users_to_load:
+            make_user_role_association(user, role)
+        for group in groups_to_load:
+            make_group_role_association(group, role)
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+        # users and groups for creating new associations
+        new_users_to_add = [users[0], users[2], users[4]]
+        user_ids = [u.id for u in new_users_to_add]
+        new_groups_to_add = [groups[0], groups[2], groups[4]]
+        group_ids = [g.id for g in new_groups_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+
+        # set associations
+        GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids, group_ids=group_ids)
+        # verify new associations
+        verify_role_associations(role, new_users_to_add, new_groups_to_add)
+
+    def test_drop_associations(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_group_role_association,
+        make_user_role_association,
+    ):
+        """
+        State: role exists in database AND has user and group associations.
+        Action: drop all associations.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        groups = [make_group() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[1], users[3]]
+        groups_to_load = [groups[0], groups[2]]
+        for user in users_to_load:
+            make_user_role_association(user, role)
+        for group in groups_to_load:
+            make_group_role_association(group, role)
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+        # drop associations
+        GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=[], group_ids=[])
+
+        # verify associations dropped
+        verify_role_associations(role, [], [])
+
+    def test_invalid_user(self, session, make_role, make_user_and_role):
+        """
+        State: role exists in database, has no user and group eassociations.
+        action: try to add several associations, last one having an invalid user id.
+        expect: no associations are added, appropriate error is raised.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+
+        # users for creating associations
+        user_ids = [users[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(role.users) == 0
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids)
+
+        # verify no change
+        assert len(role.users) == 0
+
+    def test_invalid_group(self, session, make_role, make_group):
+        """
+        State: role exists in database, has no user and group eassociations.
+        Action: try to add several associations, last one having an invalid group id.
+        Expect: no associations are added, appropriate error is raised.
+        """
+        role = make_role()
+        groups = [make_group() for _ in range(5)]
+
+        # groups for creating associations
+        group_ids = [groups[0].id, -1]  # first is valid, second is invalid
+
+        # verify no preexisting associations
+        assert len(role.groups) == 0
+
+        # try to set associations
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_role_user_and_group_associations(role, group_ids=group_ids)
+
+        # verify no change
+        assert len(role.groups) == 0
+
+    def test_duplicate_user(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_group_role_association,
+        make_user_role_association,
+    ):
+        """
+        State: role exists in database and has group and user associations.
+        Action: try update group and user associations including a duplicate user
+        Expect: error raised, no change is made to role groups and role users.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        groups = [make_group() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[1], users[3]]
+        groups_to_load = [groups[0], groups[2]]
+        for user in users_to_load:
+            make_user_role_association(user, role)
+        for group in groups_to_load:
+            make_group_role_association(group, role)
+
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+        # users and groups for creating new associations
+        new_users_to_add = users + [users[0]]  # include a duplicate user
+        user_ids = [u.id for u in new_users_to_add]
+
+        new_groups_to_add = groups  # NO duplicate groups
+        group_ids = [g.id for g in new_groups_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids, group_ids=group_ids)
+
+        # verify associations not updated
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+    def test_duplicate_group(
+        self,
+        session,
+        make_user_and_role,
+        make_role,
+        make_group,
+        make_group_role_association,
+        make_user_role_association,
+    ):
+        """
+        State: role exists in database and has group and user associations.
+        Action: try update group and user associations including a duplicate group
+        Expect: error raised, no change is made to role groups and role users.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        groups = [make_group() for _ in range(5)]
+
+        # load and verify existing associations
+        users_to_load = [users[1], users[3]]
+        groups_to_load = [groups[0], groups[2]]
+        for user in users_to_load:
+            make_user_role_association(user, role)
+        for group in groups_to_load:
+            make_group_role_association(group, role)
+
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+        # users and groups for creating new associations
+        new_users_to_add = users  # NO duplicate users
+        user_ids = [u.id for u in new_users_to_add]
+
+        new_groups_to_add = groups + [groups[0]]  # include a duplicate group
+        group_ids = [g.id for g in new_groups_to_add]
+
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+        assert not have_same_elements(groups_to_load, new_groups_to_add)
+
+        with pytest.raises(RequestParameterInvalidException):
+            GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids, group_ids=group_ids)
+
+        # verify associations not updated
+        verify_role_associations(role, users_to_load, groups_to_load)
+
+    def test_delete_default_user_permissions_and_default_history_permissions(
+        self,
+        session,
+        make_role,
+        make_user_and_role,
+        make_user_role_association,
+        make_default_user_permissions,
+        make_default_history_permissions,
+        make_history,
+    ):
+        """
+        When setting role users, we check check previously associated users to:
+        - delete DefaultUserPermissions for users that are being removed from this role;
+        - delete DefaultHistoryPermissions for histories associated with users that are being removed from this role.
+        """
+        role = make_role()
+        users = [make_user_and_role()[0] for _ in range(5)]
+        # load and verify existing associations
+        user1, user2 = users[0], users[1]
+        users_to_load = [user1, user2]
+        for user in users_to_load:
+            make_user_role_association(user, role)
+        verify_role_associations(role, users_to_load, [])
+
+        # users and groups for creating new associations
+        new_users_to_add = [users[1], users[2]]  # REMOVE users[0], LEAVE users[1], ADD users[2]
+        user_ids = [u.id for u in new_users_to_add]
+        # sanity check: ensure we are trying to change existing associations
+        assert not have_same_elements(users_to_load, new_users_to_add)
+
+        # load default user permissions
+        dup1 = make_default_user_permissions(user=user1, role=role)
+        dup2 = make_default_user_permissions(user=user2, role=role)
+        assert have_same_elements(user1.default_permissions, [dup1])
+        assert have_same_elements(user2.default_permissions, [dup2])
+
+        # load and verify default history permissions for users associated with this role
+        history1, history2 = make_history(user=user1), make_history(user=user1)  # 2 histories for user 1
+        history3 = make_history(user=user2)  # 1 history for user 2
+        dhp1 = make_default_history_permissions(history=history1, role=role)
+        dhp2 = make_default_history_permissions(history=history2, role=role)
+        dhp3 = make_default_history_permissions(history=history3, role=role)
+        assert have_same_elements(history1.default_permissions, [dhp1])
+        assert have_same_elements(history2.default_permissions, [dhp2])
+        assert have_same_elements(history3.default_permissions, [dhp3])
+
+        # now update role users
+        GalaxyRBACAgent(session).set_role_user_and_group_associations(role, user_ids=user_ids)
+
+        # verify user role associations
+        verify_role_associations(role, new_users_to_add, [])
+
+        # verify default user permissions
+        assert have_same_elements(user1.default_permissions, [])  # user1 was removed from role
+        assert have_same_elements(user2.default_permissions, [dup2])  # user2 was NOT removed from role
+
+        # verify default history permissions
+        assert have_same_elements(history1.default_permissions, [])
+        assert have_same_elements(history2.default_permissions, [])
+        assert have_same_elements(history3.default_permissions, [dhp3])
+
+
+def verify_group_associations(group, expected_users, expected_roles):
+    new_group_users = [assoc.user for assoc in group.users]
+    new_group_roles = [assoc.role for assoc in group.roles]
+    assert have_same_elements(new_group_users, expected_users)
+    assert have_same_elements(new_group_roles, expected_roles)
+
+
+def verify_user_associations(user, expected_groups, expected_roles):
+    new_user_groups = [assoc.group for assoc in user.groups]
+    new_user_roles = [assoc.role for assoc in user.roles]
+    assert have_same_elements(new_user_groups, expected_groups)
+    assert have_same_elements(new_user_roles, expected_roles)
+
+
+def verify_role_associations(role, expected_users, expected_groups):
+    new_role_users = [assoc.user for assoc in role.users]
+    new_role_groups = [assoc.group for assoc in role.groups]
+    assert have_same_elements(new_role_users, expected_users)
+    assert have_same_elements(new_role_groups, expected_groups)

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

@@ -7,7 +7,7 @@ from galaxy.model.db.user import (
     get_users_by_ids,
     get_users_for_index,
 )
-from . import verify_items
+from . import have_same_elements
 
 
 @pytest.fixture
@@ -42,7 +42,7 @@ def test_get_users_by_ids(session, make_random_users):
 
     users2 = get_users_by_ids(session, ids)
     expected = [u1, u2, u3]
-    verify_items(users2, expected)
+    have_same_elements(users2, expected)
 
 
 def test_get_users_for_index(session, make_user):
@@ -54,25 +54,25 @@ def test_get_users_for_index(session, make_user):
     u6 = make_user(email="z", username="i")
 
     users = get_users_for_index(session, False, f_email="a", expose_user_email=True)
-    verify_items(users, [u1])
+    have_same_elements(users, [u1])
     users = get_users_for_index(session, False, f_email="c", is_admin=True)
-    verify_items(users, [u2])
+    have_same_elements(users, [u2])
     users = get_users_for_index(session, False, f_name="f", expose_user_name=True)
-    verify_items(users, [u3])
+    have_same_elements(users, [u3])
     users = get_users_for_index(session, False, f_name="h", is_admin=True)
-    verify_items(users, [u4])
+    have_same_elements(users, [u4])
     users = get_users_for_index(session, False, f_any="i", is_admin=True)
-    verify_items(users, [u5, u6])
+    have_same_elements(users, [u5, u6])
     users = get_users_for_index(session, False, f_any="i", expose_user_email=True, expose_user_name=True)
-    verify_items(users, [u5, u6])
+    have_same_elements(users, [u5, u6])
     users = get_users_for_index(session, False, f_any="i", expose_user_email=True)
-    verify_items(users, [u5])
+    have_same_elements(users, [u5])
     users = get_users_for_index(session, False, f_any="i", expose_user_name=True)
-    verify_items(users, [u6])
+    have_same_elements(users, [u6])
 
     u1.deleted = True
     users = get_users_for_index(session, True)
-    verify_items(users, [u1])
+    have_same_elements(users, [u1])
 
 
 def test_username_is_unique(make_user):

test/unit/data/model/migration_fixes/test_migrations.py

@@ -1,6 +1,12 @@
 import pytest
+from sqlalchemy import select
 
-from galaxy.model import User
+from galaxy.model import (
+    GroupRoleAssociation,
+    User,
+    UserGroupAssociation,
+    UserRoleAssociation,
+)
 from galaxy.model.unittest_utils.migration_scripts_testing_utils import (  # noqa: F401 - contains fixtures we have to import explicitly
     run_command,
     tmp_directory,
@@ -152,3 +158,222 @@ def test_d619fdfa6168(monkeypatch, session, make_user):
     assert u1_fixed.deleted is True
     assert u2_fixed.deleted is True
     assert u3_fixed.deleted is False
+
+
+def test_349dd9d9aac9(monkeypatch, session, make_user, make_role, make_user_role_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 1cf595475b58")
+
+    # Load duplicate records
+    u1, u2 = make_user(), make_user()
+    r1, r2 = make_role(), make_role()
+    make_user_role_association(user=u1, role=r1)
+    make_user_role_association(user=u1, role=r2)
+    make_user_role_association(user=u1, role=r2)  # duplicate
+    make_user_role_association(user=u2, role=r1)
+    make_user_role_association(user=u2, role=r1)  # duplicate
+
+    # Verify duplicates
+    assert len(u1.roles) == 3
+    assert len(u2.roles) == 2
+    all_associations = session.execute(select(UserRoleAssociation)).all()
+    assert len(all_associations) == 5
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 349dd9d9aac9")
+    session.expire_all()
+
+    # Verify clean data
+    assert len(u1.roles) == 2
+    assert len(u2.roles) == 1
+    all_associations = session.execute(select(UserRoleAssociation)).all()
+    assert len(all_associations) == 3
+
+
+def test_56ddf316dbd0(monkeypatch, session, make_user, make_group, make_user_group_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 1fdd615f2cdb")
+
+    # Load duplicate records
+    u1, u2 = make_user(), make_user()
+    g1, g2 = make_group(), make_group()
+    make_user_group_association(user=u1, group=g1)
+    make_user_group_association(user=u1, group=g2)
+    make_user_group_association(user=u1, group=g2)  # duplicate
+    make_user_group_association(user=u2, group=g1)
+    make_user_group_association(user=u2, group=g1)  # duplicate
+
+    # Verify duplicates
+    assert len(u1.groups) == 3
+    assert len(u2.groups) == 2
+    all_associations = session.execute(select(UserGroupAssociation)).all()
+    assert len(all_associations) == 5
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 56ddf316dbd0")
+    session.expire_all()
+
+    # Verify clean data
+    assert len(u1.groups) == 2
+    assert len(u2.groups) == 1
+    all_associations = session.execute(select(UserGroupAssociation)).all()
+    assert len(all_associations) == 3
+
+
+def test_9ef6431f3a4e(monkeypatch, session, make_group, make_role, make_group_role_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 13fe10b8e35b")
+
+    # Load duplicate records
+    g1, g2 = make_group(), make_group()
+    r1, r2 = make_role(), make_role()
+    make_group_role_association(group=g1, role=r1)
+    make_group_role_association(group=g1, role=r2)
+    make_group_role_association(group=g1, role=r2)  # duplicate
+    make_group_role_association(group=g2, role=r1)
+    make_group_role_association(group=g2, role=r1)  # duplicate
+
+    # Verify duplicates
+    assert len(g1.roles) == 3
+    assert len(g2.roles) == 2
+    all_associations = session.execute(select(GroupRoleAssociation)).all()
+    assert len(all_associations) == 5
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 9ef6431f3a4e")
+    session.expire_all()
+
+    # Verify clean data
+    assert len(g1.roles) == 2
+    assert len(g2.roles) == 1
+    all_associations = session.execute(select(GroupRoleAssociation)).all()
+    assert len(all_associations) == 3
+
+
+def test_1fdd615f2cdb(monkeypatch, session, make_user, make_role, make_user_role_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 349dd9d9aac9")
+
+    # Load records w/nulls
+    ura1 = make_user_role_association(user=make_user(), role=make_role())
+    ura2 = make_user_role_association(user=make_user(), role=make_role())
+    ura3 = make_user_role_association(user=make_user(), role=make_role())
+    ura1.user_id = None
+    ura2.role_id = None
+    ura3.user_id = None
+    ura3.role_id = None
+    session.add_all([ura1, ura2, ura3])
+    session.commit()
+
+    # Load record w/o nulls
+    make_user_role_association(user=make_user(), role=make_role())
+
+    # Verify data
+    all_associations = session.execute(select(UserRoleAssociation)).all()
+    assert len(all_associations) == 4
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 1fdd615f2cdb")
+    session.expire_all()
+
+    # Verify clean data
+    all_associations = session.execute(select(UserRoleAssociation)).all()
+    assert len(all_associations) == 1
+
+
+def test_13fe10b8e35b(monkeypatch, session, make_user, make_group, make_user_group_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 56ddf316dbd0")
+
+    # Load records w/nulls
+    uga1 = make_user_group_association(user=make_user(), group=make_group())
+    uga2 = make_user_group_association(user=make_user(), group=make_group())
+    uga3 = make_user_group_association(user=make_user(), group=make_group())
+    uga1.user_id = None
+    uga2.group_id = None
+    uga3.user_id = None
+    uga3.group_id = None
+    session.add_all([uga1, uga2, uga3])
+    session.commit()
+
+    # Load record w/o nulls
+    make_user_group_association(user=make_user(), group=make_group())
+
+    # Verify data
+    all_associations = session.execute(select(UserGroupAssociation)).all()
+    assert len(all_associations) == 4
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 13fe10b8e35b")
+    session.expire_all()
+
+    # Verify clean data
+    all_associations = session.execute(select(UserGroupAssociation)).all()
+    assert len(all_associations) == 1
+
+
+def test_25b092f7938b(monkeypatch, session, make_group, make_role, make_group_role_association):
+    # Initialize db and migration environment
+    dburl = str(session.bind.url)
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+    run_command(f"{COMMAND} init")
+
+    # Load pre-migration state
+    run_command(f"{COMMAND} downgrade 9ef6431f3a4e")
+
+    # Load records w/nulls
+    gra1 = make_group_role_association(group=make_group(), role=make_role())
+    gra2 = make_group_role_association(group=make_group(), role=make_role())
+    gra3 = make_group_role_association(group=make_group(), role=make_role())
+    gra1.group_id = None
+    gra2.role_id = None
+    gra3.group_id = None
+    gra3.role_id = None
+    session.add_all([gra1, gra2, gra3])
+    session.commit()
+
+    # Load record w/o nulls
+    make_group_role_association(group=make_group(), role=make_role())
+
+    # Verify data
+    all_associations = session.execute(select(GroupRoleAssociation)).all()
+    assert len(all_associations) == 4
+
+    # Run migration
+    run_command(f"{COMMAND} upgrade 25b092f7938b")
+    session.expire_all()
+
+    # Verify clean data
+    all_associations = session.execute(select(GroupRoleAssociation)).all()
+    assert len(all_associations) == 1

test/unit/data/test_galaxy_mapping.py

@@ -456,7 +456,7 @@ class TestMappings(BaseModelTestCase):
         assert counts.root["scheduled"] == 1
 
     def test_role_creation(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
 
         def check_private_role(private_role, email):
             assert private_role.type == model.Role.types.PRIVATE
@@ -489,7 +489,7 @@ class TestMappings(BaseModelTestCase):
         check_private_role(role, email)
 
     def test_private_share_role(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
 
         u_from, u_to, u_other = self._three_users("private_share_role")
 
@@ -504,7 +504,7 @@ class TestMappings(BaseModelTestCase):
         assert not security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)
 
     def test_make_dataset_public(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, u_other = self._three_users("make_dataset_public")
 
         h = model.History(name="History for Annotation", user=u_from)
@@ -520,7 +520,7 @@ class TestMappings(BaseModelTestCase):
         assert security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)
 
     def test_set_all_dataset_permissions(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, _, u_other = self._three_users("set_all_perms")
 
         h = model.History(name="History for Annotation", user=u_from)
@@ -541,7 +541,7 @@ class TestMappings(BaseModelTestCase):
         assert not security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)
 
     def test_can_manage_privately_shared_dataset(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, u_other = self._three_users("can_manage_dataset")
 
         h = model.History(name="History for Prevent Sharing", user=u_from)
@@ -556,7 +556,7 @@ class TestMappings(BaseModelTestCase):
         assert not security_agent.can_manage_dataset(u_to.all_roles(), d1.dataset)
 
     def test_can_manage_private_dataset(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, _, u_other = self._three_users("can_manage_dataset_ps")
 
         h = model.History(name="History for Prevent Sharing", user=u_from)
@@ -570,7 +570,7 @@ class TestMappings(BaseModelTestCase):
         assert not security_agent.can_manage_dataset(u_other.all_roles(), d1.dataset)
 
     def test_cannot_make_private_objectstore_dataset_public(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, _ = self._three_users("cannot_make_private_public")
 
         h = self.model.History(name="History for Prevent Sharing", user=u_from)
@@ -587,7 +587,7 @@ class TestMappings(BaseModelTestCase):
         assert galaxy.model.CANNOT_SHARE_PRIVATE_DATASET_MESSAGE in str(exec_info.value)
 
     def test_cannot_make_private_objectstore_dataset_shared(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, _ = self._three_users("cannot_make_private_shared")
 
         h = self.model.History(name="History for Prevent Sharing", user=u_from)
@@ -604,7 +604,7 @@ class TestMappings(BaseModelTestCase):
         assert galaxy.model.CANNOT_SHARE_PRIVATE_DATASET_MESSAGE in str(exec_info.value)
 
     def test_cannot_set_dataset_permisson_on_private(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, _ = self._three_users("cannot_set_permissions_on_private")
 
         h = self.model.History(name="History for Prevent Sharing", user=u_from)
@@ -624,7 +624,7 @@ class TestMappings(BaseModelTestCase):
         assert galaxy.model.CANNOT_SHARE_PRIVATE_DATASET_MESSAGE in str(exec_info.value)
 
     def test_cannot_make_private_dataset_public(self):
-        security_agent = GalaxyRBACAgent(self.model)
+        security_agent = GalaxyRBACAgent(self.model.session)
         u_from, u_to, u_other = self._three_users("cannot_make_private_dataset_public")
 
         h = self.model.History(name="History for Annotation", user=u_from)