Debug the assign and unassign views

Author: AlanCoding

ansible_base/rbac/service_api/serializers.py

@@ -1,7 +1,7 @@
 from django.apps import apps
 from rest_framework import serializers
 
-from ..models import DABContentType, DABPermission, RoleTeamAssignment, RoleUserAssignment
+from ..models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ..remote import RemoteObject
 
 
@@ -21,13 +21,27 @@ class Meta:
         fields = ['api_slug', 'codename', 'content_type', 'name']
 
 
+class ActorAnsibleIDField(serializers.Field):
+    def to_representation(self, actor):
+        return str(actor.resource.ansible_id)
+
+    def to_internal_value(self, data):
+        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
+        try:
+            resource = resource_cls.objects.get(ansible_id=data)
+        except resource_cls.DoesNotExist:
+            raise serializers.ValidationError(f"No {self.source} found with ansible_id={data}")
+
+        return resource.content_object
+
+
 assignment_common_fields = ('created', 'created_by_ansible_id', 'object_id', 'object_ansible_id', 'content_type', 'role_definition')
 
 
 class BaseAssignmentSerializer(serializers.ModelSerializer):
     content_type = serializers.SlugRelatedField(read_only=True, slug_field='api_slug')
-    role_definition = serializers.SlugRelatedField(read_only=True, slug_field='name')
-    created_by_ansible_id = serializers.SerializerMethodField()
+    role_definition = serializers.SlugRelatedField(slug_field='name', queryset=RoleDefinition.objects.all())
+    created_by_ansible_id = ActorAnsibleIDField(source='created_by', required=False)
     object_ansible_id = serializers.SerializerMethodField()
     # TODO: use the from_service to control what we sync back to
     from_service = serializers.CharField(write_only=True)
@@ -44,35 +58,25 @@ def get_object_ansible_id(self, obj):
         return None
 
     def find_existing_assignment(self, queryset):
-        actor_ansible_id = self.validated_data[f'{self.actor_field}_ansible_id']
+        actor = self.validated_data[self.actor_field]
         object_id = self.validated_data['object_id']
         role_definition = self.validated_data['role_definition']
-
-        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
-        actor_resource = resource_cls.objects.get(ansible_id=actor_ansible_id)
-        actor = actor_resource.content_object
         return queryset.filter(object_id=object_id, role_definition=role_definition, **{self.actor_field: actor}).first()
 
 
 class RoleUserAssignmentSerializer(BaseAssignmentSerializer):
-    user_ansible_id = serializers.SerializerMethodField()
+    user_ansible_id = ActorAnsibleIDField(source='user', required=True)
     actor_field = 'user'
 
     class Meta:
         model = RoleUserAssignment
         fields = assignment_common_fields + ('user_ansible_id',)
 
-    def get_user_ansible_id(self, obj):
-        return str(obj.user.resource.ansible_id)
-
 
 class RoleTeamAssignmentSerializer(BaseAssignmentSerializer):
-    user_ansible_id = serializers.SerializerMethodField()
+    team_ansible_id = ActorAnsibleIDField(source='team', required=True)
     actor_field = 'team'
 
     class Meta:
         model = RoleTeamAssignment
         fields = assignment_common_fields + ('team_ansible_id',)
-
-    def get_team_ansible_id(self, obj):
-        return str(obj.team.resource.ansible_id)

ansible_base/rest_filters/rest_framework/ansible_id_backend.py

@@ -135,4 +135,4 @@ def filter_queryset(self, request, queryset, view):
                 query |= q
             queryset = queryset.filter(query)
 
-        return super().filter_queryset(request, queryset, view)
+        return queryset

test_app/tests/rbac/remote/test_service_api.py

@@ -116,3 +116,19 @@ def test_list_role_user_assignments(admin_api_client, rando, inv_rd, inventory):
     assert int(from_api['object_id']) == inventory.id
     assert from_api['user_ansible_id'] == str(rando.resource.ansible_id)
     assert from_api['content_type'] == 'aap.inventory'
+
+
+@pytest.mark.django_db
+def test_apply_role_assignment(admin_api_client, rando, inv_rd, inventory):
+    url = get_relative_url('serviceuserassignment-assign')
+
+    data = {"role_definition": inv_rd.name, "user_ansible_id": str(rando.resource.ansible_id), "object_id": inventory.pk}
+
+    assert not rando.has_obj_perm(inventory, 'change')
+    response = admin_api_client.post(url, data=data)
+    assert response.status_code == 201, response.data
+    assert not rando.has_obj_perm(inventory, 'change')
+
+    # Second try, response code indicates assignment already exists
+    response = admin_api_client.post(url, data=data)
+    assert response.status_code == 409, response.data