Sync to 2 different DAB team member roles

galaxy_ng/app/signals/handlers.py

@@ -15,6 +15,7 @@
 from django.db.models import CharField, Value
 from django.db.models.functions import Concat
 from django.contrib.contenttypes.models import ContentType
+from django.contrib.auth.models import Group
 from django.conf import settings
 from rest_framework.exceptions import ValidationError
 from django.apps import apps
@@ -121,6 +122,7 @@ def _update_metadata():
 # ___ DAB RBAC ___
 
 TEAM_MEMBER_ROLE = 'Galaxy Team Member'
+SHARED_TEAM_ROLE = 'Team Member'
 
 
 def create_managed_roles(*args, **kwargs) -> None:
@@ -495,7 +497,7 @@ def copy_dab_user_role_assignment(sender, instance, created, **kwargs):
     if rbac_signal_in_progress():
         return
     with dab_rbac_signals():
-        if instance.role_definition.name == TEAM_MEMBER_ROLE and \
+        if instance.role_definition.name in (TEAM_MEMBER_ROLE, SHARED_TEAM_ROLE) and \
                 isinstance(instance, RoleUserAssignment):
             instance.content_object.group.user_set.add(instance.user)
             return
@@ -508,12 +510,18 @@ def delete_dab_user_role_assignment(sender, instance, **kwargs):
     if rbac_signal_in_progress():
         return
     with dab_rbac_signals():
-        if instance.role_definition.name == TEAM_MEMBER_ROLE and \
+        if instance.role_definition.name in (TEAM_MEMBER_ROLE, SHARED_TEAM_ROLE) and \
                 isinstance(instance, RoleUserAssignment):
+            if instance.role_definition.name == TEAM_MEMBER_ROLE:
+                dup_name = SHARED_TEAM_ROLE
+            else:
+                dup_name = TEAM_MEMBER_ROLE
             # If the assignment does not have a content_object then it may be a global group role
             # this type of role is not compatible with DAB RBAC and what we do is still TBD
             if instance.content_object:
-                instance.content_object.group.user_set.remove(instance.user)
+                # Only remove assignment if other role does not grant this, otherwise leave it
+                if not RoleUserAssignment.objects.filter(role_definition__name=dup_name, user=instance.user, object_id=instance.object_id).exists():
+                    instance.content_object.group.user_set.remove(instance.user)
                 return
         _unapply_dab_assignment(instance)
 
@@ -545,39 +553,41 @@ def copy_dab_group_to_role(instance, action, model, pk_set, reverse, **kwargs):
         return
 
     member_rd = RoleDefinition.objects.get(name=TEAM_MEMBER_ROLE)
+    shared_member_rd = RoleDefinition.objects.get(name=SHARED_TEAM_ROLE)
     if reverse:
-        # NOTE: for we might prefer to use pk_set
-        # but it appears to be incorrectly empty on the reverse signal,
-        # the models itself on post_ actions seems good so we use that
-        team = Team.objects.get(group_id=instance.pk)
+        groups = [instance]
+    else:
+        if action == 'post_clear':
+            qs = RoleUserAssignment.objects.filter(role_definition=member_rd, user=instance)
+            groups = [assignment.content_object for assignment in qs]
+        else:
+            groups = Group.objects.filter(pk__in=pk_set)
+
+    for group in groups:
+        team = Team.objects.get(group_id=group.pk)
         current_dab_members = set(
             assignment.user for assignment in RoleUserAssignment.objects.filter(
                 role_definition=member_rd, object_id=team.pk
             )
         )
-        desired_members = set(instance.user_set.all())
+        current_dab_shared_members = set(
+            assignment.user for assignment in RoleUserAssignment.objects.filter(
+                role_definition=member_rd, object_id=team.pk
+            )
+        )
+        current_pulp_members = set(group.user_set.all())
+        not_allowed = current_dab_shared_members - current_pulp_members
+        if not_allowed:
+            # TODO: this looks non-viable with current JWT data for tests, maybe make into a log?
+            raise ValidationError(f'Can not remove team members from resource provider: {", ".join([u.username for u in not_allowed])}')
+        desired_members = current_pulp_members - current_dab_shared_members
         users_to_add = desired_members - current_dab_members
         users_to_remove = current_dab_members - desired_members
         with dab_rbac_signals():
             for user in users_to_add:
                 member_rd.give_permission(user, team)
             for user in users_to_remove:
                 member_rd.remove_permission(user, team)
-        return
-
-    with dab_rbac_signals():
-        if action == 'post_add':
-            for group_id in pk_set:
-                team = Team.objects.get(group_id=group_id)
-                member_rd.give_permission(instance, team)
-        elif action == 'post_remove':
-            for group_id in pk_set:
-                team = Team.objects.get(group_id=group_id)
-                member_rd.remove_permission(instance, team)
-        elif action == 'post_clear':
-            qs = RoleUserAssignment.objects.filter(role_definition=member_rd, user=instance)
-            for assignment in qs:
-                member_rd.remove_permission(instance, assignment.content_object)
 
 
 m2m_changed.connect(copy_dab_group_to_role, sender=User.groups.through)

galaxy_ng/tests/integration/dab/test_dab_rbac_contract.py

@@ -501,7 +501,7 @@ def _rf(user_id, group_id, expected=True):
             assert group["name"] not in [g["name"] for g in user["groups"]]
 
         assignment_r = gc.get(
-            f"_ui/v2/role_user_assignments/?user={user['id']}&content_type__model=team"
+            f"_ui/v2/role_user_assignments/?user={user['id']}&content_type__model=team&role_definition__name=Galaxy Team Member"
         )
         group_ids = [assignment["object_id"] for assignment in assignment_r["results"]]
         if expected: