diff --git a/ansible_base/rbac/migrations/0004_remote_permissions_additions.py b/ansible_base/rbac/migrations/0004_remote_permissions_additions.py index c7159b886..cb00b6c42 100644 --- a/ansible_base/rbac/migrations/0004_remote_permissions_additions.py +++ b/ansible_base/rbac/migrations/0004_remote_permissions_additions.py @@ -3,6 +3,7 @@ import ansible_base.rbac.models.content_type import ansible_base.rbac.remote from django.db import migrations, models +import django.utils.timezone class Migration(migrations.Migration): diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py index a0f7852a0..028ffecd9 100644 --- a/ansible_base/rbac/migrations/0005_remote_permissions_data.py +++ b/ansible_base/rbac/migrations/0005_remote_permissions_data.py @@ -23,6 +23,7 @@ def create_types_if_needed(apps, schema_editor): create_DAB_contenttypes(apps=apps) + def migrate_content_type(apps, schema_editor): ct_cls = apps.get_model('dab_rbac', 'DABContentType') ct_cls.objects.clear_cache() diff --git a/ansible_base/rbac/migrations/0009_object_created_field.py b/ansible_base/rbac/migrations/0009_object_created_field.py new file mode 100644 index 000000000..43da16317 --- /dev/null +++ b/ansible_base/rbac/migrations/0009_object_created_field.py @@ -0,0 +1,45 @@ +# Generated by Django 4.2.21 on 2025-09-10 17:11 +# Modified by Claude (Sonnet 4) - Added object_created field and data migration + +from django.db import migrations, models +import django.utils.timezone + + +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) + + +class Migration(migrations.Migration): + + dependencies = [ + ('dab_rbac', '0008_remote_permissions_cleanup'), + ] + + operations = [ + # Added field for a checksum like purpose for remote models + migrations.AddField( + model_name='roleuserassignment', + name='object_created', + field=models.DateTimeField(help_text='The created timestamp of related object, if applicable.', null=True), + ), + migrations.AddField( + model_name='roleteamassignment', + name='object_created', + field=models.DateTimeField(help_text='The created timestamp of related object, if applicable.', null=True), + ), + # Make assignment created timestamp backdateable + migrations.AlterField( + model_name='roleteamassignment', + name='created', + field=models.DateTimeField(default=django.utils.timezone.now, editable=False, help_text='The date/time this resource was created.'), + ), + migrations.AlterField( + model_name='roleuserassignment', + name='created', + field=models.DateTimeField(default=django.utils.timezone.now, editable=False, help_text='The date/time this resource was created.'), + ), + # Data migration to populate object_created field + 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 index 83cb54d6d..4f9dc47fa 100644 --- 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/models/role.py b/ansible_base/rbac/models/role.py index 07c49d641..8166c1570 100644 --- a/ansible_base/rbac/models/role.py +++ b/ansible_base/rbac/models/role.py @@ -1,5 +1,6 @@ import logging from collections.abc import Iterable +from datetime import datetime from typing import Optional, Type, Union from uuid import UUID @@ -9,6 +10,7 @@ from django.db.models.functions import Cast from django.db.models.query import QuerySet from django.db.utils import IntegrityError +from django.utils import timezone from django.utils.translation import gettext_lazy as _ # Django-rest-framework @@ -63,6 +65,18 @@ def __getattr__(self, attr): return rd +def get_created_timestamp(obj: Union[models.Model, RemoteObject]) -> Optional[datetime]: + """Given some obj from the users app, try to infer the created timestamp""" + if isinstance(obj, RemoteObject): + return obj.created + for field_name in ('created', 'created_at'): + if hasattr(obj, field_name): + val = getattr(obj, field_name) + if isinstance(val, datetime): + return val + return None + + class RoleDefinitionManager(models.Manager): def contribute_to_class(self, cls: Type[models.Model], name: str) -> None: """After Django populates the model for the manager, attach the manager role manager""" @@ -178,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): - return self.give_or_remove_global_permission(actor, giving=True) + 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): + 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') @@ -202,6 +218,10 @@ def give_or_remove_global_permission(self, actor, giving=True): raise RuntimeError(f'Cannot {giving and "give" or "remove"} permission for {actor}, must be a user or team') 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() @@ -221,8 +241,10 @@ def give_or_remove_global_permission(self, actor, giving=True): return assignment - def give_permission(self, actor, content_object): - return self.give_or_remove_permission(actor, content_object, giving=True) + 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) @@ -247,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): + 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) @@ -278,15 +300,32 @@ 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 = {} + # 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 + assignment = None if actor._meta.model_name == 'user': if giving: - assignment, created = RoleUserAssignment.objects.get_or_create(user=actor, object_role=object_role) + try: + assignment = RoleUserAssignment.objects.get(user=actor, object_role=object_role) + except RoleUserAssignment.DoesNotExist: + assignment = RoleUserAssignment.objects.create(user=actor, object_role=object_role, **assignment_defaults) else: object_role.users.remove(actor) elif isinstance(actor, permission_registry.team_model): if giving: - assignment, created = RoleTeamAssignment.objects.get_or_create(team=actor, object_role=object_role) + try: + assignment = RoleTeamAssignment.objects.get(team=actor, object_role=object_role) + except RoleTeamAssignment.DoesNotExist: + assignment = RoleTeamAssignment.objects.create(team=actor, object_role=object_role, **assignment_defaults) else: object_role.teams.remove(actor) @@ -410,6 +449,14 @@ class AssignmentBase(ImmutableCommonModel, ObjectRoleFields): null=True, blank=True, help_text=_('The primary key of the object this assignment applies to; null value indicates system-wide assignment.') ) content_type = models.ForeignKey(DABContentType, on_delete=models.CASCADE, null=True, help_text=_("The content type this applies to.")) + # The object_created field can be used for a checksum-like purpose to verify nothing strange happened with the related object + object_created = models.DateTimeField(help_text=_("The created timestamp of related object, if applicable."), null=True) + # Define this with default to make it possible to backdate if necessary, for sync + created = models.DateTimeField( + default=timezone.now, + editable=False, + help_text=_("The date/time this resource was created."), + ) # object_role is internal, and not shown in serializer # content_type does not have a link, and ResourceType will be used in lieu sometime diff --git a/ansible_base/rbac/remote.py b/ansible_base/rbac/remote.py index 0a91fa382..d0fc20bfc 100644 --- a/ansible_base/rbac/remote.py +++ b/ansible_base/rbac/remote.py @@ -64,9 +64,11 @@ def __init__(self, ct: models.Model, abstract=False): class RemoteObject: """Placeholder for objects that live in another project.""" - def __init__(self, content_type: models.Model, object_id: Union[int, str], parent_reference=None): + def __init__(self, content_type: models.Model, object_id: Union[int, str], parent_reference=None, created=None): self.content_type = content_type self.object_id = object_id + # Allow tracking details of the object + self.created = created # Since object is remote, we do not have its properties here, so a pointer to the parent can be specified here self.parent_reference = parent_reference if not hasattr(self, '_meta'): diff --git a/ansible_base/rbac/service_api/serializers.py b/ansible_base/rbac/service_api/serializers.py index 8335a9391..6bd2a5cc9 100644 --- a/ansible_base/rbac/service_api/serializers.py +++ b/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): @@ -57,6 +57,10 @@ class BaseAssignmentSerializer(serializers.ModelSerializer): 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) + # Force created field to be writable + created = serializers.DateTimeField(required=False) + # Force object_created field to be writable + object_created = serializers.DateTimeField(required=False, allow_null=True) def to_representation(self, instance): # hack to surface content_object for ObjectIDAnsibleIDField @@ -131,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 = 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 = rd.give_global_permission( + actor, assignment_created=validated_data.get('created'), assignment_object_created=validated_data.get('object_created') + ) return assignment diff --git a/test_app/tests/rbac/remote/test_remote_assignment.py b/test_app/tests/rbac/remote/test_remote_assignment.py index ad837183b..cfd0656b8 100644 --- a/test_app/tests/rbac/remote/test_remote_assignment.py +++ b/test_app/tests/rbac/remote/test_remote_assignment.py @@ -150,3 +150,59 @@ def test_org_roles_same_type_different_service(rando, organization): ], f'User should have permission to exactly {service_name} resource' rds[service_name].remove_permission(rando, organization) + + +@pytest.mark.django_db +def test_object_created_field_local_model(rando, org_inv_rd, organization): + """ + Test that the object_created field is set to the local object's created timestamp + when creating an assignment for a local model like inventory. + + This test should FAIL because the object_created field doesn't exist yet. + """ + # Create the assignment + assignment = org_inv_rd.give_permission(rando, organization) + + # Verify the assignment was created + assert assignment.user == rando + assert assignment.role_definition == org_inv_rd + assert assignment.object_id == organization.pk + + # This should FAIL - the object_created field should be set to inventory.created + assert hasattr(assignment, 'object_created'), "Assignment should have an object_created field" + + assert assignment.object_created == organization.created, ( + f"object_created should be set to the organization's created timestamp. " + f"Expected: {organization.created}, but object_created field is missing or has wrong value" + ) + + +@pytest.mark.django_db +def test_object_created_field_remote_object(rando, foo_type, foo_rd): + """ + Test that the object_created field is properly handled when creating an assignment + for a remote object (stand-in object pattern). + + This test should FAIL because the object_created field doesn't exist yet. + """ + # Create a remote object stand-in + remote_foo = RemoteObject(content_type=foo_type, object_id=42) + + # Create the assignment + assignment = foo_rd.give_permission(rando, remote_foo) + + # Verify the assignment was created + assert assignment.user == rando + assert assignment.role_definition == foo_rd + assert assignment.object_id == 42 + assert isinstance(assignment.content_object, RemoteObject) + + # This should FAIL - the object_created field should exist and be handled for remote objects + assert hasattr(assignment, 'object_created'), "Assignment should have an object_created field even for remote objects" + + # For remote objects, the object_created field might be None or set to a default value + # since we don't have the actual remote object's creation timestamp + # The exact behavior will depend on implementation, but the field should exist + assert assignment.object_created is not None or assignment.object_created is None, ( + f"object_created field should exist for remote objects. " f"Current value: {getattr(assignment, 'object_created', 'FIELD_MISSING')}" + ) diff --git a/test_app/tests/rbac/remote/test_service_api.py b/test_app/tests/rbac/remote/test_service_api.py index 074a67323..6296946d5 100644 --- a/test_app/tests/rbac/remote/test_service_api.py +++ b/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 +from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleUserAssignment from test_app.models import Team, User @@ -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 @@ -491,3 +442,133 @@ def test_serializer_allows_null_values_in_validation(self, admin_api_client, ran # Verify that created_by is None in validated_data when null is passed 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 + in responses but not applied when creating assignments via POST to /assign/. + + This test should FAIL, showing that custom timestamps are ignored and auto-generated instead. + """ + from datetime import datetime, timezone + + from django.utils.dateparse import parse_datetime + + url = get_relative_url('serviceuserassignment-assign') + + creator_user = User.objects.create(username='timestamp_creator') + + # Set a specific timestamp that's different from "now" + custom_timestamp = datetime(2023, 1, 15, 10, 30, 45, tzinfo=timezone.utc) + custom_timestamp_str = custom_timestamp.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), + "created": custom_timestamp_str, + "from_service": "test_service", + } + + 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) + + # Test if the custom timestamp was properly set + expected_created = custom_timestamp + actual_created = assignment.created + + # This should FAIL, demonstrating the field sync issue + assert actual_created == expected_created, ( + f"FIELD SYNC ISSUE: Expected created timestamp '{expected_created}' but got '{actual_created}'. " + f"The 'created' field is displayed in responses but not applied from POST data." + ) + + # Verify response contains the timestamp field (showing it's "displayed") + 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 +@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. 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') + + 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": 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 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=role_def, object_id=target_object.pk) + + # 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 {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'] == 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 + 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}'" 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 index 771c63b0b..d909cba46 100644 --- 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'])