From 4aeb0e6386b6e135d00cc9afa903571831bdc387 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Tue, 29 Jul 2025 12:11:10 -0400 Subject: [PATCH 1/4] Synchronize the created and add object_created field --- .../0004_remote_permissions_additions.py | 23 ++++++++ ansible_base/rbac/models/role.py | 53 ++++++++++++++--- ansible_base/rbac/remote.py | 4 +- ansible_base/rbac/service_api/serializers.py | 6 +- .../rbac/remote/test_remote_assignment.py | 58 +++++++++++++++++++ .../tests/rbac/remote/test_service_api.py | 50 +++++++++++++++- 6 files changed, 182 insertions(+), 12 deletions(-) diff --git a/ansible_base/rbac/migrations/0004_remote_permissions_additions.py b/ansible_base/rbac/migrations/0004_remote_permissions_additions.py index c7159b886..d2159ce8a 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): @@ -159,6 +160,28 @@ class Migration(migrations.Migration): model_name='roleevaluationuuid', name='dab_rbac_ro_role_id_4fe905_idx', ), + # 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.'), + ), # Fields unique to DAB RBAC and not generally shared with ContentType or Permission migrations.AddField( model_name='dabcontenttype', diff --git a/ansible_base/rbac/models/role.py b/ansible_base/rbac/models/role.py index 07c49d641..c701b0b94 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,13 @@ 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): + return self.give_or_remove_global_permission(actor, giving=True, assignment_created=assignment_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): if giving and (self.content_type is not None): raise ValidationError('Role definition content type must be null to assign globally') @@ -202,6 +216,8 @@ 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 assignment, _ = cls.objects.get_or_create(**kwargs) else: assignment = cls.objects.filter(**kwargs).first() @@ -221,8 +237,8 @@ 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): + return self.give_or_remove_permission(actor, content_object, giving=True, assignment_created=assignment_created) def remove_permission(self, actor, content_object): return self.give_or_remove_permission(actor, content_object, giving=False) @@ -247,7 +263,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): "Shortcut method to do whatever needed to give user or team these permissions" validate_assignment(self, actor, content_object) @@ -278,15 +294,28 @@ 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 + 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 +439,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..7dfe340cb 100644 --- a/ansible_base/rbac/service_api/serializers.py +++ b/ansible_base/rbac/service_api/serializers.py @@ -57,6 +57,8 @@ 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) def to_representation(self, instance): # hack to surface content_object for ObjectIDAnsibleIDField @@ -131,10 +133,10 @@ 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')) else: with transaction.atomic(): - assignment = rd.give_global_permission(actor) + assignment = rd.give_global_permission(actor, assignment_created=validated_data.get('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..26bc8ee87 100644 --- a/test_app/tests/rbac/remote/test_remote_assignment.py +++ b/test_app/tests/rbac/remote/test_remote_assignment.py @@ -150,3 +150,61 @@ 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. + """ + from datetime import datetime, timezone + + # 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..e9cf9d572 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, RoleTeamAssignment, RoleUserAssignment from test_app.models import Team, User @@ -491,3 +491,51 @@ 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}'" From da63deaf4558d480021046cf44deaefc4d4b6f9c Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Tue, 9 Sep 2025 22:38:04 -0400 Subject: [PATCH 2/4] Handle object_created and created field transmissions --- ansible_base/rbac/models/role.py | 28 +++-- ansible_base/rbac/service_api/serializers.py | 12 +- .../rbac/remote/test_remote_assignment.py | 2 - .../tests/rbac/remote/test_service_api.py | 114 +++++++++++++++++- 4 files changed, 141 insertions(+), 15 deletions(-) diff --git a/ansible_base/rbac/models/role.py b/ansible_base/rbac/models/role.py index c701b0b94..8166c1570 100644 --- a/ansible_base/rbac/models/role.py +++ b/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 diff --git a/ansible_base/rbac/service_api/serializers.py b/ansible_base/rbac/service_api/serializers.py index 7dfe340cb..5112fdb0e 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): @@ -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 diff --git a/test_app/tests/rbac/remote/test_remote_assignment.py b/test_app/tests/rbac/remote/test_remote_assignment.py index 26bc8ee87..cfd0656b8 100644 --- a/test_app/tests/rbac/remote/test_remote_assignment.py +++ b/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) diff --git a/test_app/tests/rbac/remote/test_service_api.py b/test_app/tests/rbac/remote/test_service_api.py index e9cf9d572..5548768a3 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, 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}'" From e530791a7960ac5cf04ec56028ded8fdd4af9f0a Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 10 Sep 2025 11:22:06 -0400 Subject: [PATCH 3/4] Add data migration for object_created and test --- .../0005_remote_permissions_data.py | 8 + ansible_base/rbac/migrations/_utils.py | 68 ++++++ ansible_base/rbac/service_api/serializers.py | 2 +- .../tests/rbac/remote/test_service_api.py | 221 ++++++------------ .../test_resources_api_rest_client.py | 1 + 5 files changed, 149 insertions(+), 151 deletions(-) diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py index a0f7852a0..ac5246674 100644 --- 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 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/service_api/serializers.py b/ansible_base/rbac/service_api/serializers.py index 5112fdb0e..6bd2a5cc9 100644 --- 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 index 5548768a3..6296946d5 100644 --- 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,11 +494,19 @@ 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 @@ -555,34 +514,46 @@ def test_service_assignment_object_created_timestamp_sync(admin_api_client, rand 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') @@ -590,7 +561,7 @@ def test_service_assignment_object_created_timestamp_sync(admin_api_client, rand 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 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']) From b1e9307a628a0542ad8d36e482446c05da7351df Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 10 Sep 2025 13:29:17 -0400 Subject: [PATCH 4/4] Move object_created to new migration step --- .../0004_remote_permissions_additions.py | 22 --------- .../0005_remote_permissions_data.py | 7 --- .../migrations/0009_object_created_field.py | 45 +++++++++++++++++++ 3 files changed, 45 insertions(+), 29 deletions(-) create mode 100644 ansible_base/rbac/migrations/0009_object_created_field.py diff --git a/ansible_base/rbac/migrations/0004_remote_permissions_additions.py b/ansible_base/rbac/migrations/0004_remote_permissions_additions.py index d2159ce8a..cb00b6c42 100644 --- a/ansible_base/rbac/migrations/0004_remote_permissions_additions.py +++ b/ansible_base/rbac/migrations/0004_remote_permissions_additions.py @@ -160,28 +160,6 @@ class Migration(migrations.Migration): model_name='roleevaluationuuid', name='dab_rbac_ro_role_id_4fe905_idx', ), - # 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.'), - ), # Fields unique to DAB RBAC and not generally shared with ContentType or Permission migrations.AddField( model_name='dabcontenttype', diff --git a/ansible_base/rbac/migrations/0005_remote_permissions_data.py b/ansible_base/rbac/migrations/0005_remote_permissions_data.py index ac5246674..028ffecd9 100644 --- a/ansible_base/rbac/migrations/0005_remote_permissions_data.py +++ b/ansible_base/rbac/migrations/0005_remote_permissions_data.py @@ -1,5 +1,4 @@ # 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 @@ -24,11 +23,6 @@ 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') @@ -75,5 +69,4 @@ 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/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), + ]