diff --git a/awx/main/signals.py b/awx/main/signals.py index 20dcd2dcd6..9faf3b991d 100644 --- a/awx/main/signals.py +++ b/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 diff --git a/awx/main/tests/functional/models/test_activity_stream.py b/awx/main/tests/functional/models/test_activity_stream.py index b6e63a4377..b9c1611f05 100644 --- a/awx/main/tests/functional/models/test_activity_stream.py +++ b/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