Unified solution to serialization of ansible_id

AlanCoding · AlanCoding · commit 287a4a7476a1 · 2025-07-30T15:04:54.000-04:00
diff --git a/ansible_base/rbac/api/fields.py b/ansible_base/rbac/api/fields.py
@@ -0,0 +1,41 @@
+from django.apps import apps
+from django.core.exceptions import ObjectDoesNotExist
+from rest_framework import serializers
+
+
+class ActorAnsibleIdField(serializers.UUIDField):
+    """
+    UUID field that serializes actor objects to their ansible_id and accepts ansible_id for deserialization.
+
+    Always resolves ansible_id input to the corresponding actor object.
+    Uses the source parameter to determine which field to populate.
+    """
+
+    def to_representation(self, actor):
+        """Convert actor object to its ansible_id UUID"""
+        if actor is None:
+            return None
+        try:
+            if hasattr(actor, 'resource') and actor.resource:
+                return super().to_representation(actor.resource.ansible_id)
+        except ObjectDoesNotExist:
+            # Resource doesn't exist, return None
+            pass
+        return None
+
+    def to_internal_value(self, data):
+        """Convert ansible_id UUID to actor object"""
+        if data is None:
+            return None
+
+        # Let UUIDField handle validation and conversion
+        uuid_value = super().to_internal_value(data)
+
+        # Always resolve to actor object
+        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
+        try:
+            resource = resource_cls.objects.get(ansible_id=uuid_value)
+        except resource_cls.DoesNotExist:
+            source_name = getattr(self, 'source', 'actor')
+            raise serializers.ValidationError(f"No {source_name} found with ansible_id={uuid_value}")
+        return resource.content_object
diff --git a/ansible_base/rbac/api/serializers.py b/ansible_base/rbac/api/serializers.py
@@ -20,6 +20,7 @@
 
 from ..models import DABContentType, DABPermission
 from ..remote import RemoteObject
+from .fields import ActorAnsibleIdField
 from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
 
 logger = logging.getLogger(__name__)
@@ -116,16 +117,28 @@ def get_by_ansible_id(self, ansible_id, requesting_user, for_field):
             raise ValidationError({for_field: msg.format(pk_value=ansible_id)})
         return resource.content_object
 
-    def get_actor_from_data(self, validated_data, requesting_user):
+    def validate(self, attrs):
+        """Validate that exactly one of actor or actor_ansible_id is provided"""
         actor_aid_field = f'{self.actor_field}_ansible_id'
-        if validated_data.get(self.actor_field) and validated_data.get(actor_aid_field):
+
+        # Check what was actually provided in the request
+        has_actor_in_request = self.actor_field in self.initial_data
+        has_actor_aid_in_request = actor_aid_field in self.initial_data
+
+        if has_actor_in_request and has_actor_aid_in_request:
             self.raise_id_fields_error(self.actor_field, actor_aid_field)
-        elif validated_data.get(self.actor_field):
+        elif not has_actor_in_request and not has_actor_aid_in_request:
+            self.raise_id_fields_error(self.actor_field, actor_aid_field)
+
+        return super().validate(attrs)
+
+    def get_actor_from_data(self, validated_data, requesting_user):
+        actor_aid_field = f'{self.actor_field}_ansible_id'
+        if validated_data.get(self.actor_field):
             actor = validated_data[self.actor_field]
-        elif ansible_id := validated_data.get(actor_aid_field):
-            actor = self.get_by_ansible_id(ansible_id, requesting_user, for_field=actor_aid_field)
         else:
-            self.raise_id_fields_error(self.actor_field, f'{self.actor_field}_ansible_id')
+            # Actor is already resolved by ActorAnsibleIdField and validated
+            actor = validated_data[self.actor_field]
         return actor
 
     def get_object_from_data(self, validated_data, role_definition, requesting_user):
@@ -220,7 +233,8 @@ def _get_summary_fields(self, obj) -> dict[str, dict]:
 
 class RoleUserAssignmentSerializer(BaseAssignmentSerializer):
     actor_field = 'user'
-    user_ansible_id = serializers.UUIDField(
+    user_ansible_id = ActorAnsibleIdField(
+        source='user',
         required=False,
         help_text=_('The resource ID of the user who will receive permissions from this assignment. An alternative to user field.'),
         allow_null=True,  # for ease of use of the browseable API
@@ -236,7 +250,8 @@ def get_actor_queryset(self, requesting_user):
 
 class RoleTeamAssignmentSerializer(BaseAssignmentSerializer):
     actor_field = 'team'
-    team_ansible_id = serializers.UUIDField(
+    team_ansible_id = ActorAnsibleIdField(
+        source='team',
         required=False,
         help_text=_('The resource ID of the team who will receive permissions from this assignment. An alternative to team field.'),
         allow_null=True,
diff --git a/ansible_base/rbac/service_api/serializers.py b/ansible_base/rbac/service_api/serializers.py
@@ -4,6 +4,7 @@
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
+from ..api.fields import ActorAnsibleIdField
 from ..models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ..remote import RemoteObject
 
@@ -24,25 +25,6 @@ class Meta:
         fields = ['api_slug', 'codename', 'content_type', 'name']
 
 
-class ActorAnsibleIDField(serializers.Field):
-    def to_representation(self, actor):
-        if actor is None:
-            return None
-        resource = actor.resource
-        if resource is None:
-            return None
-        return str(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
-
-
 class ObjectIDAnsibleIDField(serializers.Field):
     "This is an ansible_id field intended to be used with source pointing to object_id, so, does conversion"
 
@@ -71,7 +53,7 @@ def to_internal_value(self, value):
 class BaseAssignmentSerializer(serializers.ModelSerializer):
     content_type = serializers.SlugRelatedField(read_only=True, slug_field='api_slug')
     role_definition = serializers.SlugRelatedField(slug_field='name', queryset=RoleDefinition.objects.all())
-    created_by_ansible_id = ActorAnsibleIDField(source='created_by', required=False)
+    created_by_ansible_id = ActorAnsibleIdField(source='created_by', required=False)
     object_ansible_id = ObjectIDAnsibleIDField(source='object_id', required=False, allow_null=True)
     object_id = serializers.CharField(allow_blank=True, required=False, allow_null=True)
     from_service = serializers.CharField(write_only=True)
@@ -158,7 +140,7 @@ def create(self, validated_data):
 
 
 class ServiceRoleUserAssignmentSerializer(BaseAssignmentSerializer):
-    user_ansible_id = ActorAnsibleIDField(source='user', required=True)
+    user_ansible_id = ActorAnsibleIdField(source='user', required=True)
     actor_field = 'user'
 
     class Meta:
@@ -167,7 +149,7 @@ class Meta:
 
 
 class ServiceRoleTeamAssignmentSerializer(BaseAssignmentSerializer):
-    team_ansible_id = ActorAnsibleIDField(source='team', required=True)
+    team_ansible_id = ActorAnsibleIdField(source='team', required=True)
     actor_field = 'team'
 
     class Meta:
diff --git a/test_app/tests/rbac/api/test_role_assignment_ansible_id_serialization.py b/test_app/tests/rbac/api/test_role_assignment_ansible_id_serialization.py
@@ -0,0 +1,99 @@
+import pytest
+
+from ansible_base.lib.utils.response import get_relative_url
+from test_app.models import Team, User
+
+
+@pytest.mark.django_db
+class TestRoleAssignmentAnsibleIdSerialization:
+    """Test serialization of ansible_id fields in role assignment endpoints"""
+
+    def test_role_user_assignment_serializes_user_ansible_id(self, admin_api_client, inventory, inv_rd):
+        """Test that RoleUserAssignment API returns user_ansible_id instead of null"""
+        user = User.objects.create(username='test-user')
+        assignment = inv_rd.give_permission(user, inventory)
+
+        # Verify user has a resource and ansible_id
+        assert hasattr(user, 'resource')
+        assert user.resource is not None
+        assert user.resource.ansible_id is not None
+        expected_ansible_id = str(user.resource.ansible_id)
+
+        # Get the assignment via API
+        url = get_relative_url('roleuserassignment-detail', kwargs={'pk': assignment.pk})
+        response = admin_api_client.get(url)
+        assert response.status_code == 200
+
+        # The user_ansible_id field should be populated, not null
+        assert response.data['user_ansible_id'] is not None
+        assert response.data['user_ansible_id'] == expected_ansible_id
+
+        # Also verify the user field contains the primary key for backward compatibility
+        assert response.data['user'] == user.pk
+
+    def test_role_team_assignment_serializes_team_ansible_id(self, admin_api_client, inventory, inv_rd, organization):
+        """Test that RoleTeamAssignment API returns team_ansible_id instead of null"""
+        team = Team.objects.create(name='test-team', organization=organization)
+        assignment = inv_rd.give_permission(team, inventory)
+
+        # Verify team has a resource and ansible_id
+        assert hasattr(team, 'resource')
+        assert team.resource is not None
+        assert team.resource.ansible_id is not None
+        expected_ansible_id = str(team.resource.ansible_id)
+
+        # Get the assignment via API
+        url = get_relative_url('roleteamassignment-detail', kwargs={'pk': assignment.pk})
+        response = admin_api_client.get(url)
+        assert response.status_code == 200
+
+        # The team_ansible_id field should be populated, not null
+        assert response.data['team_ansible_id'] is not None
+        assert response.data['team_ansible_id'] == expected_ansible_id
+
+        # Also verify the team field contains the primary key for backward compatibility
+        assert response.data['team'] == team.pk
+
+    def test_role_user_assignment_list_includes_ansible_id(self, admin_api_client, inventory, inv_rd):
+        """Test that RoleUserAssignment list endpoint includes user_ansible_id"""
+        user = User.objects.create(username='test-user-list')
+        assignment = inv_rd.give_permission(user, inventory)
+        expected_ansible_id = str(user.resource.ansible_id)
+
+        # Get assignments list via API
+        url = get_relative_url('roleuserassignment-list')
+        response = admin_api_client.get(url)
+        assert response.status_code == 200
+
+        # Find our assignment in the results
+        assignment_data = None
+        for item in response.data['results']:
+            if item['id'] == assignment.pk:
+                assignment_data = item
+                break
+
+        assert assignment_data is not None
+        assert assignment_data['user_ansible_id'] is not None
+        assert assignment_data['user_ansible_id'] == expected_ansible_id
+
+    def test_role_team_assignment_list_includes_ansible_id(self, admin_api_client, inventory, inv_rd, organization):
+        """Test that RoleTeamAssignment list endpoint includes team_ansible_id"""
+        team = Team.objects.create(name='test-team-list', organization=organization)
+        assignment = inv_rd.give_permission(team, inventory)
+        expected_ansible_id = str(team.resource.ansible_id)
+
+        # Get assignments list via API
+        url = get_relative_url('roleteamassignment-list')
+        response = admin_api_client.get(url)
+        assert response.status_code == 200
+
+        # Find our assignment in the results
+        assignment_data = None
+        for item in response.data['results']:
+            if item['id'] == assignment.pk:
+                assignment_data = item
+                break
+
+        assert assignment_data is not None
+        assert assignment_data['team_ansible_id'] is not None
+        assert assignment_data['team_ansible_id'] == expected_ansible_id
