new tests, and stricter can_copy can_edit

awx/api/serializers.py

@@ -23,6 +23,7 @@ from django.db import models
 # from django.utils.translation import ugettext_lazy as _
 from django.utils.encoding import force_text
 from django.utils.text import capfirst
+from django.forms.models import model_to_dict
 
 # Django REST Framework
 from rest_framework.exceptions import ValidationError
@@ -1783,19 +1784,27 @@ class JobTemplateSerializer(UnifiedJobTemplateSerializer, JobOptionsSerializer):
         if obj.survey_spec is not None and ('name' in obj.survey_spec and 'description' in obj.survey_spec):
             d['survey'] = dict(title=obj.survey_spec['name'], description=obj.survey_spec['description'])
         request = self.context.get('request', None)
-        if request is not None and request.user is not None and obj.inventory is not None and obj.project is not None:
-            d['can_copy'] = request.user.can_access(JobTemplate, 'add',
-                                                    {'inventory': obj.inventory.pk,
-                                                     'project': obj.project.pk})
-            d['can_edit'] = request.user.can_access(JobTemplate, 'change', obj,
-                                                    {'inventory': obj.inventory.pk,
-                                                     'project': obj.project.pk})
-        elif request is not None and request.user is not None and request.user.is_superuser:
-            d['can_copy'] = True
-            d['can_edit'] = True
-        else:
+
+        # Conditions that would create a validation error if coppied
+        validation_pass = True
+        if obj.inventory is None and not obj.ask_inventory_on_launch:
+            validation_pass = False
+        if obj.credential is None and not obj.ask_credential_on_launch:
+            validation_pass = False
+        if obj.project is None and not obj.job_type != PERM_INVENTORY_SCAN:
+            validation_pass = False
+
+        if request is None or request.user is None:
             d['can_copy'] = False
             d['can_edit'] = False
+        elif request.user.is_superuser:
+            d['can_copy'] = validation_pass
+            d['can_edit'] = True
+        else:
+            jt_data = model_to_dict(obj)
+            d['can_copy'] = validation_pass and request.user.can_access(JobTemplate, 'add', jt_data)
+            d['can_edit'] = request.user.can_access(JobTemplate, 'change', obj, jt_data)
+
         d['recent_jobs'] = self._recent_jobs(obj)
         return d
 

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

@@ -0,0 +1,202 @@
+import pytest
+import mock
+
+# AWX
+from awx.api.serializers import JobTemplateSerializer
+from awx.main.access import JobTemplateAccess
+from awx.main.models.jobs import JobTemplate
+
+# Django
+from django.test.client import RequestFactory
+from django.forms.models import model_to_dict
+from django.core.urlresolvers import reverse
+
+
+@pytest.fixture
+def jt_copy_edit(project):
+    return JobTemplate.objects.create(
+        job_type='run',
+        project=project,
+        playbook='hello_world.yml',
+        ask_inventory_on_launch=True,
+        ask_credential_on_launch=True,
+        name='copy-edit-job-template'
+    )
+
+# Test protection against limited set of validation problems
+
+@pytest.mark.django_db
+def test_bad_data_copy_edit(admin_user, project):
+    "If a required resource (inventory here) was deleted, copying not allowed"
+
+    jt_res = JobTemplate.objects.create(
+        job_type='run',
+        project=project,
+        inventory=None,  ask_inventory_on_launch=False, # not allowed
+        credential=None, ask_credential_on_launch=True,
+        name='deploy-job-template'
+    )
+    serializer = JobTemplateSerializer(jt_res)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = admin_user
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_res)
+    assert not response['summary_fields']['can_copy']
+    assert response['summary_fields']['can_edit']
+
+# Tests for correspondence between view info and actual access
+
+@pytest.mark.django_db
+def test_admin_copy_edit(jt_copy_edit, admin_user):
+    "Absent a validation error, system admins can do everything"
+
+    # Serializer can_copy/can_edit fields
+    serializer = JobTemplateSerializer(jt_copy_edit)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = admin_user
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_copy_edit)
+    assert response['summary_fields']['can_copy']
+    assert response['summary_fields']['can_edit']
+
+    # Access
+    jt_access = JobTemplateAccess(admin_user)
+    jt_dict = model_to_dict(jt_copy_edit)
+    assert jt_access.can_add(jt_dict)
+    assert jt_access.can_change(jt_copy_edit, jt_dict)
+
+@pytest.mark.django_db
+def test_org_admin_copy_edit(jt_copy_edit, org_admin):
+    "Organization admins SHOULD be able to copy a JT firmly in their org"
+
+    # Serializer can_copy/can_edit fields
+    serializer = JobTemplateSerializer(jt_copy_edit)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = org_admin
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_copy_edit)
+    assert response['summary_fields']['can_copy']
+    assert response['summary_fields']['can_edit']
+
+    # Access
+    jt_access = JobTemplateAccess(org_admin)
+    jt_dict = model_to_dict(jt_copy_edit)
+    assert jt_access.can_add(jt_dict)
+    assert jt_access.can_change(jt_copy_edit, jt_dict)
+
+@pytest.mark.django_db
+@pytest.mark.skip(reason="Waiting on issue 1981")
+def test_org_admin_foreign_cred_no_copy_edit(jt_copy_edit, org_admin, machine_credential):
+    "Organization admins SHOULD NOT be able to copy JT without resource access"
+
+    # Attach credential to JT that org admin can not use
+    jt_copy_edit.credential = machine_credential
+    jt_copy_edit.save()
+
+    # Serializer can_copy/can_edit fields
+    serializer = JobTemplateSerializer(jt_copy_edit)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = org_admin
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_copy_edit)
+    assert not response['summary_fields']['can_copy']
+    assert response['summary_fields']['can_edit']
+
+    # Access
+    jt_access = JobTemplateAccess(org_admin)
+    jt_dict = model_to_dict(jt_copy_edit)
+    assert not jt_access.can_add(jt_dict)
+    assert jt_access.can_change(jt_copy_edit, jt_dict)
+
+@pytest.mark.django_db
+def test_jt_admin_copy_edit(jt_copy_edit, rando):
+    "JT admins wihout access to associated resources SHOULD NOT be able to copy"
+
+    # random user given JT admin access only
+    jt_copy_edit.admin_role.members.add(rando)
+    jt_copy_edit.save()
+
+    # Serializer can_copy/can_edit fields
+    serializer = JobTemplateSerializer(jt_copy_edit)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = rando
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_copy_edit)
+    assert not response['summary_fields']['can_copy']
+    assert not response['summary_fields']['can_edit']
+
+    # Access
+    jt_access = JobTemplateAccess(rando)
+    jt_dict = model_to_dict(jt_copy_edit)
+    print ' jt_dict: ' + str(jt_dict)
+    assert not jt_access.can_add(jt_dict)
+    assert not jt_access.can_change(jt_copy_edit, jt_dict)
+
+@pytest.mark.django_db
+def test_proj_jt_admin_copy_edit(jt_copy_edit, rando):
+    "JT admins with access to associated resources SHOULD be able to copy"
+
+    # random user given JT and project admin abilities
+    jt_copy_edit.admin_role.members.add(rando)
+    jt_copy_edit.save()
+    jt_copy_edit.project.admin_role.members.add(rando)
+    jt_copy_edit.project.save()
+
+    # Serializer can_copy/can_edit fields
+    serializer = JobTemplateSerializer(jt_copy_edit)
+    request = RequestFactory().get('/api/v1/job_templates/12/')
+    request.user = rando
+    serializer.context['request'] = request
+    response = serializer.to_representation(jt_copy_edit)
+    assert response['summary_fields']['can_copy']
+    assert response['summary_fields']['can_edit']
+
+    # Access
+    jt_access = JobTemplateAccess(rando)
+    jt_dict = model_to_dict(jt_copy_edit)
+    assert jt_access.can_add(jt_dict)
+    assert jt_access.can_change(jt_copy_edit, jt_dict)
+
+# Functional tests - create new JT with all returned fields, as the UI does
+
+@pytest.mark.django_db
+def test_org_admin_copy_edit_functional(jt_copy_edit, org_admin, get, post):
+    get_response = get(reverse('api:job_template_detail', args=[jt_copy_edit.pk]), user=org_admin)
+
+    post_data = get_response.data
+    post_data['name'] = '%s @ 12:19:47 pm' % post_data['name']
+
+    assert get_response.status_code == 200
+    assert get_response.data['summary_fields']['can_copy']
+
+    with mock.patch(
+            'awx.main.models.projects.ProjectOptions.playbooks',
+            new_callable=mock.PropertyMock(return_value=['hello_world.yml'])):
+        post_response = post(reverse('api:job_template_list', args=[]), user=org_admin, data=post_data)
+
+    print '\n post_response: ' + str(post_response.data)
+    assert post_response.status_code == 201
+    assert post_response.data['name'] == 'copy-edit-job-template @ 12:19:47 pm'
+
+@pytest.mark.django_db
+def test_jt_admin_copy_edit_functional(jt_copy_edit, rando, get, post):
+
+    # Grant random user JT admin access only
+    jt_copy_edit.admin_role.members.add(rando)
+    jt_copy_edit.save()
+
+    get_response = get(reverse('api:job_template_detail', args=[jt_copy_edit.pk]), user=rando)
+
+    assert get_response.status_code == 200
+    assert not get_response.data['summary_fields']['can_copy']
+
+    post_data = get_response.data
+    post_data['name'] = '%s @ 12:19:47 pm' % post_data['name']
+
+    with mock.patch(
+            'awx.main.models.projects.ProjectOptions.playbooks',
+            new_callable=mock.PropertyMock(return_value=['hello_world.yml'])):
+        post_response = post(reverse('api:job_template_list', args=[]), user=rando, data=post_data)
+
+    print '\n post_response: ' + str(post_response.data)
+    assert post_response.status_code == 403