Add data migration for object_created and test

AlanCoding · AlanCoding · commit f3120b82257d · 2025-09-10T11:22:06.000-04:00
diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py
@@ -1,4 +1,5 @@
 # Generated by Django 4.2.23 on 2025-06-30 12:48
+# Modified by Claude (Sonnet 4) - Added populate_object_created_field data migration
 
 import logging
 
@@ -23,6 +24,12 @@ def create_types_if_needed(apps, schema_editor):
         create_DAB_contenttypes(apps=apps)
 
 
+def populate_object_created_field(apps, schema_editor):
+    """Populate the object_created field for existing role assignments."""
+    from ._utils import populate_object_created_field as _populate_object_created_field
+    return _populate_object_created_field(apps, schema_editor)
+
+
 def migrate_content_type(apps, schema_editor):
     ct_cls = apps.get_model('dab_rbac', 'DABContentType')
     ct_cls.objects.clear_cache()
@@ -68,4 +75,5 @@ class Migration(migrations.Migration):
     operations = [
         migrations.RunPython(create_types_if_needed, migrations.RunPython.noop),
         migrations.RunPython(migrate_content_type, migrations.RunPython.noop),
+        migrations.RunPython(populate_object_created_field, migrations.RunPython.noop),
     ]
diff --git a/ansible_base/rbac/migrations/_utils.py b/ansible_base/rbac/migrations/_utils.py
@@ -1,5 +1,9 @@
+import logging
+from datetime import datetime
 from django.db import models
 
+logger = logging.getLogger(__name__)
+
 # This method has moved, and this is put here temporarily to make branch management easier
 from ansible_base.rbac.management import create_dab_permissions as create_custom_permissions  # noqa
 
@@ -45,3 +49,67 @@ def give_permissions(apps, rd, users=(), teams=(), object_id=None, content_type_
                 for team_id in teams
             ]
         RoleTeamAssignment.objects.bulk_create(team_assignments, ignore_conflicts=True)
+
+
+def get_model_class_from_content_type(apps, content_type):
+    """
+    Get a model class from a content type in a migration-safe way.
+
+    This is needed because content_type.model_class() is not available in migrations.
+    """
+    try:
+        return apps.get_model(content_type.app_label, content_type.model)
+    except (LookupError, AttributeError):
+        return None
+
+
+def populate_object_created_field(apps, schema_editor=None):
+    """Populate the object_created field for existing role assignments."""
+    assignment_models = [
+        ('roleuserassignment', 'RoleUserAssignment'),
+        ('roleteamassignment', 'RoleTeamAssignment'),
+    ]
+
+    updated_count = 0
+
+    for model_name, model_class_name in assignment_models:
+        assignment_cls = apps.get_model('dab_rbac', model_name)
+        assignments_to_update = assignment_cls.objects.filter(object_created__isnull=True)
+
+        for assignment in assignments_to_update:
+            object_created_value = None
+
+            # Try to get the actual object to extract its created timestamp
+            if assignment.object_id and assignment.content_type:
+                try:
+                    # Get the model class from the old content_type field (before migration to DABContentType)
+                    model_class = get_model_class_from_content_type(apps, assignment.content_type)
+                    if model_class:
+                        try:
+                            # Try to get the actual object
+                            actual_object = model_class.objects.get(pk=assignment.object_id)
+
+                            # Try to get created timestamp from common field names
+                            for field_name in ('created', 'created_at'):
+                                if hasattr(actual_object, field_name):
+                                    val = getattr(actual_object, field_name)
+                                    if isinstance(val, datetime):
+                                        object_created_value = val
+                                        break
+                        except (model_class.DoesNotExist, ValueError, TypeError):
+                            # Object doesn't exist or can't be retrieved, skip
+                            pass
+                except (AttributeError, LookupError):
+                    # Content type or model class issues, skip
+                    pass
+
+            # Update the assignment if we found a created timestamp
+            if object_created_value:
+                assignment.object_created = object_created_value
+                assignment.save(update_fields=['object_created'])
+                updated_count += 1
+
+    if updated_count:
+        logger.info(f'Populated object_created field for {updated_count} existing role assignments')
+
+    return updated_count
diff --git a/ansible_base/rbac/service_api/serializers.py b/ansible_base/rbac/service_api/serializers.py
@@ -60,7 +60,7 @@ class BaseAssignmentSerializer(serializers.ModelSerializer):
     # Force created field to be writable
     created = serializers.DateTimeField(required=False)
     # Force object_created field to be writable
-    object_created = serializers.DateTimeField(required=False)
+    object_created = serializers.DateTimeField(required=False, allow_null=True)
 
     def to_representation(self, instance):
         # hack to surface content_object for ObjectIDAnsibleIDField
diff --git a/test_app/tests/rbac/remote/test_service_api.py b/test_app/tests/rbac/remote/test_service_api.py
@@ -320,89 +320,40 @@ def test_role_types_and_permissions_payload_shape(user_api_client):
 class TestCreatedByAnsibleIdAllowNull:
     """Test that created_by_ansible_id field accepts null values and omissions"""
 
-    def test_service_user_assignment_with_null_created_by(self, admin_api_client, rando, inv_rd, inventory):
-        """Test that ServiceRoleUserAssignmentSerializer accepts null created_by_ansible_id"""
-        url = get_relative_url('serviceuserassignment-assign')
-        data = {
-            "role_definition": inv_rd.name,
-            "user_ansible_id": str(rando.resource.ansible_id),
-            "object_id": inventory.pk,
-            "created_by_ansible_id": "",  # Use empty string instead of None
-        }
-
-        response = admin_api_client.post(url, data=data)
-        assert response.status_code == 201, response.data
-        assert rando.has_obj_perm(inventory, 'change')
-
-    def test_service_user_assignment_without_created_by(self, admin_api_client, rando, inv_rd, inventory):
-        """Test that ServiceRoleUserAssignmentSerializer works when created_by_ansible_id is omitted"""
-        url = get_relative_url('serviceuserassignment-assign')
-        data = {
-            "role_definition": inv_rd.name,
-            "user_ansible_id": str(rando.resource.ansible_id),
-            "object_id": inventory.pk,
-            # created_by_ansible_id is intentionally omitted
-        }
-
-        response = admin_api_client.post(url, data=data)
-        assert response.status_code == 201, response.data
-        assert rando.has_obj_perm(inventory, 'change')
-
-    def test_service_user_assignment_with_valid_created_by(self, admin_api_client, rando, inv_rd, inventory):
-        """Test that valid created_by_ansible_id values still work correctly"""
-        creator = User.objects.create(username='creator-user')
-        url = get_relative_url('serviceuserassignment-assign')
-        data = {
-            "role_definition": inv_rd.name,
-            "user_ansible_id": str(rando.resource.ansible_id),
-            "object_id": inventory.pk,
-            "created_by_ansible_id": str(creator.resource.ansible_id),
-        }
-
-        response = admin_api_client.post(url, data=data)
-        assert response.status_code == 201, response.data
-        assert rando.has_obj_perm(inventory, 'change')
-
-    def test_service_team_assignment_with_null_created_by(self, admin_api_client, team, inv_rd, inventory, member_rd, rando):
-        """Test that ServiceRoleTeamAssignmentSerializer accepts null created_by_ansible_id"""
-        member_rd.give_permission(rando, team)
-        url = get_relative_url('serviceteamassignment-assign')
-        data = {
-            "role_definition": inv_rd.name,
-            "team_ansible_id": str(team.resource.ansible_id),
-            "object_id": inventory.pk,
-            "created_by_ansible_id": "",  # Use empty string instead of None
-        }
-
-        response = admin_api_client.post(url, data=data)
-        assert response.status_code == 201, response.data
-        assert rando.has_obj_perm(inventory, 'change')
-
-    def test_service_team_assignment_without_created_by(self, admin_api_client, team, inv_rd, inventory, member_rd, rando):
-        """Test that ServiceRoleTeamAssignmentSerializer works when created_by_ansible_id is omitted"""
-        member_rd.give_permission(rando, team)
-        url = get_relative_url('serviceteamassignment-assign')
+    @pytest.mark.parametrize(
+        'actor_type,created_by_value',
+        [
+            ('user', ''),  # empty string
+            ('user', None),  # omitted (None means field not present)
+            ('user', 'valid'),  # valid creator
+            ('team', ''),  # empty string
+            ('team', None),  # omitted
+            ('team', 'valid'),  # valid creator
+        ],
+    )
+    def test_service_assignment_created_by_handling(self, admin_api_client, rando, inv_rd, inventory, team, member_rd, actor_type, created_by_value):
+        """Test that ServiceRoleAssignmentSerializer handles created_by_ansible_id correctly"""
+        # Setup for team assignments
+        if actor_type == 'team':
+            member_rd.give_permission(rando, team)
+            actor = team
+        else:
+            actor = rando
+
+        url = get_relative_url(f'service{actor_type}assignment-assign')
         data = {
             "role_definition": inv_rd.name,
-            "team_ansible_id": str(team.resource.ansible_id),
+            f"{actor_type}_ansible_id": str(actor.resource.ansible_id),
             "object_id": inventory.pk,
         }
 
-        response = admin_api_client.post(url, data=data)
-        assert response.status_code == 201, response.data
-        assert rando.has_obj_perm(inventory, 'change')
-
-    def test_service_team_assignment_with_valid_created_by(self, admin_api_client, team, inv_rd, inventory, member_rd, rando):
-        """Test that valid created_by_ansible_id values still work correctly for teams"""
-        member_rd.give_permission(rando, team)
-        creator = User.objects.create(username='team-creator-user')
-        url = get_relative_url('serviceteamassignment-assign')
-        data = {
-            "role_definition": inv_rd.name,
-            "team_ansible_id": str(team.resource.ansible_id),
-            "object_id": inventory.pk,
-            "created_by_ansible_id": str(creator.resource.ansible_id),
-        }
+        # Handle different created_by_value scenarios
+        if created_by_value == '':
+            data["created_by_ansible_id"] = ""
+        elif created_by_value == 'valid':
+            creator = User.objects.create(username=f'{actor_type}-creator-user')
+            data["created_by_ansible_id"] = str(creator.resource.ansible_id)
+        # None means field is omitted (not added to data)
 
         response = admin_api_client.post(url, data=data)
         assert response.status_code == 201, response.data
@@ -543,54 +494,74 @@ def test_service_assignment_created_timestamp_sync(admin_api_client, rando, inv_
 
 
 @pytest.mark.django_db
-def test_service_assignment_object_created_timestamp_sync(admin_api_client, rando, inv_rd, inventory):
+@pytest.mark.parametrize(
+    'object_created_source',
+    [
+        'custom',  # Provide custom timestamp
+        'local_object',  # Use local object's created timestamp
+    ],
+)
+def test_service_assignment_object_created_sync(admin_api_client, rando, inv_rd, inventory, org_inv_rd, organization, object_created_source):
     """
     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
+    2. When no object_created is provided, it defaults to the local object's created timestamp
+    3. 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()
+    if object_created_source == 'custom':
+        # Use inventory with custom timestamp
+        target_object = inventory
+        role_def = inv_rd
+        custom_object_created = datetime(2022, 6, 15, 14, 30, 0, tzinfo=timezone.utc)
+        expected_object_created = custom_object_created
 
-    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",
-    }
+        post_data = {
+            "role_definition": role_def.name,
+            "user_ansible_id": str(rando.resource.ansible_id),
+            "object_id": str(target_object.pk),
+            "object_created": custom_object_created.isoformat(),
+            "from_service": "test_service",
+        }
+    else:  # local_object
+        # Use organization without providing object_created
+        target_object = organization
+        role_def = org_inv_rd
+        expected_object_created = organization.created
+
+        post_data = {
+            "role_definition": role_def.name,
+            "user_ansible_id": str(rando.resource.ansible_id),
+            "object_id": str(target_object.pk),
+            "from_service": "test_service",
+            # Note: no object_created provided - should default to target_object.created
+        }
 
-    # Test 1: POST accepts object_created value
+    # Test 1: POST accepts object_created value or defaults to local object
     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)
+    assignment = RoleUserAssignment.objects.get(user=rando, role_definition=role_def, object_id=target_object.pk)
 
-    # Verify the custom object_created timestamp was properly set
-    expected_object_created = custom_object_created
+    # Verify the object_created timestamp was properly set
     actual_object_created = assignment.object_created
-
+    operation_type = 'synchronized' if object_created_source == 'custom' else 'defaulted to local object'
     assert (
         actual_object_created == expected_object_created
-    ), f"object_created should be synchronized: Expected '{expected_object_created}' but got '{actual_object_created}'"
+    ), f"object_created should be {operation_type}: 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)]
+    assignments = [a for a in response.data['results'] if a['role_definition'] == role_def.name and str(a['object_id']) == str(target_object.pk)]
     assert len(assignments) >= 1, "Should find at least our assignment"
 
     # Check that object_created is properly serialized
@@ -601,53 +572,3 @@ def test_service_assignment_object_created_timestamp_sync(admin_api_client, rand
     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}'"
diff --git a/test_app/tests/resource_registry/test_resources_api_rest_client.py b/test_app/tests/resource_registry/test_resources_api_rest_client.py
@@ -193,6 +193,7 @@ def test_list_role_permissions_all_pages(resource_client):
 def _assert_assignment_matches_data(assignment, data, obj, actor):
     assert 'created' in data, data
     # assert DateTimeField().to_representation(assignment.created) == data['created']  # TODO
+    assert 'object_created' in data, data
     assert str(assignment.created_by.resource.ansible_id) == data['created_by_ansible_id']
     assert assignment.object_id == obj.id
     assert str(assignment.object_id) == str(data['object_id'])
