Cleanup sharing roles on user purge, refactor, test

1. Update sharing role name removing purged user email
2. If sharing role has only one user association, delete it.
3. Factor out cleanup, add test.

Note: this will not work if user email has been udpated after the sharing
role was created.

lib/galaxy/managers/users.py

@@ -40,6 +40,7 @@
 from galaxy.managers.base import combine_lists
 from galaxy.model import (
     Job,
+    Role,
     User,
     UserAddress,
     UserQuotaUsage,
@@ -214,10 +215,7 @@ def purge(self, user, flush=True):
         # Delete UserGroupAssociations
         for uga in user.groups:
             self.session().delete(uga)
-        # Delete UserRoleAssociations EXCEPT FOR THE PRIVATE ROLE
-        for ura in user.roles:
-            if ura.role_id != private_role.id:
-                self.session().delete(ura)
+        _cleanup_nonprivate_user_roles(self.session(), user, private_role.id)
         # Delete UserAddresses
         for address in user.addresses:
             self.session().delete(address)
@@ -891,3 +889,26 @@ def generate_next_available_username(session, username, model_class=User):
     while session.execute(select(model_class).where(model_class.username == f"{username}-{i}")).first():
         i += 1
     return f"{username}-{i}"
+
+
+def _cleanup_nonprivate_user_roles(session, user, private_role_id):
+    """
+    Delete UserRoleAssociations EXCEPT FOR THE PRIVATE ROLE;
+    Delete sharing roles that are associated with this user only;
+    Remove user email from sharing role names associated with multiple users.
+
+    Note: this method updates the session without flushing or committing.
+    """
+    user_roles = [ura for ura in user.roles if ura.role_id != private_role_id]
+    for user_role_assoc in user_roles:
+        role = user_role_assoc.role
+        if role.type == Role.types.SHARING:
+            if len(role.users) == 1:
+                # This role is associated with this user only, so we can delete it
+                session.delete(role)
+            elif user.email in role.name:
+                # Remove user email from sharing role's name
+                role.name = role.name.replace(user.email, "[USER PURGED]")
+                session.add(role)
+        # Delete user role association
+        session.delete(user_role_assoc)

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

@@ -1,6 +1,12 @@
 import pytest
+from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
+from galaxy.managers.users import _cleanup_nonprivate_user_roles
+from galaxy.model import (
+    Role,
+    UserRoleAssociation,
+)
 from galaxy.model.db.user import (
     get_user_by_email,
     get_user_by_username,
@@ -80,3 +86,78 @@ def test_username_is_unique(make_user):
     make_user(username="a")
     with pytest.raises(IntegrityError):
         make_user(username="a")
+
+
+def test_cleanup_nonprivate_user_roles(session, make_user_and_role, make_role, make_user_role_association):
+    # Create 3 users with private roles
+    user1, role_private1 = make_user_and_role(email="[email protected]")
+    user2, role_private2 = make_user_and_role(email="[email protected]")
+    user3, role_private3 = make_user_and_role(email="[email protected]")
+
+    # Create role_sharing1 and associated it with user1 and user2
+    role_sharing1 = make_role(type=Role.types.SHARING, name="sharing role for [email protected], [email protected]")
+    make_user_role_association(user1, role_sharing1)
+    make_user_role_association(user2, role_sharing1)
+
+    # Create role_sharing2 and associated it with user3
+    role_sharing2 = make_role(type=Role.types.SHARING, name="sharing role for [email protected]")
+    make_user_role_association(user3, role_sharing2)
+
+    # Create another role and associate it with all users
+    role6 = make_role()
+    make_user_role_association(user1, role6)
+    make_user_role_association(user2, role6)
+    make_user_role_association(user3, role6)
+
+    # verify number of all user role associations
+    associations = session.scalars(select(UserRoleAssociation)).all()
+    assert len(associations) == 9  # 3 private, 2 + 1 sharing, 3 with role6
+
+    # verify user1 roles
+    assert len(user1.roles) == 3
+    assert role_private1 in [ura.role for ura in user1.roles]
+    assert role_sharing1 in [ura.role for ura in user1.roles]
+    assert role6 in [ura.role for ura in user1.roles]
+
+    # run cleanup on user user1
+    _cleanup_nonprivate_user_roles(session, user1, role_private1.id)
+    session.commit()  # must commit since method does not commit
+
+    # private role not deleted, associations with role6 and with sharing role deleted, sharing role name updated
+    assert len(user1.roles) == 1
+    assert user1.roles[0].id == role_private1.id
+    assert role_sharing1.name == "sharing role for [USER PURGED], [email protected]"
+
+    # verify user2 roles
+    assert len(user2.roles) == 3
+    assert role_private2 in [ura.role for ura in user2.roles]
+    assert role_sharing1 in [ura.role for ura in user2.roles]
+    assert role6 in [ura.role for ura in user2.roles]
+
+    # run cleanup on user user2
+    _cleanup_nonprivate_user_roles(session, user2, role_private2.id)
+    session.commit()
+
+    # private role not deleted, association with sharing role deleted
+    assert len(user2.roles) == 1
+    assert user2.roles[0].id == role_private2.id
+    # sharing role1 deleted since it has no more users associated with it
+    roles = session.scalars(select(Role)).all()
+    assert len(roles) == 5  # 3 private, role6, sharing2
+    assert role_sharing1.id not in [role.id for role in roles]
+
+    # verify user3 roles
+    assert len(user3.roles) == 3
+    assert role_private3 in [ura.role for ura in user3.roles]
+    assert role_sharing2 in [ura.role for ura in user3.roles]
+    assert role6 in [ura.role for ura in user3.roles]
+
+    # run cleanup on user user3
+    _cleanup_nonprivate_user_roles(session, user3, role_private3.id)
+    session.commit()
+
+    # remaining: 3 private roles + 3 associations, role6
+    roles = session.scalars(select(Role)).all()
+    assert len(roles) == 4
+    associations = session.scalars(select(UserRoleAssociation)).all()
+    assert len(associations) == 3