diff --git a/ansible_base/activitystream/models/entry.py b/ansible_base/activitystream/models/entry.py index 6de1c5aae..07d042d2d 100644 --- a/ansible_base/activitystream/models/entry.py +++ b/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 diff --git a/ansible_base/rbac/models/dummy_models.py b/ansible_base/rbac/models/dummy_models.py new file mode 100644 index 000000000..9f80b63b4 --- /dev/null +++ b/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 diff --git a/ansible_base/rbac/models/role.py b/ansible_base/rbac/models/role.py index ee90a1446..3a8048352 100644 --- a/ansible_base/rbac/models/role.py +++ b/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'] diff --git a/test_app/tests/rbac/test_dummy_models.py b/test_app/tests/rbac/test_dummy_models.py new file mode 100644 index 000000000..e27f36443 --- /dev/null +++ b/test_app/tests/rbac/test_dummy_models.py @@ -0,0 +1,45 @@ +"""Test dummy models for optional django-ansible-base apps.""" + +import pytest + +from ansible_base.rbac.models.dummy_models import DummyAuditableModel + + +class TestDummyModel(DummyAuditableModel): + """Test model using DummyAuditableModel for testing.""" + + class Meta: + app_label = 'test_app' + + +@pytest.mark.django_db +def test_dummy_auditable_model_interface(): + """Test that DummyAuditableModel provides expected interface without crashing.""" + # Create test instance + test_obj = TestDummyModel() + + # Test activity_stream_entries property + entries = test_obj.activity_stream_entries + assert entries.count() == 0, "Dummy model should return empty count" + assert entries.last() is None, "Dummy model should return None for last()" + assert list(entries) == [], "Dummy model should return empty list" + assert entries.all() == entries, "all() should return self" + assert entries.order_by('id') == entries, "order_by() should return self" + + # Test class attributes + assert hasattr(test_obj, 'activity_stream_excluded_field_names') + assert test_obj.activity_stream_excluded_field_names == [] + assert hasattr(test_obj, 'activity_stream_limit_field_names') + assert test_obj.activity_stream_limit_field_names == [] + + # Test extra_related_fields method + assert test_obj.extra_related_fields(None) == {} + + +def test_dummy_model_import_safety(): + """Test that DummyAuditableModel can be imported safely.""" + # The main purpose is that this import doesn't crash AWX/EDA + from ansible_base.rbac.models.dummy_models import DummyAuditableModel + + # Basic verification that it's properly configured + assert DummyAuditableModel._meta.abstract is True diff --git a/test_app/tests/rbac/test_rbac_activity_stream.py b/test_app/tests/rbac/test_rbac_activity_stream.py new file mode 100644 index 000000000..7c0917e90 --- /dev/null +++ b/test_app/tests/rbac/test_rbac_activity_stream.py @@ -0,0 +1,129 @@ +"""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')