External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-14 11:20:39 -03:30
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1238 from wwitzel3/rbac

Browse Source 
RBAC Updates
This commit is contained in:
Akita Noek 2016-03-14 14:43:28 -04:00
commit 44bfc44555
15 changed files with 224 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
awx/main/migrations/0007_v300_rbac_changes.py
View File

@ -18,6 +18,22 @@ class Migration(migrations.Migration):
]
operations = [
migrations.RenameField(
'Organization',
'admins',
'deprecated_admins',
),
migrations.RenameField(
'Organization',
'users',
'deprecated_users',
),
migrations.RenameField(
'Team',
'users',
'deprecated_users',
),
migrations.CreateModel(
name='Role',
fields=[

26
awx/main/migrations/0009_v300_rbac_drop_fields.py
View File

@ -1,26 +0,0 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('main', '0008_v300_rbac_migrations'),
]
operations = [
migrations.RemoveField(
model_name='organization',
name='admins',
),
migrations.RemoveField(
model_name='organization',
name='users',
),
migrations.RemoveField(
model_name='team',
name='users',
),
]

66
awx/main/migrations/_old_access.py
View File

@ -231,7 +231,7 @@ class UserAccess(BaseAccess):
# Admin implies changing all user fields.
if self.user.is_superuser:
return True
return bool(obj.organizations.filter(active=True, admins__in=[self.user]).exists())
return bool(obj.organizations.filter(active=True, deprecated_admins__in=[self.user]).exists())
def can_delete(self, obj):
if obj == self.user:
@ -242,7 +242,7 @@ class UserAccess(BaseAccess):
# cannot delete the last active superuser
return False
return bool(self.user.is_superuser or
obj.organizations.filter(active=True, admins__in=[self.user]).exists())
obj.organizations.filter(active=True, deprecated_admins__in=[self.user]).exists())
class OrganizationAccess(BaseAccess):
'''
@ -261,11 +261,11 @@ class OrganizationAccess(BaseAccess):
qs = qs.select_related('created_by', 'modified_by')
if self.user.is_superuser:
return qs
return qs.filter(Q(admins__in=[self.user]) | Q(users__in=[self.user]))
return qs.filter(Q(deprecated_admins__in=[self.user]) | Q(deprecated_users__in=[self.user]))
def can_change(self, obj, data):
return bool(self.user.is_superuser or
self.user in obj.admins.all())
self.user in obj.deprecated_admins.all())
def can_delete(self, obj):
self.check_license(feature='multiple_organizations', check_expiration=False)
@ -300,7 +300,7 @@ class InventoryAccess(BaseAccess):
if self.user.is_superuser:
return qs
qs = qs.filter(organization__active=True)
admin_of = qs.filter(organization__admins__in=[self.user]).distinct()
admin_of = qs.filter(organization__deprecated_admins__in=[self.user]).distinct()
has_user_kw = dict(
permissions__user__in=[self.user],
permissions__permission_type__in=allowed,
@ -310,7 +310,7 @@ class InventoryAccess(BaseAccess):
has_user_kw['permissions__run_ad_hoc_commands'] = ad_hoc
has_user_perms = qs.filter(**has_user_kw).distinct()
has_team_kw = dict(
permissions__team__users__in=[self.user],
permissions__team__deprecated_users__in=[self.user],
permissions__team__active=True,
permissions__permission_type__in=allowed,
permissions__active=True,
@ -581,7 +581,7 @@ class CredentialAccess(BaseAccess):
Q(user__organizations__id__in=orgs_as_admin_ids) |
Q(user__admin_of_organizations__id__in=orgs_as_admin_ids) |
Q(team__organization__id__in=orgs_as_admin_ids, team__active=True) |
Q(team__users__in=[self.user], team__active=True)
Q(team__deprecated_users__in=[self.user], team__active=True)
)
def can_add(self, data):
@ -607,12 +607,12 @@ class CredentialAccess(BaseAccess):
if obj.user:
if self.user == obj.user:
return True
if obj.user.organizations.filter(active=True, admins__in=[self.user]).exists():
if obj.user.organizations.filter(active=True, deprecated_admins__in=[self.user]).exists():
return True
if obj.user.admin_of_organizations.filter(active=True, admins__in=[self.user]).exists():
if obj.user.admin_of_organizations.filter(active=True, deprecated_admins__in=[self.user]).exists():
return True
if obj.team:
if self.user in obj.team.organization.admins.filter(is_active=True):
if self.user in obj.team.organization.deprecated_admins.filter(is_active=True):
return True
return False
@ -642,8 +642,8 @@ class TeamAccess(BaseAccess):
if self.user.is_superuser:
return qs
return qs.filter(
Q(organization__admins__in=[self.user], organization__active=True) |
Q(users__in=[self.user])
Q(organization__deprecated_admins__in=[self.user], organization__active=True) |
Q(deprecated_users__in=[self.user])
)
def can_add(self, data):
@ -663,7 +663,7 @@ class TeamAccess(BaseAccess):
raise PermissionDenied('Unable to change organization on a team')
if self.user.is_superuser:
return True
if self.user in obj.organization.admins.all():
if self.user in obj.organization.deprecated_admins.all():
return True
return False
@ -693,10 +693,10 @@ class ProjectAccess(BaseAccess):
qs = qs.select_related('modified_by', 'credential', 'current_job', 'last_job')
if self.user.is_superuser:
return qs
team_ids = set(Team.objects.filter(users__in=[self.user]).values_list('id', flat=True))
team_ids = set(Team.objects.filter(deprecated_users__in=[self.user]).values_list('id', flat=True))
qs = qs.filter(Q(created_by=self.user, organizations__isnull=True) |
Q(organizations__admins__in=[self.user], organizations__active=True) |
Q(organizations__users__in=[self.user], organizations__active=True) |
Q(organizations__deprecated_admins__in=[self.user], organizations__active=True) |
Q(organizations__deprecated_users__in=[self.user], organizations__active=True) |
Q(teams__in=team_ids))
allowed_deploy = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY]
allowed_check = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY, PERM_INVENTORY_CHECK]
@ -728,7 +728,7 @@ class ProjectAccess(BaseAccess):
return True
if obj.created_by == self.user and not obj.organizations.filter(active=True).count():
return True
if obj.organizations.filter(active=True, admins__in=[self.user]).exists():
if obj.organizations.filter(active=True, deprecated_admins__in=[self.user]).exists():
return True
return False
@ -787,7 +787,7 @@ class PermissionAccess(BaseAccess):
Q(user__admin_of_organizations__in=orgs_as_admin_ids) |
Q(team__organization__in=orgs_as_admin_ids, team__active=True) |
Q(user=self.user) |
Q(team__users__in=[self.user], team__active=True)
Q(team__deprecated_users__in=[self.user], team__active=True)
)
def can_add(self, data):
@ -880,14 +880,14 @@ class JobTemplateAccess(BaseAccess):
Q(cloud_credential_id__in=credential_ids) | Q(cloud_credential__isnull=True),
)
org_admin_ids = base_qs.filter(
Q(project__organizations__admins__in=[self.user]) |
(Q(project__isnull=True) & Q(job_type=PERM_INVENTORY_SCAN) & Q(inventory__organization__admins__in=[self.user]))
Q(project__organizations__deprecated_admins__in=[self.user]) |
(Q(project__isnull=True) & Q(job_type=PERM_INVENTORY_SCAN) & Q(inventory__organization__deprecated_admins__in=[self.user]))
)
allowed_deploy = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY]
allowed_check = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY, PERM_INVENTORY_CHECK]
team_ids = Team.objects.filter(users__in=[self.user])
team_ids = Team.objects.filter(deprecated_users__in=[self.user])
# TODO: I think the below queries can be combined
deploy_permissions_ids = Permission.objects.filter(
@ -983,7 +983,7 @@ class JobTemplateAccess(BaseAccess):
# Otherwise, check for explicitly granted permissions to create job templates
# for the project and inventory.
permission_qs = Permission.objects.filter(
Q(user=self.user) | Q(team__users__in=[self.user]),
Q(user=self.user) | Q(team__deprecated_users__in=[self.user]),
inventory=inventory,
project=project,
active=True,
@ -1041,7 +1041,7 @@ class JobTemplateAccess(BaseAccess):
# Otherwise check for explicitly granted permissions
permission_qs = Permission.objects.filter(
Q(user=self.user) | Q(team__users__in=[self.user]),
Q(user=self.user) | Q(team__deprecated_users__in=[self.user]),
inventory=obj.inventory,
project=obj.project,
active=True,
@ -1097,13 +1097,13 @@ class JobAccess(BaseAccess):
credential_id__in=credential_ids,
)
org_admin_ids = base_qs.filter(
Q(project__organizations__admins__in=[self.user]) |
(Q(project__isnull=True) & Q(job_type=PERM_INVENTORY_SCAN) & Q(inventory__organization__admins__in=[self.user]))
Q(project__organizations__deprecated_admins__in=[self.user]) |
(Q(project__isnull=True) & Q(job_type=PERM_INVENTORY_SCAN) & Q(inventory__organization__deprecated_admins__in=[self.user]))
)
allowed_deploy = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY]
allowed_check = [PERM_JOBTEMPLATE_CREATE, PERM_INVENTORY_DEPLOY, PERM_INVENTORY_CHECK]
team_ids = Team.objects.filter(users__in=[self.user])
team_ids = Team.objects.filter(deprecated_users__in=[self.user])
# TODO: I think the below queries can be combined
deploy_permissions_ids = Permission.objects.filter(
@ -1219,7 +1219,7 @@ class AdHocCommandAccess(BaseAccess):
return qs
credential_ids = set(self.user.get_queryset(Credential).values_list('id', flat=True))
team_ids = set(Team.objects.filter(active=True, users__in=[self.user]).values_list('id', flat=True))
team_ids = set(Team.objects.filter(active=True, deprecated_users__in=[self.user]).values_list('id', flat=True))
permission_ids = set(Permission.objects.filter(
Q(user=self.user) | Q(team__in=team_ids),
@ -1229,7 +1229,7 @@ class AdHocCommandAccess(BaseAccess):
).values_list('id', flat=True))
inventory_qs = self.user.get_queryset(Inventory)
inventory_qs = inventory_qs.filter(Q(permissions__in=permission_ids) | Q(organization__admins__in=[self.user]))
inventory_qs = inventory_qs.filter(Q(permissions__in=permission_ids) | Q(organization__deprecated_admins__in=[self.user]))
inventory_ids = set(inventory_qs.values_list('id', flat=True))
qs = qs.filter(
@ -1512,7 +1512,7 @@ class ActivityStreamAccess(BaseAccess):
user_orgs = self.user.organizations.all()
#Organization filter
qs = qs.filter(Q(organization__admins__in=[self.user]) | Q(organization__users__in=[self.user]))
qs = qs.filter(Q(organization__deprecated_admins__in=[self.user]) | Q(organization__deprecated_users__in=[self.user]))
#User filter
qs = qs.filter(Q(user__pk=self.user.pk) |
@ -1542,11 +1542,11 @@ class ActivityStreamAccess(BaseAccess):
Q(credential__user__organizations__in=user_admin_orgs) |
Q(credential__user__admin_of_organizations__in=user_admin_orgs) |
Q(credential__team__organization__in=user_admin_orgs) |
Q(credential__team__users__in=[self.user]))
Q(credential__team__deprecated_users__in=[self.user]))
#Team Filter
qs.filter(Q(team__organization__admins__in=[self.user]) |
Q(team__users__in=[self.user]))
qs.filter(Q(team__organization__deprecated_admins__in=[self.user]) |
Q(team__deprecated_users__in=[self.user]))
#Project Filter
project_qs = self.user.get_queryset(Project)
@ -1616,7 +1616,7 @@ class CustomInventoryScriptAccess(BaseAccess):
def get_queryset(self):
qs = self.model.objects.filter(active=True).distinct()
if not self.user.is_superuser:
qs = qs.filter(Q(organization__admins__in=[self.user]) | Q(organization__users__in=[self.user]))
qs = qs.filter(Q(organization__deprecated_admins__in=[self.user]) | Q(organization__deprecated_users__in=[self.user]))
return qs
def can_read(self, obj):

8
awx/main/migrations/_rbac.py
View File

@ -35,10 +35,10 @@ def migrate_organization(apps, schema_editor):
migrations = defaultdict(list)
organization = apps.get_model('main', "Organization")
for org in organization.objects.all():
for admin in org.admins.all():
for admin in org.deprecated_admins.all():
org.admin_role.members.add(admin)
migrations[org.name].append(admin)
for user in org.users.all():
for user in org.deprecated_users.all():
org.auditor_role.members.add(user)
migrations[org.name].append(user)
return migrations
@ -47,7 +47,7 @@ def migrate_team(apps, schema_editor):
migrations = defaultdict(list)
team = apps.get_model('main', 'Team')
for t in team.objects.all():
for user in t.users.all():
for user in t.deprecated_users.all():
t.member_role.members.add(user)
migrations[t.name].append(user)
return migrations
@ -143,7 +143,7 @@ def migrate_projects(apps, schema_editor):
if project.organizations.count() > 0:
for org in project.organizations.all():
for user in org.users.all():
for user in org.deprecated_users.all():
project.member_role.members.add(user)
migrations[project.name]['users'].add(user)

15
awx/main/models/organization.py
View File

@ -38,6 +38,16 @@ class Organization(CommonModel, NotificationFieldsModel, ResourceMixin):
app_label = 'main'
ordering = ('name',)
deprecated_users = models.ManyToManyField(
'auth.User',
blank=True,
related_name='organizations',
)
deprecated_admins = models.ManyToManyField(
'auth.User',
blank=True,
related_name='admin_of_organizations',
)
projects = models.ManyToManyField(
'Project',
blank=True,
@ -86,6 +96,11 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
unique_together = [('organization', 'name')]
ordering = ('organization__name', 'name')
deprecated_users = models.ManyToManyField(
'auth.User',
blank=True,
related_name='teams',
)
organization = models.ForeignKey(
'Organization',
blank=False,

8
awx/main/tests/functional/test_rbac_inventory.py
View File

@ -81,7 +81,7 @@ def test_inventory_admin_team(inventory, permissions, user, team):
u = user('admin', False)
perm = Permission(team=team, inventory=inventory, permission_type='admin')
perm.save()
team.users.add(u)
team.deprecated_users.add(u)
assert inventory.accessible_by(u, permissions['admin']) is False
@ -105,7 +105,7 @@ def test_inventory_auditor(inventory, permissions, user, team):
u = user('auditor', False)
perm = Permission(team=team, inventory=inventory, permission_type='read')
perm.save()
team.users.add(u)
team.deprecated_users.add(u)
assert inventory.accessible_by(u, permissions['admin']) is False
assert inventory.accessible_by(u, permissions['auditor']) is False
@ -129,7 +129,7 @@ def test_inventory_updater(inventory, permissions, user, team):
u = user('updater', False)
perm = Permission(team=team, inventory=inventory, permission_type='write')
perm.save()
team.users.add(u)
team.deprecated_users.add(u)
assert inventory.accessible_by(u, permissions['admin']) is False
assert inventory.accessible_by(u, permissions['auditor']) is False
@ -154,7 +154,7 @@ def test_inventory_executor(inventory, permissions, user, team):
u = user('executor', False)
perm = Permission(team=team, inventory=inventory, permission_type='read', run_ad_hoc_commands=True)
perm.save()
team.users.add(u)
team.deprecated_users.add(u)
assert inventory.accessible_by(u, permissions['admin']) is False
assert inventory.accessible_by(u, permissions['auditor']) is False

12
awx/main/tests/functional/test_rbac_job_templates.py
View File

@ -16,7 +16,7 @@ def test_job_template_migration_check(deploy_jobtemplate, check_jobtemplate, use
joe = user('joe')
check_jobtemplate.project.organizations.all()[0].users.add(joe)
check_jobtemplate.project.organizations.all()[0].deprecated_users.add(joe)
Permission(user=joe, inventory=check_jobtemplate.inventory, permission_type='read').save()
Permission(user=joe, inventory=check_jobtemplate.inventory,
@ -45,7 +45,7 @@ def test_job_template_migration_deploy(deploy_jobtemplate, check_jobtemplate, us
joe = user('joe')
deploy_jobtemplate.project.organizations.all()[0].users.add(joe)
deploy_jobtemplate.project.organizations.all()[0].deprecated_users.add(joe)
Permission(user=joe, inventory=deploy_jobtemplate.inventory, permission_type='read').save()
Permission(user=joe, inventory=deploy_jobtemplate.inventory,
@ -73,11 +73,11 @@ def test_job_template_migration_deploy(deploy_jobtemplate, check_jobtemplate, us
def test_job_template_team_migration_check(deploy_jobtemplate, check_jobtemplate, organization, team, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
team.users.add(joe)
team.deprecated_users.add(joe)
team.organization = organization
team.save()
check_jobtemplate.project.organizations.all()[0].users.add(joe)
check_jobtemplate.project.organizations.all()[0].deprecated_users.add(joe)
Permission(team=team, inventory=check_jobtemplate.inventory, permission_type='read').save()
Permission(team=team, inventory=check_jobtemplate.inventory,
@ -108,11 +108,11 @@ def test_job_template_team_migration_check(deploy_jobtemplate, check_jobtemplate
def test_job_template_team_deploy_migration(deploy_jobtemplate, check_jobtemplate, organization, team, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
team.users.add(joe)
team.deprecated_users.add(joe)
team.organization = organization
team.save()
deploy_jobtemplate.project.organizations.all()[0].users.add(joe)
deploy_jobtemplate.project.organizations.all()[0].deprecated_users.add(joe)
Permission(team=team, inventory=deploy_jobtemplate.inventory, permission_type='read').save()
Permission(team=team, inventory=deploy_jobtemplate.inventory,

10
awx/main/tests/functional/test_rbac_organization.py
View File

@ -12,7 +12,7 @@ from django.apps import apps
@pytest.mark.django_db
def test_organization_migration_admin(organization, permissions, user):
u = user('admin', False)
organization.admins.add(u)
organization.deprecated_admins.add(u)
# Undo some automatic work that we're supposed to be testing with our migration
organization.admin_role.members.remove(u)
@ -26,7 +26,7 @@ def test_organization_migration_admin(organization, permissions, user):
@pytest.mark.django_db
def test_organization_migration_user(organization, permissions, user):
u = user('user', False)
organization.users.add(u)
organization.deprecated_users.add(u)
# Undo some automatic work that we're supposed to be testing with our migration
organization.member_role.members.remove(u)
@ -42,14 +42,14 @@ def test_organization_migration_user(organization, permissions, user):
@pytest.mark.django_db
def test_organization_access_superuser(cl, organization, user):
access = OrganizationAccess(user('admin', True))
organization.users.add(user('user', False))
organization.deprecated_users.add(user('user', False))
assert access.can_change(organization, None)
assert access.can_delete(organization)
org = access.get_queryset()[0]
assert len(org.admins.all()) == 0
assert len(org.users.all()) == 1
assert len(org.deprecated_admins.all()) == 0
assert len(org.deprecated_users.all()) == 1
@mock.patch.object(BaseAccess, 'check_license', return_value=None)

173
awx/main/tests/functional/test_rbac_project.py
View File

@ -2,95 +2,96 @@ import pytest
from awx.main.migrations import _rbac as rbac
from awx.main.models import Role
from awx.main.models.organization import Permission
from django.apps import apps
from awx.main.migrations import _old_access as old_access
#@pytest.mark.django_db
#def test_project_user_project(user_project, project, user):
# u = user('owner')
#
# assert old_access.check_user_access(u, user_project.__class__, 'read', user_project)
# assert old_access.check_user_access(u, project.__class__, 'read', project) is False
#
# assert user_project.accessible_by(u, {'read': True}) is False
# assert project.accessible_by(u, {'read': True}) is False
# migrations = rbac.migrate_projects(apps, None)
# assert len(migrations[user_project.name]['users']) == 1
# assert len(migrations[user_project.name]['teams']) == 0
# assert user_project.accessible_by(u, {'read': True}) is True
# assert project.accessible_by(u, {'read': True}) is False
#
#@pytest.mark.django_db
#def test_project_accessible_by_sa(user, project):
# u = user('systemadmin', is_superuser=True)
# # This gets setup by a signal, but we want to test the migration which will set this up too, so remove it
# Role.singleton('System Administrator').members.remove(u)
#
# assert project.accessible_by(u, {'read': True}) is False
# rbac.migrate_organization(apps, None)
# su_migrations = rbac.migrate_users(apps, None)
# migrations = rbac.migrate_projects(apps, None)
# assert len(su_migrations) == 1
# assert len(migrations[project.name]['users']) == 0
# assert len(migrations[project.name]['teams']) == 0
# print(project.admin_role.ancestors.all())
# print(project.admin_role.ancestors.all())
# assert project.accessible_by(u, {'read': True, 'write': True}) is True
#
#@pytest.mark.django_db
#def test_project_org_members(user, organization, project):
# admin = user('orgadmin')
# member = user('orgmember')
#
# assert project.accessible_by(admin, {'read': True}) is False
# assert project.accessible_by(member, {'read': True}) is False
#
# organization.admin_role.members.add(admin)
# organization.member_role.members.add(member)
#
# rbac.migrate_organization(apps, None)
# migrations = rbac.migrate_projects(apps, None)
#
# assert len(migrations[project.name]['users']) == 0
# assert len(migrations[project.name]['teams']) == 0
# assert project.accessible_by(admin, {'read': True, 'write': True}) is True
# assert project.accessible_by(member, {'read': True}) is False
@pytest.mark.django_db
def test_project_user_project(user_project, project, user):
u = user('owner')
#@pytest.mark.django_db
#def test_project_team(user, team, project):
# nonmember = user('nonmember')
# member = user('member')
#
# team.users.add(member)
# project.teams.add(team)
#
# assert project.accessible_by(nonmember, {'read': True}) is False
# assert project.accessible_by(member, {'read': True}) is False
#
# rbac.migrate_team(apps, None)
# rbac.migrate_organization(apps, None)
# migrations = rbac.migrate_projects(apps, None)
#
# assert len(migrations[project.name]['users']) == 0
# assert len(migrations[project.name]['teams']) == 1
# assert project.accessible_by(member, {'read': True}) is True
# assert project.accessible_by(nonmember, {'read': True}) is False
assert old_access.check_user_access(u, user_project.__class__, 'read', user_project)
assert old_access.check_user_access(u, project.__class__, 'read', project) is False
#@pytest.mark.django_db
#def test_project_explicit_permission(user, team, project, organization):
# u = user('prjuser')
#
# assert old_access.check_user_access(u, project.__class__, 'read', project) is False
#
# organization.users.add(u)
# p = Permission(user=u, project=project, permission_type='create', name='Perm name')
# p.save()
#
# assert project.accessible_by(u, {'read': True}) is False
#
# rbac.migrate_organization(apps, None)
# migrations = rbac.migrate_projects(apps, None)
#
# assert len(migrations[project.name]['users']) == 1
# assert project.accessible_by(u, {'read': True}) is True
assert user_project.accessible_by(u, {'read': True}) is False
assert project.accessible_by(u, {'read': True}) is False
migrations = rbac.migrate_projects(apps, None)
assert len(migrations[user_project.name]['users']) == 1
assert len(migrations[user_project.name]['teams']) == 0
assert user_project.accessible_by(u, {'read': True}) is True
assert project.accessible_by(u, {'read': True}) is False
@pytest.mark.django_db
def test_project_accessible_by_sa(user, project):
u = user('systemadmin', is_superuser=True)
# This gets setup by a signal, but we want to test the migration which will set this up too, so remove it
Role.singleton('System Administrator').members.remove(u)
assert project.accessible_by(u, {'read': True}) is False
rbac.migrate_organization(apps, None)
su_migrations = rbac.migrate_users(apps, None)
migrations = rbac.migrate_projects(apps, None)
assert len(su_migrations) == 1
assert len(migrations[project.name]['users']) == 0
assert len(migrations[project.name]['teams']) == 0
print(project.admin_role.ancestors.all())
print(project.admin_role.ancestors.all())
assert project.accessible_by(u, {'read': True, 'write': True}) is True
@pytest.mark.django_db
def test_project_org_members(user, organization, project):
admin = user('orgadmin')
member = user('orgmember')
assert project.accessible_by(admin, {'read': True}) is False
assert project.accessible_by(member, {'read': True}) is False
organization.deprecated_admins.add(admin)
organization.deprecated_users.add(member)
rbac.migrate_organization(apps, None)
migrations = rbac.migrate_projects(apps, None)
assert len(migrations[project.name]['users']) == 1
assert len(migrations[project.name]['teams']) == 0
assert project.accessible_by(admin, {'read': True, 'write': True}) is True
assert project.accessible_by(member, {'read': True})
@pytest.mark.django_db
def test_project_team(user, team, project):
nonmember = user('nonmember')
member = user('member')
team.deprecated_users.add(member)
project.teams.add(team)
assert project.accessible_by(nonmember, {'read': True}) is False
assert project.accessible_by(member, {'read': True}) is False
rbac.migrate_team(apps, None)
rbac.migrate_organization(apps, None)
migrations = rbac.migrate_projects(apps, None)
assert len(migrations[project.name]['users']) == 0
assert len(migrations[project.name]['teams']) == 1
assert project.accessible_by(member, {'read': True}) is True
assert project.accessible_by(nonmember, {'read': True}) is False
@pytest.mark.django_db
def test_project_explicit_permission(user, team, project, organization):
u = user('prjuser')
assert old_access.check_user_access(u, project.__class__, 'read', project) is False
organization.deprecated_users.add(u)
p = Permission(user=u, project=project, permission_type='create', name='Perm name')
p.save()
assert project.accessible_by(u, {'read': True}) is False
rbac.migrate_organization(apps, None)
migrations = rbac.migrate_projects(apps, None)
assert len(migrations[project.name]['users']) == 1
assert project.accessible_by(u, {'read': True}) is True

20
awx/main/tests/old/inventory.py
View File

@ -47,9 +47,9 @@ class InventoryTest(BaseTest):
self.setup_instances()
self.setup_users()
self.organizations = self.make_organizations(self.super_django_user, 3)
self.organizations[0].admins.add(self.normal_django_user)
self.organizations[0].users.add(self.other_django_user)
self.organizations[0].users.add(self.normal_django_user)
self.organizations[0].deprecated_admins.add(self.normal_django_user)
self.organizations[0].deprecated_users.add(self.other_django_user)
self.organizations[0].deprecated_users.add(self.normal_django_user)
self.inventory_a = Inventory.objects.create(name='inventory-a', description='foo', organization=self.organizations[0])
self.inventory_b = Inventory.objects.create(name='inventory-b', description='bar', organization=self.organizations[1])
@ -78,7 +78,7 @@ class InventoryTest(BaseTest):
self.check_get_list(url, self.super_django_user, qs)
# an org admin can list inventories but is filtered to what he adminsters
normal_qs = qs.filter(organization__admins__in=[self.normal_django_user])
normal_qs = qs.filter(organization__deprecated_admins__in=[self.normal_django_user])
self.check_get_list(url, self.normal_django_user, normal_qs)
# a user who is on a team who has a read permissions on an inventory can see filtered inventories
@ -269,9 +269,9 @@ class InventoryTest(BaseTest):
def test_inventory_access_deleted_permissions(self):
temp_org = self.make_organizations(self.super_django_user, 1)[0]
temp_org.admins.add(self.normal_django_user)
temp_org.users.add(self.other_django_user)
temp_org.users.add(self.normal_django_user)
temp_org.deprecated_admins.add(self.normal_django_user)
temp_org.deprecated_users.add(self.other_django_user)
temp_org.deprecated_users.add(self.normal_django_user)
temp_inv = temp_org.inventories.create(name='Delete Org Inventory')
temp_inv.groups.create(name='Delete Org Inventory Group')
@ -1195,9 +1195,9 @@ class InventoryUpdatesTest(BaseTransactionTest):
self.setup_instances()
self.setup_users()
self.organization = self.make_organizations(self.super_django_user, 1)[0]
self.organization.admins.add(self.normal_django_user)
self.organization.users.add(self.other_django_user)
self.organization.users.add(self.normal_django_user)
self.organization.deprecated_admins.add(self.normal_django_user)
self.organization.deprecated_users.add(self.other_django_user)
self.organization.deprecated_users.add(self.normal_django_user)
self.inventory = self.organization.inventories.create(name='Cloud Inventory')
self.group = self.inventory.groups.create(name='Cloud Group')
self.inventory2 = self.organization.inventories.create(name='Cloud Inventory 2')

2
awx/main/tests/old/licenses.py
View File

@ -19,7 +19,7 @@ class LicenseTests(BaseTest):
self.setup_users()
u = self.super_django_user
org = Organization.objects.create(name='o1', created_by=u)
org.admins.add(self.normal_django_user)
org.deprecated_admins.add(self.normal_django_user)
self.inventory = Inventory.objects.create(name='hi', organization=org, created_by=u)
Host.objects.create(name='a1', inventory=self.inventory, created_by=u)
Host.objects.create(name='a2', inventory=self.inventory, created_by=u)

10
awx/main/tests/old/organizations.py
View File

@ -88,12 +88,12 @@ class OrganizationsTest(BaseTest):
# nobody_user is a user not a member of any organizations
for x in self.organizations:
x.admins.add(self.super_django_user)
x.users.add(self.super_django_user)
x.users.add(self.other_django_user)
x.deprecated_admins.add(self.super_django_user)
x.deprecated_users.add(self.super_django_user)
x.deprecated_users.add(self.other_django_user)
self.organizations[0].users.add(self.normal_django_user)
self.organizations[1].admins.add(self.normal_django_user)
self.organizations[0].deprecated_users.add(self.normal_django_user)
self.organizations[1].deprecated_admins.add(self.normal_django_user)
def test_get_organization_list(self):
url = reverse('api:organization_list')

26
awx/main/tests/old/projects.py
View File

@ -75,10 +75,10 @@ class ProjectsTest(BaseTransactionTest):
for x in self.organizations:
# NOTE: superuser does not have to be explicitly added to admin group
# x.admins.add(self.super_django_user)
x.users.add(self.super_django_user)
x.deprecated_users.add(self.super_django_user)
self.organizations[0].users.add(self.normal_django_user)
self.organizations[1].admins.add(self.normal_django_user)
self.organizations[0].deprecated_users.add(self.normal_django_user)
self.organizations[1].deprecated_admins.add(self.normal_django_user)
self.team1 = Team.objects.create(
name = 'team1', organization = self.organizations[0]
@ -97,8 +97,8 @@ class ProjectsTest(BaseTransactionTest):
self.team2.projects.add(self.projects[5])
self.team1.save()
self.team2.save()
self.team1.users.add(self.normal_django_user)
self.team2.users.add(self.other_django_user)
self.team1.deprecated_users.add(self.normal_django_user)
self.team2.deprecated_users.add(self.other_django_user)
def test_playbooks(self):
def write_test_file(project, name, content):
@ -312,11 +312,11 @@ class ProjectsTest(BaseTransactionTest):
# Verify that creatorship doesn't imply access if access is removed
a_new_proj = self.make_project(created_by=self.other_django_user, playbook_content=TEST_PLAYBOOK)
self.organizations[0].admins.add(self.other_django_user)
self.organizations[0].deprecated_admins.add(self.other_django_user)
self.organizations[0].projects.add(a_new_proj)
proj_detail = reverse('api:project_detail', args=(a_new_proj.pk,))
self.patch(proj_detail, data=dict(description="test"), expect=200, auth=self.get_other_credentials())
self.organizations[0].admins.remove(self.other_django_user)
self.organizations[0].deprecated_admins.remove(self.other_django_user)
self.patch(proj_detail, data=dict(description="test_now"), expect=403, auth=self.get_other_credentials())
self.delete(proj_detail, expect=403, auth=self.get_other_credentials())
a_new_proj.delete()
@ -337,7 +337,7 @@ class ProjectsTest(BaseTransactionTest):
self.assertEquals(got['url'], reverse('api:team_detail', args=(self.team1.pk,)))
got = self.get(team1, expect=200, auth=self.get_normal_credentials())
got = self.get(team1, expect=403, auth=self.get_other_credentials())
self.team1.users.add(User.objects.get(username='other'))
self.team1.deprecated_users.add(User.objects.get(username='other'))
self.team1.save()
got = self.get(team1, expect=200, auth=self.get_other_credentials())
got = self.get(team1, expect=403, auth=self.get_nobody_credentials())
@ -421,8 +421,8 @@ class ProjectsTest(BaseTransactionTest):
team = Team.objects.filter(active=True, organization__pk=self.organizations[1].pk)[0]
team_users = reverse('api:team_users_list', args=(team.pk,))
for x in team.users.all():
team.users.remove(x)
for x in team.deprecated_users.all():
team.deprecated_users.remove(x)
team.save()
# can list uses on teams
@ -454,7 +454,7 @@ class ProjectsTest(BaseTransactionTest):
self.post(team_users, data=y, expect=403, auth=self.get_nobody_credentials())
self.post(team_users, data=y, expect=204, auth=self.get_normal_credentials())
self.assertEquals(Team.objects.get(pk=team.pk).users.count(), 1) # Leaving just the super user we created
self.assertEquals(Team.objects.get(pk=team.pk).deprecated_users.count(), 1) # Leaving just the super user we created
# =====================================================================
# USER TEAMS
@ -787,7 +787,7 @@ class ProjectsTest(BaseTransactionTest):
# User is still a team member
self.get(reverse('api:project_detail', args=(project.pk,)), expect=200, auth=self.get_other_credentials())
team.users.remove(self.other_django_user)
team.deprecated_users.remove(self.other_django_user)
# User is no longer a team member and has no permissions
self.get(reverse('api:project_detail', args=(project.pk,)), expect=403, auth=self.get_other_credentials())
@ -1351,7 +1351,7 @@ class ProjectUpdatesTest(BaseTransactionTest):
'scm_url': scm_url,
}
org = self.make_organizations(self.super_django_user, 1)[0]
org.admins.add(self.normal_django_user)
org.deprecated_admins.add(self.normal_django_user)
with self.current_user(self.super_django_user):
del_proj = self.post(projects_url, project_data, expect=201)
del_proj = Project.objects.get(pk=del_proj["id"])

8
awx/main/tests/old/schedules.py
View File

@ -54,12 +54,12 @@ class ScheduleTest(BaseTest):
self.setup_instances()
self.setup_users()
self.organizations = self.make_organizations(self.super_django_user, 2)
self.organizations[0].admins.add(self.normal_django_user)
self.organizations[0].users.add(self.other_django_user)
self.organizations[0].users.add(self.normal_django_user)
self.organizations[0].deprecated_admins.add(self.normal_django_user)
self.organizations[0].deprecated_users.add(self.other_django_user)
self.organizations[0].deprecated_users.add(self.normal_django_user)
self.diff_org_user = self.make_user('fred')
self.organizations[1].users.add(self.diff_org_user)
self.organizations[1].deprecated_users.add(self.diff_org_user)
self.cloud_source = Credential.objects.create(kind='awx', user=self.super_django_user,
username='Dummy', password='Dummy')

42
awx/main/tests/old/users.py
View File

@ -100,7 +100,7 @@ class AuthTokenProxyTest(BaseTest):
self.setup_users()
self.setup_instances()
self.organizations = self.make_organizations(self.super_django_user, 2)
self.organizations[0].admins.add(self.normal_django_user)
self.organizations[0].deprecated_admins.add(self.normal_django_user)
self.assertIn('REMOTE_ADDR', settings.REMOTE_HOST_HEADERS)
self.assertIn('REMOTE_HOST', settings.REMOTE_HOST_HEADERS)
@ -174,10 +174,10 @@ class UsersTest(BaseTest):
super(UsersTest, self).setUp()
self.setup_users()
self.organizations = self.make_organizations(self.super_django_user, 2)
self.organizations[0].admins.add(self.normal_django_user)
self.organizations[0].users.add(self.other_django_user)
self.organizations[0].users.add(self.normal_django_user)
self.organizations[1].users.add(self.other_django_user)
self.organizations[0].deprecated_admins.add(self.normal_django_user)
self.organizations[0].deprecated_users.add(self.other_django_user)
self.organizations[0].deprecated_users.add(self.normal_django_user)
self.organizations[1].deprecated_users.add(self.other_django_user)
def test_user_creation_fails_without_password(self):
url = reverse('api:user_list')
@ -790,8 +790,8 @@ class UsersTest(BaseTest):
self.check_get_list(url, self.super_django_user, qs)
# Filter by related organizations admins username.
url = '%s?organizations__admins__username__startswith=norm' % base_url
qs = base_qs.filter(organizations__admins__username__startswith='norm')
url = '%s?organizations__deprecated_admins__username__startswith=norm' % base_url
qs = base_qs.filter(organizations__deprecated_admins__username__startswith='norm')
self.assertTrue(qs.count())
self.check_get_list(url, self.super_django_user, qs)
@ -839,11 +839,11 @@ class UsersTest(BaseTest):
self.check_get_list(url, self.super_django_user, base_qs, expect=400)
# Filter by invalid field across lookups.
url = '%s?organizations__users__teams__laser=green' % base_url
url = '%s?organizations__deprecated_users__teams__laser=green' % base_url
self.check_get_list(url, self.super_django_user, base_qs, expect=400)
# Filter by invalid relation within lookups.
url = '%s?organizations__users__llamas__name=freddie' % base_url
url = '%s?organizations__deprecated_users__llamas__name=freddie' % base_url
self.check_get_list(url, self.super_django_user, base_qs, expect=400)
# Filter by invalid query string field names.
@ -1020,13 +1020,13 @@ class LdapTest(BaseTest):
for org_name, org_result in settings.AUTH_LDAP_ORGANIZATION_MAP_RESULT.items():
org = Organization.objects.get(name=org_name)
if org_result.get('admins', False):
self.assertTrue(user in org.admins.all())
self.assertTrue(user in org.deprecated_admins.all())
else:
self.assertFalse(user in org.admins.all())
self.assertFalse(user in org.deprecated_admins.all())
if org_result.get('users', False):
self.assertTrue(user in org.users.all())
self.assertTrue(user in org.deprecated_users.all())
else:
self.assertFalse(user in org.users.all())
self.assertFalse(user in org.deprecated_users.all())
# Try again with different test mapping.
self.use_test_setting('ORGANIZATION_MAP', {},
from_name='ORGANIZATION_MAP_2')
@ -1038,13 +1038,13 @@ class LdapTest(BaseTest):
for org_name, org_result in settings.AUTH_LDAP_ORGANIZATION_MAP_RESULT.items():
org = Organization.objects.get(name=org_name)
if org_result.get('admins', False):
self.assertTrue(user in org.admins.all())
self.assertTrue(user in org.deprecated_admins.all())
else:
self.assertFalse(user in org.admins.all())
self.assertFalse(user in org.deprecated_admins.all())
if org_result.get('users', False):
self.assertTrue(user in org.users.all())
self.assertTrue(user in org.deprecated_users.all())
else:
self.assertFalse(user in org.users.all())
self.assertFalse(user in org.deprecated_users.all())
def test_ldap_team_mapping(self):
for name in ('USER_SEARCH', 'ALWAYS_UPDATE_USER', 'USER_ATTR_MAP',
@ -1062,9 +1062,9 @@ class LdapTest(BaseTest):
for team_name, team_result in settings.AUTH_LDAP_TEAM_MAP_RESULT.items():
team = Team.objects.get(name=team_name)
if team_result.get('users', False):
self.assertTrue(user in team.users.all())
self.assertTrue(user in team.deprecated_users.all())
else:
self.assertFalse(user in team.users.all())
self.assertFalse(user in team.deprecated_users.all())
# Try again with different test mapping.
self.use_test_setting('TEAM_MAP', {}, from_name='TEAM_MAP_2')
self.use_test_setting('TEAM_MAP_RESULT', {},
@ -1075,9 +1075,9 @@ class LdapTest(BaseTest):
for team_name, team_result in settings.AUTH_LDAP_TEAM_MAP_RESULT.items():
team = Team.objects.get(name=team_name)
if team_result.get('users', False):
self.assertTrue(user in team.users.all())
self.assertTrue(user in team.deprecated_users.all())
else:
self.assertFalse(user in team.users.all())
self.assertFalse(user in team.deprecated_users.all())
def test_prevent_changing_ldap_user_fields(self):
for name in ('USER_SEARCH', 'ALWAYS_UPDATE_USER', 'USER_ATTR_MAP',