mirror of
https://github.com/ansible/awx.git
synced 2024-11-02 09:51:09 +03:00
ActivityStream access checking test for most resources
This commit is contained in:
parent
4e2e4667de
commit
9cc8ae9329
@ -12,8 +12,7 @@ def mock_feature_enabled(feature, bypass_database=None):
|
||||
return True
|
||||
|
||||
@pytest.fixture
|
||||
def activity_stream_entry(organization, rando):
|
||||
rando.roles.add(organization.admin_role)
|
||||
def activity_stream_entry(organization, org_admin):
|
||||
return ActivityStream.objects.filter(organization__pk=organization.pk, operation='associate').first()
|
||||
|
||||
@pytest.mark.skipif(not getattr(settings, 'ACTIVITY_STREAM_ENABLED', True), reason="Activity stream not enabled")
|
||||
@ -70,9 +69,9 @@ def test_middleware_actor_added(monkeypatch, post, get, user):
|
||||
@pytest.mark.skipif(not getattr(settings, 'ACTIVITY_STREAM_ENABLED', True), reason="Activity stream not enabled")
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
def test_rbac_stream_resource_roles(activity_stream_entry, organization, rando):
|
||||
def test_rbac_stream_resource_roles(activity_stream_entry, organization, org_admin):
|
||||
|
||||
assert activity_stream_entry.user.first() == rando
|
||||
assert activity_stream_entry.user.first() == org_admin
|
||||
assert activity_stream_entry.organization.first() == organization
|
||||
assert activity_stream_entry.role.first() == organization.admin_role
|
||||
assert activity_stream_entry.object_relationship_type == 'awx.main.models.organization.Organization.admin_role'
|
||||
@ -80,9 +79,9 @@ def test_rbac_stream_resource_roles(activity_stream_entry, organization, rando):
|
||||
@pytest.mark.skipif(not getattr(settings, 'ACTIVITY_STREAM_ENABLED', True), reason="Activity stream not enabled")
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
def test_rbac_stream_user_roles(activity_stream_entry, organization, rando):
|
||||
def test_rbac_stream_user_roles(activity_stream_entry, organization, org_admin):
|
||||
|
||||
assert activity_stream_entry.user.first() == rando
|
||||
assert activity_stream_entry.user.first() == org_admin
|
||||
assert activity_stream_entry.organization.first() == organization
|
||||
assert activity_stream_entry.role.first() == organization.admin_role
|
||||
assert activity_stream_entry.object_relationship_type == 'awx.main.models.organization.Organization.admin_role'
|
||||
@ -91,8 +90,8 @@ def test_rbac_stream_user_roles(activity_stream_entry, organization, rando):
|
||||
@pytest.mark.activity_stream_access
|
||||
@pytest.mark.skipif(not getattr(settings, 'ACTIVITY_STREAM_ENABLED', True), reason="Activity stream not enabled")
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
def test_stream_access_cant_change(activity_stream_entry, organization, rando):
|
||||
access = ActivityStreamAccess(rando)
|
||||
def test_stream_access_cant_change(activity_stream_entry, organization, org_admin):
|
||||
access = ActivityStreamAccess(org_admin)
|
||||
# These should always return false because the activity stream can not be edited
|
||||
assert not access.can_add(activity_stream_entry)
|
||||
assert not access.can_change(activity_stream_entry, {'organization': None})
|
||||
@ -102,13 +101,33 @@ def test_stream_access_cant_change(activity_stream_entry, organization, rando):
|
||||
@pytest.mark.activity_stream_access
|
||||
@pytest.mark.skipif(not getattr(settings, 'ACTIVITY_STREAM_ENABLED', True), reason="Activity stream not enabled")
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
def test_stream_queryset(activity_stream_entry, organization, rando, user_project):
|
||||
# user_project and its owner has no relationship to the user rando
|
||||
rando_access = ActivityStreamAccess(rando)
|
||||
queryset = rando_access.get_queryset()
|
||||
def test_stream_queryset_hides_shows_items(
|
||||
activity_stream_entry, organization, user, org_admin,
|
||||
project, org_credential, inventory, label, deploy_jobtemplate,
|
||||
notification_template, group, host, team):
|
||||
# this user is not in any organizations and should not see any resource activity
|
||||
no_access_user = user('no-access-user', False)
|
||||
queryset = ActivityStreamAccess(no_access_user).get_queryset()
|
||||
|
||||
# Rando can see that he was appointed organization admin
|
||||
assert queryset.filter(organization__pk=organization.pk, operation='associate')
|
||||
assert not queryset.filter(project__pk=project.pk)
|
||||
assert not queryset.filter(credential__pk=org_credential.pk)
|
||||
assert not queryset.filter(inventory__pk=inventory.pk)
|
||||
assert not queryset.filter(label__pk=label.pk)
|
||||
assert not queryset.filter(job_template__pk=deploy_jobtemplate.pk)
|
||||
assert not queryset.filter(group__pk=group.pk)
|
||||
assert not queryset.filter(host__pk=host.pk)
|
||||
assert not queryset.filter(team__pk=team.pk)
|
||||
assert not queryset.filter(notification_template__pk=notification_template.pk)
|
||||
|
||||
# Rando can not see anything related to the project
|
||||
assert not queryset.filter(project__pk=user_project.pk)
|
||||
# Organization admin should be able to see most things in the ActivityStream
|
||||
queryset = ActivityStreamAccess(org_admin).get_queryset()
|
||||
|
||||
assert queryset.filter(project__pk=project.pk, operation='create').count() == 1
|
||||
assert queryset.filter(credential__pk=org_credential.pk, operation='create').count() == 1
|
||||
assert queryset.filter(inventory__pk=inventory.pk, operation='create').count() == 1
|
||||
assert queryset.filter(label__pk=label.pk, operation='create').count() == 1
|
||||
assert queryset.filter(job_template__pk=deploy_jobtemplate.pk, operation='create').count() == 1
|
||||
assert queryset.filter(group__pk=group.pk, operation='create').count() == 1
|
||||
assert queryset.filter(host__pk=host.pk, operation='create').count() == 1
|
||||
assert queryset.filter(team__pk=team.pk, operation='create').count() == 1
|
||||
assert queryset.filter(notification_template__pk=notification_template.pk, operation='create').count() == 1
|
||||
|
@ -167,6 +167,11 @@ def credential():
|
||||
def machine_credential():
|
||||
return Credential.objects.create(name='machine-cred', kind='ssh', username='test_user', password='pas4word')
|
||||
|
||||
@pytest.fixture
|
||||
def org_credential(organization, credential):
|
||||
credential.owner_role.parents.add(organization.admin_role)
|
||||
return credential
|
||||
|
||||
@pytest.fixture
|
||||
def inventory(organization):
|
||||
return organization.inventories.create(name="test-inv")
|
||||
@ -233,7 +238,7 @@ def organizations(instance):
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def group(inventory):
|
||||
def group_factory(inventory):
|
||||
def g(name):
|
||||
try:
|
||||
return Group.objects.get(name=name, inventory=inventory)
|
||||
@ -242,8 +247,8 @@ def group(inventory):
|
||||
return g
|
||||
|
||||
@pytest.fixture
|
||||
def hosts(group):
|
||||
group1 = group('group-1')
|
||||
def hosts(group_factory):
|
||||
group1 = group_factory('group-1')
|
||||
|
||||
def rf(host_count=1):
|
||||
hosts = []
|
||||
@ -256,6 +261,14 @@ def hosts(group):
|
||||
return hosts
|
||||
return rf
|
||||
|
||||
@pytest.fixture
|
||||
def group(inventory):
|
||||
return inventory.groups.create(name='single-group')
|
||||
|
||||
@pytest.fixture
|
||||
def host(group, inventory):
|
||||
return group.hosts.create(name='single-host', inventory=inventory)
|
||||
|
||||
@pytest.fixture
|
||||
def permissions():
|
||||
return {
|
||||
@ -406,8 +419,8 @@ def options():
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fact_scans(group, fact_ansible_json, fact_packages_json, fact_services_json):
|
||||
group1 = group('group-1')
|
||||
def fact_scans(group_factory, fact_ansible_json, fact_packages_json, fact_services_json):
|
||||
group1 = group_factory('group-1')
|
||||
|
||||
def rf(fact_scans=1, timestamp_epoch=timezone.now()):
|
||||
facts_json = {}
|
||||
|
@ -3,10 +3,10 @@ import pytest
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_inventory_source_notification_on_cloud_only(get, post, group, user, notification_template):
|
||||
def test_inventory_source_notification_on_cloud_only(get, post, group_factory, user, notification_template):
|
||||
u = user('admin', True)
|
||||
g_cloud = group('cloud')
|
||||
g_not = group('not_cloud')
|
||||
g_cloud = group_factory('cloud')
|
||||
g_not = group_factory('not_cloud')
|
||||
cloud_is = g_cloud.inventory_source
|
||||
not_is = g_not.inventory_source
|
||||
cloud_is.source = 'ec2'
|
||||
|
@ -81,12 +81,12 @@ def test_team_symantics(organization, team, alice):
|
||||
assert alice not in organization.auditor_role
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_auto_m2m_adjustments(organization, inventory, group, alice):
|
||||
def test_auto_m2m_adjustments(organization, inventory, group_factory, alice):
|
||||
'Ensures the auto role reparenting is working correctly through m2m maps'
|
||||
g1 = group(name='g1')
|
||||
g1 = group_factory(name='g1')
|
||||
g1.admin_role.members.add(alice)
|
||||
assert alice in g1.admin_role
|
||||
g2 = group(name='g2')
|
||||
g2 = group_factory(name='g2')
|
||||
assert alice not in g2.admin_role
|
||||
|
||||
g2.parents.add(g1)
|
||||
|
@ -170,11 +170,11 @@ def test_inventory_executor(inventory, permissions, user, team):
|
||||
assert team.member_role.is_ancestor_of(inventory.execute_role)
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_group_parent_admin(group, permissions, user):
|
||||
def test_group_parent_admin(group_factory, permissions, user):
|
||||
u = user('admin', False)
|
||||
parent1 = group('parent-1')
|
||||
parent2 = group('parent-2')
|
||||
childA = group('child-1')
|
||||
parent1 = group_factory('parent-1')
|
||||
parent2 = group_factory('parent-2')
|
||||
childA = group_factory('child-1')
|
||||
|
||||
parent1.admin_role.members.add(u)
|
||||
assert u in parent1.admin_role
|
||||
@ -230,11 +230,11 @@ def test_access_auditor(organization, inventory, user):
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_host_access(organization, inventory, user, group):
|
||||
def test_host_access(organization, inventory, user, group_factory):
|
||||
other_inventory = organization.inventories.create(name='other-inventory')
|
||||
inventory_admin = user('inventory_admin', False)
|
||||
my_group = group('my-group')
|
||||
not_my_group = group('not-my-group')
|
||||
my_group = group_factory('my-group')
|
||||
not_my_group = group_factory('not-my-group')
|
||||
group_admin = user('group_admin', False)
|
||||
|
||||
inventory_admin_access = HostAccess(inventory_admin)
|
||||
|
Loading…
Reference in New Issue
Block a user