Merge pull request #6256 from AlanCoding/relaunch_extra_creds

Job relaunch refactor to accommodate new credential system

awx/main/tests/functional/api/test_job.py

@@ -2,8 +2,9 @@ import pytest
 
 from awx.api.versioning import reverse
 
+from awx.main.models import JobTemplate, User
+
 
-# TODO: test this with RBAC and lower-priveleged users
 @pytest.mark.django_db
 def test_extra_credentials(get, organization_factory, job_template_factory, credential):
     objs = organization_factory("org", superusers=['admin'])
@@ -16,3 +17,23 @@ def test_extra_credentials(get, organization_factory, job_template_factory, cred
     url = reverse('api:job_extra_credentials_list', kwargs={'version': 'v2', 'pk': job.pk})
     response = get(url, user=objs.superusers.admin)
     assert response.data.get('count') == 1
+
+
+@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,
+                                    credential=machine_credential)
+    jt_user = User.objects.create(username='jobtemplateuser')
+    jt.execute_role.members.add(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.extra_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']

awx/main/tests/functional/api/test_job_runtime_params.py

@@ -271,43 +271,6 @@ def test_job_block_scan_job_inv_change(mocker, bad_scan_JT, runtime_data, post,
     assert 'inventory' in response.data
 
 
-@pytest.mark.django_db
-@pytest.mark.job_runtime_vars
-def test_job_relaunch_copy_vars(job_with_links, machine_credential, inventory,
-                                deploy_jobtemplate, post, mocker):
-    job_with_links.job_template = deploy_jobtemplate
-    job_with_links.limit = "my_server"
-    with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate._get_unified_job_field_names',
-                      return_value=['inventory', 'credential', 'limit']):
-        second_job = job_with_links.copy_unified_job()
-
-    # Check that job data matches the original variables
-    assert second_job.credential == job_with_links.credential
-    assert second_job.inventory == job_with_links.inventory
-    assert second_job.limit == 'my_server'
-
-
-@pytest.mark.django_db
-@pytest.mark.job_runtime_vars
-def test_job_relaunch_resource_access(job_with_links, user):
-    inventory_user = user('user1', False)
-    credential_user = user('user2', False)
-    both_user = user('user3', False)
-
-    # Confirm that a user with inventory & credential access can launch
-    job_with_links.credential.use_role.members.add(both_user)
-    job_with_links.inventory.use_role.members.add(both_user)
-    assert both_user.can_access(Job, 'start', job_with_links)
-
-    # Confirm that a user with credential access alone cannot launch
-    job_with_links.credential.use_role.members.add(credential_user)
-    assert not credential_user.can_access(Job, 'start', job_with_links)
-
-    # Confirm that a user with inventory access alone cannot launch
-    job_with_links.inventory.use_role.members.add(inventory_user)
-    assert not inventory_user.can_access(Job, 'start', job_with_links)
-
-
 @pytest.mark.django_db
 def test_job_launch_JT_with_validation(machine_credential, deploy_jobtemplate):
     deploy_jobtemplate.extra_vars = '{"job_template_var": 3}'

awx/main/tests/functional/models/test_unified_job.py

@@ -4,7 +4,7 @@ import pytest
 from django.contrib.contenttypes.models import ContentType
 
 # AWX
-from awx.main.models import UnifiedJobTemplate, JobTemplate, WorkflowJobTemplate, Project
+from awx.main.models import UnifiedJobTemplate, Job, JobTemplate, WorkflowJobTemplate, Project
 
 
 @pytest.mark.django_db
@@ -16,11 +16,11 @@ def test_subclass_types(rando):
     ])
 
 
+@pytest.mark.django_db
 class TestCreateUnifiedJob:
     '''
     Ensure that copying a job template to a job handles many to many field copy
     '''
-    @pytest.mark.django_db
     def test_many_to_many(self, mocker, job_template_labels):
         jt = job_template_labels
         _get_unified_job_field_names = mocker.patch('awx.main.models.jobs.JobTemplate._get_unified_job_field_names', return_value=['labels'])
@@ -34,7 +34,6 @@ class TestCreateUnifiedJob:
     '''
     Ensure that data is looked for in parameter list before looking at the object
     '''
-    @pytest.mark.django_db
     def test_many_to_many_kwargs(self, mocker, job_template_labels):
         jt = job_template_labels
         mocked = mocker.MagicMock()
@@ -47,3 +46,22 @@ class TestCreateUnifiedJob:
 
         _get_unified_job_field_names.assert_called_with()
         mocked.all.assert_called_with()
+
+    '''
+    Ensure that extra_credentials m2m field is copied to new relaunched job
+    '''
+    def test_job_relaunch_copy_vars(self, machine_credential, inventory,
+                                    deploy_jobtemplate, post, mocker, net_credential):
+        job_with_links = Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory)
+        job_with_links.job_template = deploy_jobtemplate
+        job_with_links.limit = "my_server"
+        job_with_links.extra_credentials.add(net_credential)
+        with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate._get_unified_job_field_names',
+                          return_value=['inventory', 'credential', 'limit']):
+            second_job = job_with_links.copy_unified_job()
+
+        # Check that job data matches the original variables
+        assert second_job.credential == job_with_links.credential
+        assert second_job.inventory == job_with_links.inventory
+        assert second_job.limit == 'my_server'
+        assert net_credential in second_job.extra_credentials.all()

awx/main/tests/functional/test_rbac_job.py

@@ -8,10 +8,12 @@ from awx.main.access import (
 )
 from awx.main.models import (
     Job,
+    JobTemplate,
     AdHocCommand,
     InventoryUpdate,
     InventorySource,
-    ProjectUpdate
+    ProjectUpdate,
+    User
 )
 
 
@@ -124,6 +126,45 @@ def test_project_org_admin_delete_allowed(normal_job, org_admin):
     assert access.can_delete(normal_job)
 
 
+@pytest.mark.django_db
+class TestJobRelaunchAccess:
+
+    def test_job_relaunch_normal_resource_access(self, user, inventory, machine_credential):
+        job_with_links = Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory)
+        inventory_user = user('user1', False)
+        credential_user = user('user2', False)
+        both_user = user('user3', False)
+
+        # Confirm that a user with inventory & credential access can launch
+        job_with_links.credential.use_role.members.add(both_user)
+        job_with_links.inventory.use_role.members.add(both_user)
+        assert both_user.can_access(Job, 'start', job_with_links, validate_license=False)
+
+        # Confirm that a user with credential access alone cannot launch
+        job_with_links.credential.use_role.members.add(credential_user)
+        assert not credential_user.can_access(Job, 'start', job_with_links, validate_license=False)
+
+        # Confirm that a user with inventory access alone cannot launch
+        job_with_links.inventory.use_role.members.add(inventory_user)
+        assert not inventory_user.can_access(Job, 'start', job_with_links, validate_license=False)
+
+    def test_job_relaunch_extra_credential_access(
+            self, post, inventory, project, credential, net_credential):
+        jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project)
+        jt.extra_credentials.add(credential)
+        job = jt.create_unified_job()
+
+        # Job is unchanged from JT, user has ability to launch
+        jt_user = User.objects.create(username='jobtemplateuser')
+        jt.execute_role.members.add(jt_user)
+        assert jt_user in job.job_template.execute_role
+        assert jt_user.can_access(Job, 'start', job, validate_license=False)
+
+        # Job has prompted extra_credential, launch denied w/ message
+        job.extra_credentials.add(net_credential)
+        assert not jt_user.can_access(Job, 'start', job, validate_license=False)
+
+
 @pytest.mark.django_db
 class TestJobAndUpdateCancels: