mirror of
https://github.com/ansible/awx.git
synced 2026-03-25 12:55:04 -02:30
Project migration and tests
This commit is contained in:
@@ -86,3 +86,49 @@ def migrate_inventory(apps, schema_editor):
|
|||||||
migrations[inventory.name]['teams'] = teams
|
migrations[inventory.name]['teams'] = teams
|
||||||
migrations[inventory.name]['users'] = users
|
migrations[inventory.name]['users'] = users
|
||||||
return migrations
|
return migrations
|
||||||
|
|
||||||
|
def migrate_projects(apps, schema_editor):
|
||||||
|
'''
|
||||||
|
I can see projects when:
|
||||||
|
X I am a superuser.
|
||||||
|
X I am an admin in an organization associated with the project.
|
||||||
|
X I am a user in an organization associated with the project.
|
||||||
|
X I am on a team associated with the project.
|
||||||
|
X I have been explicitly granted permission to run/check jobs using the
|
||||||
|
project.
|
||||||
|
X I created the project but it isn't associated with an organization
|
||||||
|
I can change/delete when:
|
||||||
|
X I am a superuser.
|
||||||
|
X I am an admin in an organization associated with the project.
|
||||||
|
X I created the project but it isn't associated with an organization
|
||||||
|
'''
|
||||||
|
migrations = defaultdict(lambda: defaultdict(set))
|
||||||
|
|
||||||
|
Project = apps.get_model('main', 'Project')
|
||||||
|
Permission = apps.get_model('main', 'Permission')
|
||||||
|
|
||||||
|
for project in Project.objects.all():
|
||||||
|
if project.organization is None and project.created_by is not None:
|
||||||
|
project.admin_role.members.add(project.created_by)
|
||||||
|
migrations[project.name]['users'].add(project.created_by)
|
||||||
|
|
||||||
|
for team in project.teams.all():
|
||||||
|
team.member_role.children.add(project.member_role)
|
||||||
|
migrations[project.name]['teams'].add(team)
|
||||||
|
|
||||||
|
if project.organization is not None:
|
||||||
|
for user in project.organization.users.all():
|
||||||
|
project.member_role.members.add(user)
|
||||||
|
migrations[project.name]['users'].add(user)
|
||||||
|
|
||||||
|
for perm in Permission.objects.filter(project=project):
|
||||||
|
# All perms at this level just imply a user or team can read
|
||||||
|
if perm.team:
|
||||||
|
team.member_role.children.add(project.member_role)
|
||||||
|
migrations[project.name]['teams'].add(team)
|
||||||
|
|
||||||
|
if perm.user:
|
||||||
|
project.member_role.members.add(perm.user)
|
||||||
|
migrations[project.name]['users'].add(perm.user)
|
||||||
|
|
||||||
|
return migrations
|
||||||
|
|||||||
@@ -229,13 +229,12 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin):
|
|||||||
)
|
)
|
||||||
member_role = ImplicitRoleField(
|
member_role = ImplicitRoleField(
|
||||||
role_name='Project Member',
|
role_name='Project Member',
|
||||||
parent_role='admin',
|
|
||||||
resource_field='resource',
|
resource_field='resource',
|
||||||
permissions = {'usage': True}
|
permissions = {'read': True}
|
||||||
)
|
)
|
||||||
scm_update_role = ImplicitRoleField(
|
scm_update_role = ImplicitRoleField(
|
||||||
role_name='Project Updater',
|
role_name='Project Updater',
|
||||||
parent_role='admin',
|
parent_role='admin_role',
|
||||||
resource_field='resource',
|
resource_field='resource',
|
||||||
permissions = {'scm_update': True}
|
permissions = {'scm_update': True}
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import pytest
|
|||||||
|
|
||||||
from awx.main.models.credential import Credential
|
from awx.main.models.credential import Credential
|
||||||
from awx.main.models.inventory import Inventory
|
from awx.main.models.inventory import Inventory
|
||||||
|
from awx.main.models.projects import Project
|
||||||
from awx.main.models.organization import (
|
from awx.main.models.organization import (
|
||||||
Organization,
|
Organization,
|
||||||
Team,
|
Team,
|
||||||
@@ -23,6 +24,15 @@ def user():
|
|||||||
def team(organization):
|
def team(organization):
|
||||||
return Team.objects.create(organization=organization, name='test-team')
|
return Team.objects.create(organization=organization, name='test-team')
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def project(organization):
|
||||||
|
return Project.objects.create(name="test-project", organization=organization, description="test-project-desc")
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def user_project(user):
|
||||||
|
owner = user('owner')
|
||||||
|
return Project.objects.create(name="test-user-project", created_by=owner, description="test-user-project-desc")
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def organization():
|
def organization():
|
||||||
return Organization.objects.create(name="test-org", description="test-org-desc")
|
return Organization.objects.create(name="test-org", description="test-org-desc")
|
||||||
|
|||||||
79
awx/main/tests/functional/test_rbac_project.py
Normal file
79
awx/main/tests/functional/test_rbac_project.py
Normal file
@@ -0,0 +1,79 @@
|
|||||||
|
import pytest
|
||||||
|
|
||||||
|
from awx.main.migrations import _rbac as rbac
|
||||||
|
from awx.main.models import Permission
|
||||||
|
from django.apps import apps
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_project_user_project(user_project, project, user):
|
||||||
|
u = user('owner')
|
||||||
|
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)
|
||||||
|
|
||||||
|
assert project.accessible_by(u, {'read': True}) is False
|
||||||
|
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
|
||||||
|
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_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)
|
||||||
|
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):
|
||||||
|
u = user('user')
|
||||||
|
p = Permission(user=u, project=project, permission_type='check')
|
||||||
|
p.save()
|
||||||
|
|
||||||
|
assert project.accessible_by(u, {'read': True}) is False
|
||||||
|
|
||||||
|
migrations = rbac.migrate_projects(apps, None)
|
||||||
|
|
||||||
|
assert len(migrations[project.name]['users']) == 1
|
||||||
|
assert project.accessible_by(u, {'read': True}) is True
|
||||||
Reference in New Issue
Block a user