avoid multi-field implicit role activity stream entries

awx/main/signals.py

@@ -199,10 +199,21 @@ def rbac_activity_stream(instance, sender, **kwargs):
                     # Ignore any singleton parents we find. If the parent for the role
                     # matches any of the implicit parents we find, skip recording the activity stream.
                     for ip in implicit_parents:
-                        if '.' not in ip and 'singleton:' not in ip:
-                            ip = instance.content_type.name + "." + ip
-                        if parent == ip:
-                            return
+                        if '.' in ip:
+                            obj = instance.content_object
+                            for next_field in ip.split('.')[:-1]:
+                                obj = getattr(obj, next_field)
+                                if obj is None:
+                                    return
+                            if role == getattr(obj, ip.split('.')[-1]):
+                                return
+                        elif 'singleton:' in ip:
+                            if parent == ip:
+                                return
+                        else:
+                            # Direct field on the content object
+                            if role == getattr(instance.content_object, ip):
+                                return
             else:
                 role = instance
             instance = instance.content_object

awx/main/tests/functional/models/test_activity_stream.py

@@ -1,13 +1,82 @@
 import pytest
 
 # AWX models
-from awx.main.models.organization import Organization
-from awx.main.models import ActivityStream
+from awx.main.models import ActivityStream, Organization, JobTemplate
 
 
 
-@pytest.mark.django_db
-def test_activity_stream_create_entries():
-    Organization.objects.create(name='test-organization2')
-    assert ActivityStream.objects.filter(organization__isnull=False).count() == 1
+class TestImplicitRolesOmitted:
+    '''
+    Test that there is exactly 1 "create" entry in the activity stream for
+    common items in the system.
+    These tests will fail if `rbac_activity_stream` creates
+    false-positive entries.
+    '''
+
+    @pytest.mark.django_db
+    def test_activity_stream_create_organization(self):
+        Organization.objects.create(name='test-organization2')
+        qs = ActivityStream.objects.filter(organization__isnull=False)
+        assert qs.count() == 1
+        assert qs[0].operation == 'create'
+
+    @pytest.mark.django_db
+    def test_activity_stream_delete_organization(self):
+        org = Organization.objects.create(name='gYSlNSOFEW')
+        org.delete()
+        qs = ActivityStream.objects.filter(changes__icontains='gYSlNSOFEW')
+        assert qs.count() == 2
+        assert qs[1].operation == 'delete'
+
+    @pytest.mark.django_db
+    def test_activity_stream_create_JT(self, project, inventory, credential):
+        JobTemplate.objects.create(
+            name='test-jt',
+            project=project,
+            inventory=inventory,
+            credential=credential
+        )
+        qs = ActivityStream.objects.filter(job_template__isnull=False)
+        assert qs.count() == 1
+        assert qs[0].operation == 'create'
+
+    @pytest.mark.django_db
+    def test_activity_stream_create_inventory(self, organization):
+        organization.inventories.create(name='test-inv')
+        qs = ActivityStream.objects.filter(inventory__isnull=False)
+        assert qs.count() == 1
+        assert qs[0].operation == 'create'
+
+    @pytest.mark.django_db
+    def test_activity_stream_create_credential(self, organization):
+        organization.inventories.create(name='test-inv')
+        qs = ActivityStream.objects.filter(inventory__isnull=False)
+        assert qs.count() == 1
+        assert qs[0].operation == 'create'
+
+
+class TestRolesAssociationEntries:
+    '''
+    Test that non-implicit role associations have a corresponding
+    activity stream entry.
+    These tests will fail if `rbac_activity_stream` skipping logic
+    finds a false-negative.
+    '''
+
+    @pytest.mark.django_db
+    def test_non_implicit_associations_are_recorded(self, project):
+        org2 = Organization.objects.create(name='test-organization2')
+        project.admin_role.parents.add(org2.admin_role)
+        assert ActivityStream.objects.filter(
+            role=org2.admin_role,
+            organization=org2,
+            project=project
+        ).count() == 1
+
+    @pytest.mark.django_db
+    def test_model_associations_are_recorded(self, organization):
+        proj1 = organization.projects.create(name='proj1')
+        proj2 = organization.projects.create(name='proj2')
+        proj2.use_role.parents.add(proj1.admin_role)
+        assert ActivityStream.objects.filter(role=proj1.admin_role, project=proj2).count() == 1