External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-08 01:47:35 -02:30
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #5726 from AlanCoding/jt_org_2020

Browse Source 
Add read-only organization field to job templates

Reviewed-by: https://github.com/apps/softwarefactory-project-zuul
This commit is contained in:
softwarefactory-project-zuul[bot]
2020-03-16 20:03:47 +00:00
committed by GitHub
parent c8510f7d75 521cda878e
commit 1e780aad38
46 changed files with 1325 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
awx/main/tests/functional/api/test_job.py
View File

@@ -39,6 +39,26 @@ def test_extra_credentials(get, organization_factory, job_template_factory, cred
@pytest.mark.django_db
def test_job_relaunch_permission_denied_response(
post, get, inventory, project, credential, net_credential, machine_credential):
jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project, ask_credential_on_launch=True)
jt.credentials.add(machine_credential)
jt_user = User.objects.create(username='jobtemplateuser')
jt.execute_role.members.add(jt_user)
with impersonate(jt_user):
job = jt.create_unified_job()
# User capability is shown for this
r = get(job.get_absolute_url(), jt_user, expect=200)
assert r.data['summary_fields']['user_capabilities']['start']
# Job has prompted extra_credential, launch denied w/ message
job.launch_config.credentials.add(net_credential)
r = post(reverse('api:job_relaunch', kwargs={'pk':job.pk}), {}, jt_user, expect=403)
assert 'launched with prompted fields you do not have access to' in r.data['detail']
@pytest.mark.django_db
def test_job_relaunch_prompts_not_accepted_response(
post, get, inventory, project, credential, net_credential, machine_credential):
jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project)
jt.credentials.add(machine_credential)
jt_user = User.objects.create(username='jobtemplateuser')
@@ -53,8 +73,6 @@ def test_job_relaunch_permission_denied_response(
# Job has prompted extra_credential, launch denied w/ message
job.launch_config.credentials.add(net_credential)
r = post(reverse('api:job_relaunch', kwargs={'pk':job.pk}), {}, jt_user, expect=403)
assert 'launched with prompted fields' in r.data['detail']
assert 'do not have permission' in r.data['detail']
@pytest.mark.django_db
@@ -209,7 +227,8 @@ def test_block_related_unprocessed_events(mocker, organization, project, delete,
status='finished',
finished=time_of_finish,
job_template=job_template,
project=project
project=project,
organization=project.organization
)
view = RelatedJobsPreventDeleteMixin()
time_of_request = time_of_finish + relativedelta(seconds=2)

107
awx/main/tests/functional/api/test_job_template.py
View File

@@ -6,7 +6,7 @@ import pytest
# AWX
from awx.api.serializers import JobTemplateSerializer
from awx.api.versioning import reverse
from awx.main.models import Job, JobTemplate, CredentialType, WorkflowJobTemplate
from awx.main.models import Job, JobTemplate, CredentialType, WorkflowJobTemplate, Organization, Project
from awx.main.migrations import _save_password_keys as save_password_keys
# Django
@@ -30,14 +30,19 @@ def test_create(post, project, machine_credential, inventory, alice, grant_proje
project.use_role.members.add(alice)
if grant_inventory:
inventory.use_role.members.add(alice)
project.organization.job_template_admin_role.members.add(alice)
r = post(reverse('api:job_template_list'), {
'name': 'Some name',
'project': project.id,
'inventory': inventory.id,
'playbook': 'helloworld.yml',
}, alice)
assert r.status_code == expect
post(
url=reverse('api:job_template_list'),
data={
'name': 'Some name',
'project': project.id,
'inventory': inventory.id,
'playbook': 'helloworld.yml'
},
user=alice,
expect=expect
)
@pytest.mark.django_db
@@ -123,14 +128,18 @@ def test_create_with_forks_exceeding_maximum_xfail(alice, post, project, invento
project.use_role.members.add(alice)
inventory.use_role.members.add(alice)
settings.MAX_FORKS = 10
response = post(reverse('api:job_template_list'), {
'name': 'Some name',
'project': project.id,
'inventory': inventory.id,
'playbook': 'helloworld.yml',
'forks': 11,
}, alice)
assert response.status_code == 400
response = post(
url=reverse('api:job_template_list'),
data={
'name': 'Some name',
'project': project.id,
'inventory': inventory.id,
'playbook': 'helloworld.yml',
'forks': 11,
},
user=alice,
expect=400
)
assert 'Maximum number of forks (10) exceeded' in str(response.data)
@@ -510,6 +519,72 @@ def test_job_template_unset_custom_virtualenv(get, patch, organization_factory,
assert resp.data['custom_virtualenv'] is None
@pytest.mark.django_db
def test_jt_organization_follows_project(post, patch, admin_user):
org1 = Organization.objects.create(name='foo1')
org2 = Organization.objects.create(name='foo2')
project_common = dict(scm_type='git', playbook_files=['helloworld.yml'])
project1 = Project.objects.create(name='proj1', organization=org1, **project_common)
project2 = Project.objects.create(name='proj2', organization=org2, **project_common)
r = post(
url=reverse('api:job_template_list'),
data={
"name": "fooo",
"ask_inventory_on_launch": True,
"project": project1.pk,
"playbook": "helloworld.yml"
},
user=admin_user,
expect=201
)
data = r.data
assert data['organization'] == project1.organization_id
data['project'] = project2.id
jt = JobTemplate.objects.get(pk=data['id'])
r = patch(
url=jt.get_absolute_url(),
data=data,
user=admin_user,
expect=200
)
assert r.data['organization'] == project2.organization_id
@pytest.mark.django_db
def test_jt_organization_field_is_read_only(patch, post, project, admin_user):
org = project.organization
jt = JobTemplate.objects.create(
name='foo_jt',
ask_inventory_on_launch=True,
project=project, playbook='helloworld.yml'
)
org2 = Organization.objects.create(name='foo2')
r = patch(
url=jt.get_absolute_url(),
data={'organization': org2.id},
user=admin_user,
expect=200
)
assert r.data['organization'] == org.id
assert JobTemplate.objects.get(pk=jt.pk).organization == org
# similar test, but on creation
r = post(
url=reverse('api:job_template_list'),
data={
'name': 'foobar',
'project': project.id,
'organization': org2.id,
'ask_inventory_on_launch': True,
'playbook': 'helloworld.yml'
},
user=admin_user,
expect=201
)
assert r.data['organization'] == org.id
assert JobTemplate.objects.get(pk=r.data['id']).organization == org
@pytest.mark.django_db
def test_callback_disallowed_null_inventory(project):
jt = JobTemplate.objects.create(

60
awx/main/tests/functional/api/test_organization_counts.py
View File

@@ -2,6 +2,8 @@ import pytest
from awx.api.versioning import reverse
from awx.main.models import Project
@pytest.fixture
def organization_resource_creator(organization, user):
@@ -19,21 +21,26 @@ def organization_resource_creator(organization, user):
for i in range(inventories):
inventory = organization.inventories.create(name="associated-inv %s" % i)
for i in range(projects):
organization.projects.create(name="test-proj %s" % i,
description="test-proj-desc")
Project.objects.create(
name="test-proj %s" % i,
description="test-proj-desc",
organization=organization
)
# Mix up the inventories and projects used by the job templates
i_proj = 0
i_inv = 0
for i in range(job_templates):
project = organization.projects.all()[i_proj]
project = Project.objects.filter(organization=organization)[i_proj]
# project = organization.projects.all()[i_proj]
inventory = organization.inventories.all()[i_inv]
project.jobtemplates.create(name="test-jt %s" % i,
description="test-job-template-desc",
inventory=inventory,
playbook="test_playbook.yml")
playbook="test_playbook.yml",
organization=organization)
i_proj += 1
i_inv += 1
if i_proj >= organization.projects.count():
if i_proj >= Project.objects.filter(organization=organization).count():
i_proj = 0
if i_inv >= organization.inventories.count():
i_inv = 0
@@ -179,12 +186,14 @@ def test_scan_JT_counted(resourced_organization, user, get):
@pytest.mark.django_db
def test_JT_not_double_counted(resourced_organization, user, get):
admin_user = user('admin', True)
proj = Project.objects.filter(organization=resourced_organization).all()[0]
# Add a run job template to the org
resourced_organization.projects.all()[0].jobtemplates.create(
proj.jobtemplates.create(
job_type='run',
inventory=resourced_organization.inventories.all()[0],
project=resourced_organization.projects.all()[0],
name='double-linked-job-template')
project=proj,
name='double-linked-job-template',
organization=resourced_organization)
counts_dict = COUNTS_PRIMES
counts_dict['job_templates'] += 1
@@ -197,38 +206,3 @@ def test_JT_not_double_counted(resourced_organization, user, get):
detail_response = get(reverse('api:organization_detail', kwargs={'pk': resourced_organization.pk}), admin_user)
assert detail_response.status_code == 200
assert detail_response.data['summary_fields']['related_field_counts'] == counts_dict
@pytest.mark.django_db
def test_JT_associated_with_project(organizations, project, user, get):
# Check that adding a project to an organization gets the project's JT
# included in the organization's JT count
external_admin = user('admin', True)
two_orgs = organizations(2)
organization = two_orgs[0]
other_org = two_orgs[1]
unrelated_inv = other_org.inventories.create(name='not-in-organization')
organization.projects.add(project)
project.jobtemplates.create(name="test-jt",
description="test-job-template-desc",
inventory=unrelated_inv,
playbook="test_playbook.yml")
response = get(reverse('api:organization_list'), external_admin)
assert response.status_code == 200
org_id = organization.id
counts = {}
for org_json in response.data['results']:
working_id = org_json['id']
counts[working_id] = org_json['summary_fields']['related_field_counts']
assert counts[org_id] == {
'users': 0,
'admins': 0,
'job_templates': 1,
'projects': 1,
'inventories': 0,
'teams': 0
}

74
awx/main/tests/functional/api/test_unified_job_template.py
View File

@@ -1,6 +1,7 @@
import pytest
from awx.api.versioning import reverse
from awx.main import models
@pytest.mark.django_db
@@ -9,3 +10,76 @@ def test_aliased_forward_reverse_field_searches(instance, options, get, admin):
response = options(url, None, admin)
assert 'job_template__search' in response.data['related_search_fields']
get(reverse("api:unified_job_template_list") + "?job_template__search=anything", user=admin, expect=200)
@pytest.mark.django_db
@pytest.mark.parametrize('model', (
'Project',
'JobTemplate',
'WorkflowJobTemplate'
))
class TestUnifiedOrganization:
def data_for_model(self, model, orm_style=False):
data = {
'name': 'foo',
'organization': None
}
if model == 'JobTemplate':
proj = models.Project.objects.create(
name="test-proj",
playbook_files=['helloworld.yml']
)
if orm_style:
data['project_id'] = proj.id
else:
data['project'] = proj.id
data['playbook'] = 'helloworld.yml'
data['ask_inventory_on_launch'] = True
return data
def test_organization_blank_on_edit_of_orphan(self, model, admin_user, patch):
cls = getattr(models, model)
data = self.data_for_model(model, orm_style=True)
obj = cls.objects.create(**data)
patch(
url=obj.get_absolute_url(),
data={'name': 'foooooo'},
user=admin_user,
expect=200
)
obj.refresh_from_db()
assert obj.name == 'foooooo'
def test_organization_blank_on_edit_of_orphan_as_nonsuperuser(self, model, rando, patch):
"""Test case reflects historical bug where ordinary users got weird error
message when editing an orphaned project
"""
cls = getattr(models, model)
data = self.data_for_model(model, orm_style=True)
obj = cls.objects.create(**data)
if model == 'JobTemplate':
obj.project.admin_role.members.add(rando)
obj.admin_role.members.add(rando)
patch(
url=obj.get_absolute_url(),
data={'name': 'foooooo'},
user=rando,
expect=200
)
obj.refresh_from_db()
assert obj.name == 'foooooo'
def test_organization_blank_on_edit_of_normal(self, model, admin_user, patch, organization):
cls = getattr(models, model)
data = self.data_for_model(model, orm_style=True)
data['organization'] = organization
obj = cls.objects.create(**data)
patch(
url=obj.get_absolute_url(),
data={'name': 'foooooo'},
user=admin_user,
expect=200
)
obj.refresh_from_db()
assert obj.name == 'foooooo'

27
awx/main/tests/functional/conftest.py
View File

@@ -180,8 +180,8 @@ def project_factory(organization):
@pytest.fixture
def job_factory(job_template, admin):
def factory(job_template=job_template, initial_state='new', created_by=admin):
def job_factory(jt_linked, admin):
def factory(job_template=jt_linked, initial_state='new', created_by=admin):
return job_template.create_unified_job(_eager_fields={
'status': initial_state, 'created_by': created_by})
return factory
@@ -701,11 +701,8 @@ def ad_hoc_command_factory(inventory, machine_credential, admin):
@pytest.fixture
def job_template(organization):
jt = JobTemplate(name='test-job_template')
jt.save()
return jt
def job_template():
return JobTemplate.objects.create(name='test-job_template')
@pytest.fixture
@@ -717,20 +714,16 @@ def job_template_labels(organization, job_template):
@pytest.fixture
def jt_linked(job_template_factory, credential, net_credential, vault_credential):
def jt_linked(organization, project, inventory, machine_credential, credential, net_credential, vault_credential):
'''
A job template with a reasonably complete set of related objects to
test RBAC and other functionality affected by related objects
'''
objects = job_template_factory(
'testJT', organization='org1', project='proj1', inventory='inventory1',
credential='cred1')
jt = objects.job_template
jt.credentials.add(vault_credential)
jt.save()
# Add AWS cloud credential and network credential
jt.credentials.add(credential)
jt.credentials.add(net_credential)
jt = JobTemplate.objects.create(
project=project, inventory=inventory, playbook='helloworld.yml',
organization=organization
)
jt.credentials.add(machine_credential, vault_credential, credential, net_credential)
return jt

5
awx/main/tests/functional/models/test_activity_stream.py
View File

@@ -12,6 +12,7 @@ from awx.main.models import (
CredentialType,
Inventory,
InventorySource,
Project,
User
)
@@ -99,8 +100,8 @@ class TestRolesAssociationEntries:
).count() == 1, 'In loop %s' % i
def test_model_associations_are_recorded(self, organization):
proj1 = organization.projects.create(name='proj1')
proj2 = organization.projects.create(name='proj2')
proj1 = Project.objects.create(name='proj1', organization=organization)
proj2 = Project.objects.create(name='proj2', organization=organization)
proj2.use_role.parents.add(proj1.admin_role)
assert ActivityStream.objects.filter(role=proj1.admin_role, project=proj2).count() == 1

30
awx/main/tests/functional/models/test_job.py
View File

@@ -1,6 +1,9 @@
import pytest
from awx.main.models import JobTemplate, Job, JobHostSummary, WorkflowJob, Inventory
from awx.main.models import (
JobTemplate, Job, JobHostSummary,
WorkflowJob, Inventory, Project, Organization
)
@pytest.mark.django_db
@@ -29,18 +32,19 @@ def test_prevent_slicing():
@pytest.mark.django_db
def test_awx_custom_virtualenv(inventory, project, machine_credential):
def test_awx_custom_virtualenv(inventory, project, machine_credential, organization):
jt = JobTemplate.objects.create(
name='my-jt',
inventory=inventory,
project=project,
playbook='helloworld.yml'
playbook='helloworld.yml',
organization=organization
)
jt.credentials.add(machine_credential)
job = jt.create_unified_job()
job.project.organization.custom_virtualenv = '/venv/fancy-org'
job.project.organization.save()
job.organization.custom_virtualenv = '/venv/fancy-org'
job.organization.save()
assert job.ansible_virtualenv_path == '/venv/fancy-org'
job.project.custom_virtualenv = '/venv/fancy-proj'
@@ -78,6 +82,22 @@ def test_job_host_summary_representation(host):
assert 'N/A changed=1 dark=2 failures=3 ignored=4 ok=5 processed=6 rescued=7 skipped=8' == str(jhs)
@pytest.mark.django_db
def test_jt_organization_follows_project():
org1 = Organization.objects.create(name='foo1')
org2 = Organization.objects.create(name='foo2')
project1 = Project.objects.create(name='proj1', organization=org1)
project2 = Project.objects.create(name='proj2', organization=org2)
jt = JobTemplate.objects.create(
name='foo', playbook='helloworld.yml',
project=project1
)
assert jt.organization == org1
jt.project = project2
jt.save()
assert JobTemplate.objects.get(pk=jt.id).organization == org2
@pytest.mark.django_db
class TestSlicingModels:

6
awx/main/tests/functional/models/test_project.py
View File

@@ -39,3 +39,9 @@ def test_foreign_key_change_changes_modified_by(project, organization):
assert project._get_fields_snapshot()['organization_id'] == organization.id
project.organization = Organization(name='foo', pk=41)
assert project._get_fields_snapshot()['organization_id'] == 41
@pytest.mark.django_db
def test_project_related_jobs(project):
update = project.create_unified_job()
assert update.id in [u.id for u in project._get_related_jobs()]

24
awx/main/tests/functional/models/test_unified_job.py
View File

@@ -13,6 +13,7 @@ from awx.main.models import (
WorkflowApprovalTemplate, Project, WorkflowJob, Schedule,
Credential
)
from awx.api.versioning import reverse
@pytest.mark.django_db
@@ -26,6 +27,29 @@ def test_subclass_types(rando):
])
@pytest.mark.django_db
def test_soft_unique_together(post, project, admin_user):
"""This tests that SOFT_UNIQUE_TOGETHER restrictions are applied correctly.
"""
jt1 = JobTemplate.objects.create(
name='foo_jt',
project=project
)
assert jt1.organization == project.organization
r = post(
url=reverse('api:job_template_list'),
data=dict(
name='foo_jt', # same as first
project=project.id,
ask_inventory_on_launch=True,
playbook='helloworld.yml'
),
user=admin_user,
expect=400
)
assert 'combination already exists' in str(r.data)
@pytest.mark.django_db
class TestCreateUnifiedJob:
'''

6
awx/main/tests/functional/test_instances.py
View File

@@ -1,7 +1,7 @@
import pytest
from unittest import mock
from awx.main.models import AdHocCommand, InventoryUpdate, Job, JobTemplate, ProjectUpdate
from awx.main.models import AdHocCommand, InventoryUpdate, JobTemplate, ProjectUpdate
from awx.main.models.ha import Instance, InstanceGroup
from awx.main.tasks import apply_cluster_membership_policies
from awx.api.versioning import reverse
@@ -310,7 +310,7 @@ class TestInstanceGroupOrdering:
assert iu.preferred_instance_groups == [ig_inv, ig_org]
def test_project_update_instance_groups(self, instance_group_factory, project, default_instance_group):
pu = ProjectUpdate.objects.create(project=project)
pu = ProjectUpdate.objects.create(project=project, organization=project.organization)
assert pu.preferred_instance_groups == [default_instance_group]
ig_org = instance_group_factory("OrgIstGrp", [default_instance_group.instances.first()])
ig_tmp = instance_group_factory("TmpIstGrp", [default_instance_group.instances.first()])
@@ -321,7 +321,7 @@ class TestInstanceGroupOrdering:
def test_job_instance_groups(self, instance_group_factory, inventory, project, default_instance_group):
jt = JobTemplate.objects.create(inventory=inventory, project=project)
job = Job.objects.create(inventory=inventory, job_template=jt, project=project)
job = jt.create_unified_job()
assert job.preferred_instance_groups == [default_instance_group]
ig_org = instance_group_factory("OrgIstGrp", [default_instance_group.instances.first()])
ig_inv = instance_group_factory("InvIstGrp", [default_instance_group.instances.first()])

22
awx/main/tests/functional/test_named_url.py
View File

@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
import pytest
from django.core.exceptions import ImproperlyConfigured
@@ -26,7 +27,7 @@ def setup_module(module):
def teardown_module(module):
# settings_registry will be persistent states unless we explicitly clean them up.
# settings_registry will be persistent states unless we explicitly clean them up.
settings_registry.unregister('NAMED_URL_FORMATS')
settings_registry.unregister('NAMED_URL_GRAPH_NODES')
@@ -58,10 +59,25 @@ def test_organization(get, admin_user):
@pytest.mark.django_db
def test_job_template(get, admin_user):
test_jt = JobTemplate.objects.create(name='test_jt')
test_org = Organization.objects.create(name='test_org')
test_jt = JobTemplate.objects.create(name='test_jt', organization=test_org)
url = reverse('api:job_template_detail', kwargs={'pk': test_jt.pk})
response = get(url, user=admin_user, expect=200)
assert response.data['related']['named_url'].endswith('/test_jt/')
assert response.data['related']['named_url'].endswith('/test_jt++test_org/')
@pytest.mark.django_db
def test_job_template_old_way(get, admin_user, mocker):
test_org = Organization.objects.create(name='test_org')
test_jt = JobTemplate.objects.create(name='test_jt ♥', organization=test_org)
url = reverse('api:job_template_detail', kwargs={'pk': test_jt.pk})
response = get(url, user=admin_user, expect=200)
new_url = response.data['related']['named_url']
old_url = '/'.join([url.rsplit('/', 2)[0], test_jt.name, ''])
assert URLModificationMiddleware._convert_named_url(new_url) == url
assert URLModificationMiddleware._convert_named_url(old_url) == url
@pytest.mark.django_db

26
awx/main/tests/functional/test_projects.py
View File

@@ -213,34 +213,14 @@ def test_project_credential_protection(post, put, project, organization, scm_cre
}, org_admin, expect=403
)
post(
reverse('api:project_list'), {
'name': 'should not create',
'organization':organization.id,
reverse('api:project_list'), {
'name': 'should not create',
'organization':organization.id,
'credential': scm_credential.id
}, org_admin, expect=403
)
@pytest.mark.django_db()
def test_create_project_null_organization(post, organization, admin):
post(reverse('api:project_list'), { 'name': 't', 'organization': None}, admin, expect=201)
@pytest.mark.django_db()
def test_create_project_null_organization_xfail(post, organization, org_admin):
post(reverse('api:project_list'), { 'name': 't', 'organization': None}, org_admin, expect=403)
@pytest.mark.django_db()
def test_patch_project_null_organization(patch, organization, project, admin):
patch(reverse('api:project_detail', kwargs={'pk':project.id,}), { 'name': 't', 'organization': organization.id}, admin, expect=200)
@pytest.mark.django_db()
def test_patch_project_null_organization_xfail(patch, project, org_admin):
patch(reverse('api:project_detail', kwargs={'pk':project.id,}), { 'name': 't', 'organization': None}, org_admin, expect=400)
@pytest.mark.django_db
def test_cannot_schedule_manual_project(manual_project, admin_user, post):
response = post(

21
awx/main/tests/functional/test_rbac_job.py
View File

@@ -1,5 +1,7 @@
import pytest
from rest_framework.exceptions import PermissionDenied
from awx.main.access import (
JobAccess,
JobLaunchConfigAccess,
@@ -19,8 +21,6 @@ from awx.main.models import (
Credential
)
from rest_framework.exceptions import PermissionDenied
from crum import impersonate
@@ -29,7 +29,8 @@ def normal_job(deploy_jobtemplate):
return Job.objects.create(
job_template=deploy_jobtemplate,
project=deploy_jobtemplate.project,
inventory=deploy_jobtemplate.inventory
inventory=deploy_jobtemplate.inventory,
organization=deploy_jobtemplate.organization
)
@@ -170,9 +171,11 @@ class TestJobRelaunchAccess:
machine_credential.use_role.members.add(u)
access = JobAccess(u)
assert access.can_start(job_with_links, validate_license=False) == can_start, (
"Inventory access: {}\nCredential access: {}\n Expected access: {}".format(inv_access, cred_access, can_start)
)
if can_start:
assert access.can_start(job_with_links, validate_license=False)
else:
with pytest.raises(PermissionDenied):
access.can_start(job_with_links, validate_license=False)
def test_job_relaunch_credential_access(
self, inventory, project, credential, net_credential):
@@ -187,7 +190,8 @@ class TestJobRelaunchAccess:
# Job has prompted net credential, launch denied w/ message
job = jt.create_unified_job(credentials=[net_credential])
assert not jt_user.can_access(Job, 'start', job, validate_license=False)
with pytest.raises(PermissionDenied):
jt_user.can_access(Job, 'start', job, validate_license=False)
def test_prompted_credential_relaunch_denied(
self, inventory, project, net_credential, rando):
@@ -200,7 +204,8 @@ class TestJobRelaunchAccess:
# Job has prompted net credential, rando lacks permission to use it
job = jt.create_unified_job(credentials=[net_credential])
assert not rando.can_access(Job, 'start', job, validate_license=False)
with pytest.raises(PermissionDenied):
rando.can_access(Job, 'start', job, validate_license=False)
def test_prompted_credential_relaunch_allowed(
self, inventory, project, net_credential, rando):

26
awx/main/tests/functional/test_rbac_job_start.py
View File

@@ -1,5 +1,7 @@
import pytest
from rest_framework.exceptions import PermissionDenied
from awx.main.models.inventory import Inventory
from awx.main.models.credential import Credential
from awx.main.models.jobs import JobTemplate, Job
@@ -89,8 +91,8 @@ def test_slice_job(slice_job_factory, rando):
@pytest.mark.django_db
class TestJobRelaunchAccess:
@pytest.fixture
def job_no_prompts(self, machine_credential, inventory):
jt = JobTemplate.objects.create(name='test-job_template', inventory=inventory)
def job_no_prompts(self, machine_credential, inventory, organization):
jt = JobTemplate.objects.create(name='test-job_template', inventory=inventory, organization=organization)
jt.credentials.add(machine_credential)
return jt.create_unified_job()
@@ -119,10 +121,20 @@ class TestJobRelaunchAccess:
job_no_prompts.job_template.execute_role.members.add(rando)
assert rando.can_access(Job, 'start', job_no_prompts)
def test_orphan_relaunch_via_organization(self, job_no_prompts, rando, organization):
"JT for job has been deleted, relevant organization roles will allow management"
assert job_no_prompts.organization == organization
organization.execute_role.members.add(rando)
job_no_prompts.job_template.delete()
job_no_prompts.job_template = None # Django should do this for us, but it does not
assert rando.can_access(Job, 'start', job_no_prompts)
def test_no_relaunch_without_prompted_fields_access(self, job_with_prompts, rando):
"Has JT execute_role but no use_role on inventory & credential - deny relaunch"
job_with_prompts.job_template.execute_role.members.add(rando)
assert not rando.can_access(Job, 'start', job_with_prompts)
with pytest.raises(PermissionDenied) as exc:
rando.can_access(Job, 'start', job_with_prompts)
assert 'Job was launched with prompted fields you do not have access to' in str(exc)
def test_can_relaunch_with_prompted_fields_access(self, job_with_prompts, rando):
"Has use_role on the prompted inventory & credential - allow relaunch"
@@ -141,11 +153,15 @@ class TestJobRelaunchAccess:
jt.ask_limit_on_launch = False
jt.save()
jt.execute_role.members.add(rando)
assert not rando.can_access(Job, 'start', job_with_prompts)
with pytest.raises(PermissionDenied):
rando.can_access(Job, 'start', job_with_prompts)
def test_can_relaunch_if_limit_was_prompt(self, job_with_prompts, rando):
"Job state differs from JT, but only on prompted fields - allow relaunch"
job_with_prompts.job_template.execute_role.members.add(rando)
job_with_prompts.limit = 'webservers'
job_with_prompts.save()
assert not rando.can_access(Job, 'start', job_with_prompts)
job_with_prompts.inventory.use_role.members.add(rando)
for cred in job_with_prompts.credentials.all():
cred.use_role.members.add(rando)
assert rando.can_access(Job, 'start', job_with_prompts)

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

@@ -8,8 +8,7 @@ from awx.main.access import (
ScheduleAccess
)
from awx.main.models.jobs import JobTemplate
from awx.main.models.organization import Organization
from awx.main.models.schedules import Schedule
from awx.main.models import Project, Organization, Inventory, Schedule, User
@mock.patch.object(BaseAccess, 'check_license', return_value=None)
@@ -24,6 +23,29 @@ def test_job_template_access_superuser(check_license, user, deploy_jobtemplate):
assert access.can_add({})
@pytest.mark.django_db
class TestImplicitAccess:
def test_org_execute(self, jt_linked, rando):
assert rando not in jt_linked.execute_role
jt_linked.organization.execute_role.members.add(rando)
assert rando in jt_linked.execute_role
def test_org_admin(self, jt_linked, rando):
assert rando not in jt_linked.execute_role
jt_linked.organization.job_template_admin_role.members.add(rando)
assert rando in jt_linked.execute_role
def test_org_auditor(self, jt_linked, rando):
assert rando not in jt_linked.read_role
jt_linked.organization.auditor_role.members.add(rando)
assert rando in jt_linked.read_role
def test_deprecated_inventory_read(self, jt_linked, rando):
assert rando not in jt_linked.read_role
jt_linked.inventory.organization.execute_role.members.add(rando)
assert rando in jt_linked.read_role
@pytest.mark.django_db
def test_job_template_access_read_level(jt_linked, rando):
ssh_cred = jt_linked.machine_credential
@@ -45,22 +67,21 @@ def test_job_template_access_read_level(jt_linked, rando):
@pytest.mark.django_db
def test_job_template_access_use_level(jt_linked, rando):
ssh_cred = jt_linked.machine_credential
vault_cred = jt_linked.vault_credentials[0]
access = JobTemplateAccess(rando)
jt_linked.project.use_role.members.add(rando)
jt_linked.inventory.use_role.members.add(rando)
ssh_cred.use_role.members.add(rando)
vault_cred.use_role.members.add(rando)
jt_linked.organization.job_template_admin_role.members.add(rando)
proj_pk = jt_linked.project.pk
assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk))
assert access.can_add(dict(credential=ssh_cred.pk, project=proj_pk))
assert access.can_add(dict(vault_credential=vault_cred.pk, project=proj_pk))
org_pk = jt_linked.organization_id
assert access.can_change(jt_linked, {'job_type': 'check', 'project': proj_pk})
assert access.can_change(jt_linked, {'job_type': 'check', 'inventory': None})
for cred in jt_linked.credentials.all():
assert not access.can_unattach(jt_linked, cred, 'credentials', {})
assert access.can_unattach(jt_linked, cred, 'credentials', {})
assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk, organization=org_pk))
assert access.can_add(dict(project=proj_pk, organization=org_pk))
@pytest.mark.django_db
@@ -69,22 +90,21 @@ def test_job_template_access_admin(role_names, jt_linked, rando):
ssh_cred = jt_linked.machine_credential
access = JobTemplateAccess(rando)
# Appoint this user as admin of the organization
#jt_linked.inventory.organization.admin_role.members.add(rando)
assert not access.can_read(jt_linked)
assert not access.can_delete(jt_linked)
for role_name in role_names:
role = getattr(jt_linked.inventory.organization, role_name)
role.members.add(rando)
# Appoint this user as admin of the organization
jt_linked.organization.admin_role.members.add(rando)
org_pk = jt_linked.organization.id
# Assign organization permission in the same way the create view does
organization = jt_linked.inventory.organization
ssh_cred.admin_role.parents.add(organization.admin_role)
proj_pk = jt_linked.project.pk
assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk))
assert access.can_add(dict(credential=ssh_cred.pk, project=proj_pk))
assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk, organization=org_pk))
assert access.can_add(dict(credential=ssh_cred.pk, project=proj_pk, organization=org_pk))
for cred in jt_linked.credentials.all():
assert access.can_unattach(jt_linked, cred, 'credentials', {})
@@ -105,11 +125,11 @@ def test_job_template_extra_credentials_prompts_access(
)
jt.credentials.add(machine_credential)
jt.execute_role.members.add(rando)
r = post(
post(
reverse('api:job_template_launch', kwargs={'pk': jt.id}),
{'credentials': [machine_credential.pk, vault_credential.pk]}, rando
{'credentials': [machine_credential.pk, vault_credential.pk]}, rando,
expect=403
)
assert r.status_code == 403
@pytest.mark.django_db
@@ -148,26 +168,41 @@ class TestOrphanJobTemplate:
@pytest.mark.django_db
@pytest.mark.job_permissions
def test_job_template_creator_access(project, rando, post):
def test_job_template_creator_access(project, organization, rando, post):
project.use_role.members.add(rando)
organization.job_template_admin_role.members.add(rando)
response = post(url=reverse('api:job_template_list'), data=dict(
name='newly-created-jt',
ask_inventory_on_launch=True,
project=project.pk,
organization=organization.id,
playbook='helloworld.yml'
), user=rando, expect=201)
project.admin_role.members.add(rando)
with mock.patch(
'awx.main.models.projects.ProjectOptions.playbooks',
new_callable=mock.PropertyMock(return_value=['helloworld.yml'])):
response = post(reverse('api:job_template_list'), dict(
name='newly-created-jt',
job_type='run',
ask_inventory_on_launch=True,
ask_credential_on_launch=True,
project=project.pk,
playbook='helloworld.yml'
), rando)
assert response.status_code == 201
jt_pk = response.data['id']
jt_obj = JobTemplate.objects.get(pk=jt_pk)
# Creating a JT should place the creator in the admin role
assert rando in jt_obj.admin_role
assert rando in jt_obj.admin_role.members.all()
@pytest.mark.django_db
@pytest.mark.job_permissions
@pytest.mark.parametrize('lacking', ['project', 'inventory'])
def test_job_template_insufficient_creator_permissions(lacking, project, inventory, organization, rando, post):
if lacking != 'project':
project.use_role.members.add(rando)
else:
project.read_role.members.add(rando)
if lacking != 'inventory':
inventory.use_role.members.add(rando)
else:
inventory.read_role.members.add(rando)
post(url=reverse('api:job_template_list'), data=dict(
name='newly-created-jt',
inventory=inventory.id,
project=project.pk,
playbook='helloworld.yml'
), user=rando, expect=403)
@pytest.mark.django_db
@@ -237,27 +272,104 @@ class TestJobTemplateSchedules:
@pytest.mark.django_db
def test_jt_org_ownership_change(user, jt_linked):
admin1 = user('admin1')
org1 = jt_linked.project.organization
org1.admin_role.members.add(admin1)
a1_access = JobTemplateAccess(admin1)
class TestProjectOrganization:
"""Tests stories related to management of JT organization via its project
which have some bearing on RBAC integrity
"""
assert a1_access.can_read(jt_linked)
def test_new_project_org_change(self, project, patch, admin_user):
org2 = Organization.objects.create(name='bar')
patch(
url=project.get_absolute_url(),
data={'organization': org2.id},
user=admin_user,
expect=200
)
assert Project.objects.get(pk=project.id).organization_id == org2.id
def test_jt_org_cannot_change(self, project, post, patch, admin_user):
post(
url=reverse('api:job_template_list'),
data={
'name': 'foo_template',
'project': project.id,
'playbook': 'helloworld.yml',
'ask_inventory_on_launch': True
},
user=admin_user,
expect=201
)
org2 = Organization.objects.create(name='bar')
r = patch(
url=project.get_absolute_url(),
data={'organization': org2.id},
user=admin_user,
expect=400
)
assert 'Organization cannot be changed' in str(r.data)
admin2 = user('admin2')
org2 = Organization.objects.create(name='mrroboto', description='domo')
org2.admin_role.members.add(admin2)
a2_access = JobTemplateAccess(admin2)
def test_orphan_JT_adoption(self, project, patch, admin_user, org_admin):
jt = JobTemplate.objects.create(
name='bar',
ask_inventory_on_launch=True,
playbook='helloworld.yml'
)
assert org_admin not in jt.admin_role
patch(
url=jt.get_absolute_url(),
data={'project': project.id},
user=admin_user,
expect=200
)
assert org_admin in jt.admin_role
assert not a2_access.can_read(jt_linked)
def test_inventory_read_transfer_direct(self, patch):
orgs = []
invs = []
admins = []
for i in range(2):
org = Organization.objects.create(name='org{}'.format(i))
org_admin = User.objects.create(username='user{}'.format(i))
inv = Inventory.objects.create(
organization=org,
name='inv{}'.format(i)
)
org.auditor_role.members.add(org_admin)
orgs.append(org)
admins.append(org_admin)
invs.append(inv)
jt_linked.project.organization = org2
jt_linked.project.save()
jt_linked.inventory.organization = org2
jt_linked.inventory.save()
jt = JobTemplate.objects.create(name='foo', inventory=invs[0])
assert admins[0] in jt.read_role
assert admins[1] not in jt.read_role
assert a2_access.can_read(jt_linked)
assert not a1_access.can_read(jt_linked)
jt.inventory = invs[1]
jt.save(update_fields=['inventory'])
assert admins[0] not in jt.read_role
assert admins[1] in jt.read_role
def test_inventory_read_transfer_indirect(self, patch):
orgs = []
admins = []
for i in range(2):
org = Organization.objects.create(name='org{}'.format(i))
org_admin = User.objects.create(username='user{}'.format(i))
org.auditor_role.members.add(org_admin)
orgs.append(org)
admins.append(org_admin)
inv = Inventory.objects.create(
organization=orgs[0],
name='inv{}'.format(i)
)
jt = JobTemplate.objects.create(name='foo', inventory=inv)
assert admins[0] in jt.read_role
assert admins[1] not in jt.read_role
inv.organization = orgs[1]
inv.save(update_fields=['organization'])
assert admins[0] not in jt.read_role
assert admins[1] in jt.read_role

101
awx/main/tests/functional/test_rbac_migration.py Normal file
View File

@@ -0,0 +1,101 @@
import pytest
from django.apps import apps
from awx.main.migrations import _rbac as rbac
from awx.main.models import (
UnifiedJobTemplate,
InventorySource, Inventory,
JobTemplate, Project,
Organization,
User
)
@pytest.mark.django_db
def test_implied_organization_subquery_inventory():
orgs = []
for i in range(3):
orgs.append(Organization.objects.create(name='foo{}'.format(i)))
orgs.append(orgs[0])
for i in range(4):
org = orgs[i]
if i == 2:
inventory = Inventory.objects.create(name='foo{}'.format(i))
else:
inventory = Inventory.objects.create(name='foo{}'.format(i), organization=org)
inv_src = InventorySource.objects.create(name='foo{}'.format(i), inventory=inventory)
sources = UnifiedJobTemplate.objects.annotate(
test_field=rbac.implicit_org_subquery(UnifiedJobTemplate, InventorySource)
)
for inv_src in sources:
assert inv_src.test_field == inv_src.inventory.organization_id
@pytest.mark.django_db
def test_implied_organization_subquery_job_template():
jts = []
for i in range(5):
if i <= 3:
org = Organization.objects.create(name='foo{}'.format(i))
else:
org = None
if i <= 4:
proj = Project.objects.create(
name='foo{}'.format(i),
organization=org
)
else:
proj = None
jts.append(JobTemplate.objects.create(
name='foo{}'.format(i),
project=proj
))
# test case of sharing same org
jts[2].project.organization = jts[3].project.organization
jts[2].save()
ujts = UnifiedJobTemplate.objects.annotate(
test_field=rbac.implicit_org_subquery(UnifiedJobTemplate, JobTemplate)
)
for jt in ujts:
if not isinstance(jt, JobTemplate): # some are projects
assert jt.test_field is None
else:
if jt.project is None:
assert jt.test_field is None
else:
assert jt.test_field == jt.project.organization_id
@pytest.mark.django_db
def test_give_explicit_inventory_permission():
dual_admin = User.objects.create(username='alice')
inv_admin = User.objects.create(username='bob')
inv_org = Organization.objects.create(name='inv-org')
proj_org = Organization.objects.create(name='proj-org')
inv_org.admin_role.members.add(inv_admin, dual_admin)
proj_org.admin_role.members.add(dual_admin)
proj = Project.objects.create(
name="test-proj",
organization=proj_org
)
inv = Inventory.objects.create(
name='test-inv',
organization=inv_org
)
jt = JobTemplate.objects.create(
name='foo',
project=proj,
inventory=inv
)
assert dual_admin in jt.admin_role
rbac.restore_inventory_admins(apps, None)
assert inv_admin in jt.admin_role.members.all()
assert dual_admin not in jt.admin_role.members.all()
assert dual_admin in jt.admin_role

5
awx/main/tests/functional/test_rbac_workflow.py
View File

@@ -62,10 +62,11 @@ class TestWorkflowJobTemplateAccess:
@pytest.mark.django_db
class TestWorkflowJobTemplateNodeAccess:
def test_no_jt_access_to_edit(self, wfjt_node, org_admin):
def test_no_jt_access_to_edit(self, wfjt_node, rando):
# without access to the related job template, admin to the WFJT can
# not change the prompted parameters
access = WorkflowJobTemplateNodeAccess(org_admin)
wfjt_node.workflow_job_template.admin_role.members.add(rando)
access = WorkflowJobTemplateNodeAccess(rando)
assert not access.can_change(wfjt_node, {'job_type': 'check'})
def test_node_edit_allowed(self, wfjt_node, org_admin):

1
awx/main/tests/unit/api/serializers/test_job_template_serializers.py
View File

@@ -30,6 +30,7 @@ def job_template(mocker):
mock_jt.host_config_key = '9283920492'
mock_jt.validation_errors = mock_JT_resource_data
mock_jt.webhook_service = ''
mock_jt.organization_id = None
return mock_jt

11
awx/main/tests/unit/models/test_unified_job_unit.py
View File

@@ -6,6 +6,7 @@ from awx.main.models import (
UnifiedJobTemplate,
WorkflowJob,
WorkflowJobNode,
WorkflowApprovalTemplate,
Job,
User,
Project,
@@ -65,6 +66,16 @@ def test_cancel_job_explanation(unified_job):
unified_job.save.assert_called_with(update_fields=['cancel_flag', 'start_args', 'status', 'job_explanation'])
def test_organization_copy_to_jobs():
'''
All unified job types should infer their organization from their template organization
'''
for cls in UnifiedJobTemplate.__subclasses__():
if cls is WorkflowApprovalTemplate:
continue # these do not track organization
assert 'organization' in cls._get_unified_job_field_names(), cls
def test_log_representation():
'''
Common representation used inside of log messages

29
awx/main/tests/unit/test_access.py
View File

@@ -148,7 +148,9 @@ def job_template_with_ids(job_template_factory):
'testJT', project=proj, inventory=inv, credential=credential,
cloud_credential=cloud_cred, network_credential=net_cred,
persisted=False)
return jt_objects.job_template
jt = jt_objects.job_template
jt.organization = Organization(id=1, pk=1, name='fooOrg')
return jt
def test_superuser(mocker):
@@ -180,21 +182,24 @@ def test_jt_existing_values_are_nonsensitive(job_template_with_ids, user_unit):
def test_change_jt_sensitive_data(job_template_with_ids, mocker, user_unit):
"""Assure that can_add is called with all ForeignKeys."""
job_template_with_ids.admin_role = Role()
class RoleReturnsTrue(Role):
def __contains__(self, accessor):
return True
job_template_with_ids.admin_role = RoleReturnsTrue()
job_template_with_ids.organization.job_template_admin_role = RoleReturnsTrue()
inv2 = Inventory()
inv2.use_role = RoleReturnsTrue()
data = {'inventory': inv2}
data = {'inventory': job_template_with_ids.inventory.id + 1}
access = JobTemplateAccess(user_unit)
mock_add = mock.MagicMock(return_value=False)
with mock.patch('awx.main.models.rbac.Role.__contains__', return_value=True):
with mocker.patch('awx.main.access.JobTemplateAccess.can_add', mock_add):
with mocker.patch('awx.main.access.JobTemplateAccess.can_read', return_value=True):
assert not access.can_change(job_template_with_ids, data)
assert not access.changes_are_non_sensitive(job_template_with_ids, data)
mock_add.assert_called_once_with({
'inventory': data['inventory'],
'project': job_template_with_ids.project.id
})
job_template_with_ids.inventory.use_role = RoleReturnsTrue()
job_template_with_ids.project.use_role = RoleReturnsTrue()
assert access.can_change(job_template_with_ids, data)
def mock_raise_none(self, add_host=False, feature=None, check_expiration=True):

63
awx/main/tests/unit/test_fields.py
View File

@@ -2,10 +2,17 @@
import pytest
from django.core.exceptions import ValidationError
from django.apps import apps
from django.db.models.fields.related import ForeignKey
from django.db.models.fields.related_descriptors import (
ReverseManyToOneDescriptor,
ForwardManyToOneDescriptor
)
from rest_framework.serializers import ValidationError as DRFValidationError
from awx.main.models import Credential, CredentialType, BaseModel
from awx.main.fields import JSONSchemaField
from awx.main.fields import JSONSchemaField, ImplicitRoleField, ImplicitRoleDescriptor
@pytest.mark.parametrize('schema, given, message', [
@@ -194,3 +201,57 @@ def test_credential_creation_validation_failure(inputs):
with pytest.raises(Exception) as e:
field.validate(inputs, cred)
assert e.type in (ValidationError, DRFValidationError)
def test_implicit_role_field_parents():
"""This assures that every ImplicitRoleField only references parents
which are relationships that actually exist
"""
app_models = apps.get_app_config('main').get_models()
for cls in app_models:
for field in cls._meta.get_fields():
if not isinstance(field, ImplicitRoleField):
continue
if not field.parent_role:
continue
field_names = field.parent_role
if type(field_names) is not list:
field_names = [field_names]
for field_name in field_names:
# this type of specification appears to have been considered
# at some point, but does not exist in the app and would
# need support and tests built out for it
assert not isinstance(field_name, tuple)
# also used to be a thing before py3 upgrade
assert not isinstance(field_name, bytes)
# this is always coherent
if field_name.startswith('singleton:'):
continue
# separate out parent role syntax
field_name, sep, field_attr = field_name.partition('.')
# now make primary assertion, that specified paths exist
assert hasattr(cls, field_name)
# inspect in greater depth
second_field = cls._meta.get_field(field_name)
second_field_descriptor = getattr(cls, field_name)
# all supported linkage types
assert isinstance(second_field_descriptor, (
ReverseManyToOneDescriptor, # not currently used
ImplicitRoleDescriptor,
ForwardManyToOneDescriptor
))
# only these links are supported
if field_attr:
if isinstance(second_field_descriptor, ReverseManyToOneDescriptor):
assert type(second_field) is ForeignKey
rel_model = cls._meta.get_field(field_name).related_model
third_field = getattr(rel_model, field_attr)
# expecting for related_model.foo_role, test role field type
assert isinstance(third_field, ImplicitRoleDescriptor)
else:
# expecting simple format of foo_role
assert type(second_field) is ImplicitRoleField