Handle object_created and created field transmissions

Author: AlanCoding

ansible_base/rbac/models/role.py

@@ -192,13 +192,15 @@ def __str__(self):
             managed_str = ', managed=True'
         return f'RoleDefinition(pk={self.id}, name={self.name}{managed_str})'
 
-    def give_global_permission(self, actor, assignment_created=None):
-        return self.give_or_remove_global_permission(actor, giving=True, assignment_created=assignment_created)
+    def give_global_permission(self, actor, assignment_created=None, assignment_object_created=None):
+        return self.give_or_remove_global_permission(
+            actor, giving=True, assignment_created=assignment_created, assignment_object_created=assignment_object_created
+        )
 
     def remove_global_permission(self, actor):
         return self.give_or_remove_global_permission(actor, giving=False)
 
-    def give_or_remove_global_permission(self, actor, giving=True, assignment_created=None):
+    def give_or_remove_global_permission(self, actor, giving=True, assignment_created=None, assignment_object_created=None):
         if giving and (self.content_type is not None):
             raise ValidationError('Role definition content type must be null to assign globally')
 
@@ -218,6 +220,8 @@ def give_or_remove_global_permission(self, actor, giving=True, assignment_create
         if giving:
             if assignment_created:
                 kwargs['created'] = assignment_created
+            if assignment_object_created:
+                kwargs['object_created'] = assignment_object_created
             assignment, _ = cls.objects.get_or_create(**kwargs)
         else:
             assignment = cls.objects.filter(**kwargs).first()
@@ -237,8 +241,10 @@ def give_or_remove_global_permission(self, actor, giving=True, assignment_create
 
         return assignment
 
-    def give_permission(self, actor, content_object, assignment_created=None):
-        return self.give_or_remove_permission(actor, content_object, giving=True, assignment_created=assignment_created)
+    def give_permission(self, actor, content_object, assignment_created=None, assignment_object_created=None):
+        return self.give_or_remove_permission(
+            actor, content_object, giving=True, assignment_created=assignment_created, assignment_object_created=assignment_object_created
+        )
 
     def remove_permission(self, actor, content_object):
         return self.give_or_remove_permission(actor, content_object, giving=False)
@@ -263,7 +269,7 @@ def get_or_create_object_role(self, kwargs, defaults):
             object_role = ObjectRole.objects.create(**kwargs, **defaults)
             return (object_role, True)
 
-    def give_or_remove_permission(self, actor, content_object, giving=True, sync_action=False, assignment_created=None):
+    def give_or_remove_permission(self, actor, content_object, giving=True, sync_action=False, assignment_created=None, assignment_object_created=None):
         "Shortcut method to do whatever needed to give user or team these permissions"
         validate_assignment(self, actor, content_object)
 
@@ -295,9 +301,13 @@ def give_or_remove_permission(self, actor, content_object, giving=True, sync_act
         update_teams, to_update = needed_updates_on_assignment(self, actor, object_role, created=created, giving=True)
 
         assignment_defaults = {}
-        object_created = get_created_timestamp(content_object)
-        if object_created:
-            assignment_defaults['object_created'] = object_created
+        # Use provided object_created if available, otherwise get from content_object
+        if assignment_object_created:
+            assignment_defaults['object_created'] = assignment_object_created
+        else:
+            object_created = get_created_timestamp(content_object)
+            if object_created:
+                assignment_defaults['object_created'] = object_created
         if assignment_created:
             assignment_defaults['created'] = assignment_created
 

ansible_base/rbac/service_api/serializers.py

@@ -47,7 +47,7 @@ def to_internal_value(self, value):
         return resource.object_id
 
 
-assignment_common_fields = ('created', 'created_by_ansible_id', 'object_id', 'object_ansible_id', 'content_type', 'role_definition')
+assignment_common_fields = ('created', 'object_created', 'created_by_ansible_id', 'object_id', 'object_ansible_id', 'content_type', 'role_definition')
 
 
 class BaseAssignmentSerializer(serializers.ModelSerializer):
@@ -59,6 +59,8 @@ class BaseAssignmentSerializer(serializers.ModelSerializer):
     from_service = serializers.CharField(write_only=True)
     # Force created field to be writable
     created = serializers.DateTimeField(required=False)
+    # Force object_created field to be writable
+    object_created = serializers.DateTimeField(required=False)
 
     def to_representation(self, instance):
         # hack to surface content_object for ObjectIDAnsibleIDField
@@ -133,10 +135,14 @@ def create(self, validated_data):
                     raise serializers.ValidationError({'object_id': _('Object must be specified for this role assignment')})
 
                 with transaction.atomic():
-                    assignment = rd.give_permission(actor, obj, assignment_created=validated_data.get('created'))
+                    assignment = rd.give_permission(
+                        actor, obj, assignment_created=validated_data.get('created'), assignment_object_created=validated_data.get('object_created')
+                    )
             else:
                 with transaction.atomic():
-                    assignment = rd.give_global_permission(actor, assignment_created=validated_data.get('created'))
+                    assignment = rd.give_global_permission(
+                        actor, assignment_created=validated_data.get('created'), assignment_object_created=validated_data.get('object_created')
+                    )
 
             return assignment
 

test_app/tests/rbac/remote/test_remote_assignment.py

@@ -185,8 +185,6 @@ def test_object_created_field_remote_object(rando, foo_type, foo_rd):
 
     This test should FAIL because the object_created field doesn't exist yet.
     """
-    from datetime import datetime, timezone
-
     # Create a remote object stand-in
     remote_foo = RemoteObject(content_type=foo_type, object_id=42)
 

test_app/tests/rbac/remote/test_service_api.py

@@ -3,7 +3,7 @@
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleUserAssignment
 from test_app.models import Team, User
 
 
@@ -492,6 +492,7 @@ def test_serializer_allows_null_values_in_validation(self, admin_api_client, ran
         validated_data = serializer.validated_data
         assert 'created_by' not in validated_data or validated_data.get('created_by') is None
 
+
 def test_service_assignment_created_timestamp_sync(admin_api_client, rando, inv_rd, inventory):
     """
     Test that demonstrates the field sync issue: the 'created' timestamp field is displayed
@@ -539,3 +540,114 @@ def test_service_assignment_created_timestamp_sync(admin_api_client, rando, inv_
     response_created = parse_datetime(response.data['created'])
     # Note: This will show the auto-generated timestamp, not our custom one
     assert response_created == expected_created, f"Response created timestamp should match: expected '{expected_created}' but got '{response_created}'"
+
+
+@pytest.mark.django_db
+def test_service_assignment_object_created_timestamp_sync(admin_api_client, rando, inv_rd, inventory):
+    """
+    Test that the 'object_created' field can be synchronized in both directions:
+    1. POST to /assign/ accepts a provided 'object_created' value
+    2. Serializing local assignments includes the 'object_created' field from the DB
+    """
+    from datetime import datetime, timezone
+
+    from django.utils.dateparse import parse_datetime
+
+    url = get_relative_url('serviceuserassignment-assign')
+
+    creator_user = User.objects.create(username='object_created_creator')
+
+    # Set a specific object_created timestamp that's different from the actual object's created time
+    custom_object_created = datetime(2022, 6, 15, 14, 30, 0, tzinfo=timezone.utc)
+    custom_object_created_str = custom_object_created.isoformat()
+
+    post_data = {
+        "role_definition": inv_rd.name,
+        "user_ansible_id": str(rando.resource.ansible_id),
+        "object_id": str(inventory.pk),
+        "created_by_ansible_id": str(creator_user.resource.ansible_id),
+        "object_created": custom_object_created_str,
+        "from_service": "test_service",
+    }
+
+    # Test 1: POST accepts object_created value
+    response = admin_api_client.post(url, data=post_data)
+    assert response.status_code == 201, response.data
+
+    assignment = RoleUserAssignment.objects.get(user=rando, role_definition=inv_rd, object_id=inventory.pk)
+
+    # Verify the custom object_created timestamp was properly set
+    expected_object_created = custom_object_created
+    actual_object_created = assignment.object_created
+
+    assert (
+        actual_object_created == expected_object_created
+    ), f"object_created should be synchronized: Expected '{expected_object_created}' but got '{actual_object_created}'"
+
+    # Test 2: Serializing local assignments includes object_created field
+    list_url = get_relative_url('serviceuserassignment-list')
+    response = admin_api_client.get(list_url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    # Find our assignment in the list
+    assignments = [a for a in response.data['results'] if a['role_definition'] == inv_rd.name and str(a['object_id']) == str(inventory.pk)]
+    assert len(assignments) >= 1, "Should find at least our assignment"
+
+    # Check that object_created is properly serialized
+    assignment_data = assignments[0]
+    assert 'object_created' in assignment_data, "object_created field should be present in serialized output"
+
+    response_object_created = parse_datetime(assignment_data['object_created'])
+    assert (
+        response_object_created == expected_object_created
+    ), f"Serialized object_created should match stored value: expected '{expected_object_created}' but got '{response_object_created}'"
+
+
+@pytest.mark.django_db
+def test_service_assignment_object_created_from_local_object(admin_api_client, rando, org_inv_rd, organization):
+    """
+    Test that when no object_created is provided in POST, the field is automatically set
+    from the local object's created timestamp and properly serialized.
+    """
+    url = get_relative_url('serviceuserassignment-assign')
+
+    post_data = {
+        "role_definition": org_inv_rd.name,
+        "user_ansible_id": str(rando.resource.ansible_id),
+        "object_id": str(organization.pk),
+        "from_service": "test_service",
+        # Note: no object_created provided - should default to organization.created
+    }
+
+    # Create assignment without providing object_created
+    response = admin_api_client.post(url, data=post_data)
+    assert response.status_code == 201, response.data
+
+    assignment = RoleUserAssignment.objects.get(user=rando, role_definition=org_inv_rd, object_id=organization.pk)
+
+    # Verify object_created was set to the organization's created timestamp
+    expected_object_created = organization.created
+    actual_object_created = assignment.object_created
+
+    assert (
+        actual_object_created == expected_object_created
+    ), f"object_created should default to organization.created: Expected '{expected_object_created}' but got '{actual_object_created}'"
+
+    # Verify serialization includes the correct object_created value
+    list_url = get_relative_url('serviceuserassignment-list')
+    response = admin_api_client.get(list_url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    # Find our assignment
+    assignments = [a for a in response.data['results'] if a['role_definition'] == org_inv_rd.name and str(a['object_id']) == str(organization.pk)]
+    assert len(assignments) >= 1, "Should find at least our assignment"
+
+    assignment_data = assignments[0]
+    assert 'object_created' in assignment_data, "object_created field should be present in serialized output"
+
+    from django.utils.dateparse import parse_datetime
+
+    response_object_created = parse_datetime(assignment_data['object_created'])
+    assert (
+        response_object_created == expected_object_created
+    ), f"Serialized object_created should match organization.created: expected '{expected_object_created}' but got '{response_object_created}'"