AAP-52229 Add AuditableModel inheritance to RBAC assignment models

Author: arrestle

ansible_base/activitystream/models/entry.py

@@ -63,6 +63,13 @@ class Meta:
     )
 
     def __str__(self):
+        # Enhanced display for RBAC role assignments
+        if self.content_type and self.content_type.model.lower() in ['roleuserassignment', 'roleteamassignment']:
+            operation_text = self.get_operation_display()
+            created_by_text = str(self.created_by) if self.created_by else "Unknown"
+            return f'[{self.created}] Role assignment {operation_text.lower()} by {created_by_text}'
+        
+        # Standard format for other entry types
         return f'[{self.created}] {self.get_operation_display()} by {self.created_by}: {self.content_type} {self.object_id}'
 
     @functools.cached_property

ansible_base/rbac/models/dummy_models.py

@@ -0,0 +1,52 @@
+"""
+Dummy models for optional django-ansible-base apps.
+
+These provide no-op implementations when optional apps are not installed,
+preventing import crashes while maintaining interface compatibility.
+"""
+
+from django.db import models
+
+
+class DummyAuditableModel(models.Model):
+    """
+    Dummy AuditableModel for services without activitystream app.
+
+    Provides the same interface as the real AuditableModel but with no
+    activity logging functionality. This prevents import crashes in services
+    like AWX/EDA that don't include 'ansible_base.activitystream' in INSTALLED_APPS.
+    """
+
+    activity_stream_excluded_field_names = []
+    activity_stream_limit_field_names = []
+
+    @property
+    def activity_stream_entries(self):
+        """Return empty queryset for dummy model."""
+
+        # Can't import Entry directly - would crash AWX/EDA
+        # Return a minimal QuerySet-like object that supports count() and last()
+        class EmptyActivityStream:
+            def count(self):
+                return 0
+
+            def last(self):
+                return None
+
+            def all(self):
+                return self
+
+            def order_by(self, *args):
+                return self
+
+            def __iter__(self):
+                return iter([])
+
+        return EmptyActivityStream()
+
+    def extra_related_fields(self, request):
+        """Return empty dict for dummy model."""
+        return {}
+
+    class Meta:
+        abstract = True

ansible_base/rbac/models/role.py

@@ -3,6 +3,8 @@
 from typing import Optional, Type, Union
 from uuid import UUID
 
+from django.apps import apps
+
 # Django
 from django.conf import settings
 from django.db import connection, models, transaction
@@ -17,6 +19,17 @@
 # ansible_base lib functions
 from ansible_base.lib.abstract_models.common import CommonModel, ImmutableCommonModel
 
+# Conditional import for activity stream support
+#
+# The activitystream app is optional - services choose whether to enable it:
+# - Gateway/test_app: Include 'ansible_base.activitystream' in INSTALLED_APPS
+# - AWX/EDA/Hub: Don't include activitystream app (use legacy activity systems)
+if apps.is_installed('ansible_base.activitystream'):
+    from ansible_base.activitystream.models import AuditableModel
+else:
+    from .dummy_models import DummyAuditableModel as AuditableModel
+
+
 # ansible_base RBAC logic imports
 from ansible_base.lib.utils.models import is_add_perm
 from ansible_base.rbac.permission_registry import permission_registry
@@ -128,7 +141,6 @@ def get_or_create(self, permissions=(), defaults=None, **kwargs):
         return super().get_or_create(defaults=defaults, **kwargs)
 
     def create_from_permissions(self, permissions=(), **kwargs):
-        "Create from a list of text-type permissions and do validation"
         perm_list: list[str] = []
         for str_perm in permissions:
             if '.' in str_perm:
@@ -434,7 +446,7 @@ def save(self, *args, **kwargs):
         return super().save(*args, **kwargs)
 
 
-class RoleUserAssignment(AssignmentBase):
+class RoleUserAssignment(AssignmentBase, AuditableModel):
     role_definition = models.ForeignKey(
         RoleDefinition,
         on_delete=models.CASCADE,
@@ -446,6 +458,9 @@ class RoleUserAssignment(AssignmentBase):
     )
     router_basename = 'roleuserassignment'
 
+    # Exclude object_role from activity stream - it's an internal implementation detail
+    activity_stream_excluded_field_names = ['object_role']
+
     class Meta:
         app_label = 'dab_rbac'
         ordering = ['id']
@@ -460,7 +475,7 @@ def actor(self):
         return self.user
 
 
-class RoleTeamAssignment(AssignmentBase):
+class RoleTeamAssignment(AssignmentBase, AuditableModel):
     role_definition = models.ForeignKey(
         RoleDefinition,
         on_delete=models.CASCADE,
@@ -472,6 +487,9 @@ class RoleTeamAssignment(AssignmentBase):
     )
     router_basename = 'roleteamassignment'
 
+    # Exclude object_role from activity stream - it's an internal implementation detail
+    activity_stream_excluded_field_names = ['object_role']
+
     class Meta:
         app_label = 'dab_rbac'
         ordering = ['id']

test_app/tests/rbac/test_rbac_activity_stream.py

@@ -0,0 +1,130 @@
+"""Test RBAC activity stream functionality."""
+
+import uuid
+
+import pytest
+from crum import impersonate
+from django.apps import apps
+from django.contrib.auth import get_user_model
+
+from ansible_base.rbac.models import DABContentType, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
+
+User = get_user_model()
+
+
+def verify_activity_entry_fields(entry, operation, admin_user, actor_id, role_def_id, actor_field):
+    """Helper to verify activity entry has correct fields."""
+    assert entry.operation == operation
+    assert entry.created_by == admin_user
+    assert entry.changes, f"{operation.title()} entry should have changes recorded"
+
+    # Get the appropriate fields dict based on operation
+    fields_dict = entry.changes['added_fields'] if operation == 'create' else entry.changes['removed_fields']
+
+    # Verify required fields are present with correct values
+    assert actor_field in fields_dict
+    assert 'role_definition' in fields_dict
+    assert str(actor_id) == fields_dict[actor_field]
+    assert str(role_def_id) == fields_dict['role_definition']
+
+
+@pytest.mark.skipif(not apps.is_installed('ansible_base.activitystream'), reason="Activity stream tests only run when activitystream app is installed")
+@pytest.mark.django_db
+def test_role_user_assignment_activity_stream_lifecycle(system_user, admin_user, organization):
+    """Test role assignment create and delete both create proper activity entries."""
+    # Create unique test user, role, and org with distinctive names
+    test_uuid = str(uuid.uuid4())[:8]
+    unique_username = f'test_rbac_user_{test_uuid}'
+    unique_role_name = f'TestRole_ActivityStream_{test_uuid}'
+    unique_org_name = f'TestOrg_ActivityStream_{test_uuid}'
+
+    test_user = User.objects.create_user(username=unique_username, email=f'{unique_username}@example.com')
+
+    # Create unique organization for this test
+    from test_app.models import Organization
+
+    test_org = Organization.objects.create(name=unique_org_name)
+
+    ct = DABContentType.objects.get_for_model(test_org)
+    role_def = RoleDefinition.objects.create(name=unique_role_name, content_type=ct)
+
+    # Create assignment (admin assigns role to user)
+    with impersonate(admin_user):
+        assignment = RoleUserAssignment.objects.create(user=test_user, role_definition=role_def, content_object=test_org, created_by=admin_user)
+
+    # Verify CREATE entry
+    assert assignment.activity_stream_entries.count() == 1
+    create_entry = assignment.activity_stream_entries.last()
+    verify_activity_entry_fields(create_entry, 'create', admin_user, test_user.id, role_def.id, 'user')
+    
+    # Verify enhanced string representation
+    entry_str = str(create_entry)
+    assert "created" in entry_str.lower()
+    assert str(admin_user) in entry_str
+
+
+    # Delete assignment and verify DELETE entry
+    assignment_id = assignment.id
+    with impersonate(admin_user):
+        assignment.delete()
+
+    # Query entries directly since assignment pk=None after delete
+    from django.contrib.contenttypes.models import ContentType
+
+    from ansible_base.activitystream.models import Entry
+
+    assignment_ct = ContentType.objects.get_for_model(RoleUserAssignment)
+    assignment_entries = Entry.objects.filter(content_type=assignment_ct, object_id=str(assignment_id)).order_by('id')
+
+    assert assignment_entries.count() == 2
+    delete_entry = assignment_entries.last()
+    verify_activity_entry_fields(delete_entry, 'delete', admin_user, test_user.id, role_def.id, 'user')
+    
+    # Verify enhanced string representation for delete
+    delete_str = str(delete_entry)
+    assert "deleted" in delete_str.lower()
+    assert str(admin_user) in delete_str
+
+
+@pytest.mark.skipif(not apps.is_installed('ansible_base.activitystream'), reason="Activity stream tests only run when activitystream app is installed")
+@pytest.mark.django_db
+def test_role_team_assignment_activity_stream(admin_user, team, organization):
+    """Test team role assignment creates activity entries."""
+    # Create unique role and org names for isolation
+    test_uuid = str(uuid.uuid4())[:8]
+    unique_role_name = f'TestTeamRole_{test_uuid}'
+    unique_org_name = f'TestTeamOrg_{test_uuid}'
+
+    # Create unique organization for this test
+    from test_app.models import Organization
+
+    test_org = Organization.objects.create(name=unique_org_name)
+
+    ct = DABContentType.objects.get_for_model(test_org)
+    role_def = RoleDefinition.objects.create(name=unique_role_name, content_type=ct)
+
+    # Create team assignment
+    with impersonate(admin_user):
+        assignment = RoleTeamAssignment.objects.create(team=team, role_definition=role_def, content_object=test_org, created_by=admin_user)
+
+    # Verify CREATE entry
+    assert assignment.activity_stream_entries.count() == 1
+    create_entry = assignment.activity_stream_entries.last()
+    verify_activity_entry_fields(create_entry, 'create', admin_user, team.id, role_def.id, 'team')
+
+    # Delete assignment and verify DELETE entry
+    assignment_id = assignment.id
+    with impersonate(admin_user):
+        assignment.delete()
+
+    # Query entries directly since assignment pk=None after delete
+    from django.contrib.contenttypes.models import ContentType
+
+    from ansible_base.activitystream.models import Entry
+
+    assignment_ct = ContentType.objects.get_for_model(RoleTeamAssignment)
+    assignment_entries = Entry.objects.filter(content_type=assignment_ct, object_id=str(assignment_id)).order_by('id')
+
+    assert assignment_entries.count() == 2
+    delete_entry = assignment_entries.last()
+    verify_activity_entry_fields(delete_entry, 'delete', admin_user, team.id, role_def.id, 'team')