Merge pull request #6168 from ryanpetrello/multicredential_job

Replace Job/JT cloud/network credentials with a single M2M relation.

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

@@ -0,0 +1,88 @@
+import pytest
+
+from awx.api.versioning import reverse
+
+
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_extra_credential_creation(get, post, organization_factory, job_template_factory, credentialtype_aws):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    job = jt.create_unified_job()
+
+    url = reverse('api:job_extra_credentials_list', kwargs={'version': 'v2', 'pk': job.pk})
+    response = post(url, {
+        'name': 'My Cred',
+        'credential_type': credentialtype_aws.pk,
+        'inputs': {
+            'username': 'bob',
+            'password': 'secret',
+        }
+    }, objs.superusers.admin)
+    assert response.status_code == 201
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 1
+
+
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_attach_extra_credential(get, post, organization_factory, job_template_factory, credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    job = jt.create_unified_job()
+
+    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') == 0
+
+    response = post(url, {
+        'associate': True,
+        'id': credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 204
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 1
+
+
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_detach_extra_credential(get, post, organization_factory, job_template_factory, credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    jt.extra_credentials.add(credential)
+    jt.save()
+    job = jt.create_unified_job()
+
+    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
+
+    response = post(url, {
+        'disassociate': True,
+        'id': credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 204
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 0
+
+
+@pytest.mark.django_db
+def test_attach_extra_credential_wrong_kind_xfail(get, post, organization_factory, job_template_factory, machine_credential):
+    """Extra credentials only allow net + cloud credentials"""
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    job = jt.create_unified_job()
+
+    url = reverse('api:job_extra_credentials_list', kwargs={'version': 'v2', 'pk': job.pk})
+    response = post(url, {
+        'associate': True,
+        'id': machine_credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 400

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

@@ -36,6 +36,133 @@ def test_create(post, project, machine_credential, inventory, alice, grant_proje
     }, alice, expect=expect)
 
 
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_extra_credential_creation(get, post, organization_factory, job_template_factory, credentialtype_aws):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+
+    url = reverse('api:job_template_extra_credentials_list', kwargs={'version': 'v2', 'pk': jt.pk})
+    response = post(url, {
+        'name': 'My Cred',
+        'credential_type': credentialtype_aws.pk,
+        'inputs': {
+            'username': 'bob',
+            'password': 'secret',
+        }
+    }, objs.superusers.admin)
+    assert response.status_code == 201
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 1
+
+
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_attach_extra_credential(get, post, organization_factory, job_template_factory, credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+
+    url = reverse('api:job_template_extra_credentials_list', kwargs={'version': 'v2', 'pk': jt.pk})
+    response = post(url, {
+        'associate': True,
+        'id': credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 204
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 1
+
+
+# TODO: test this with RBAC and lower-priveleged users
+@pytest.mark.django_db
+def test_detach_extra_credential(get, post, organization_factory, job_template_factory, credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    jt.extra_credentials.add(credential)
+    jt.save()
+
+    url = reverse('api:job_template_extra_credentials_list', kwargs={'version': 'v2', 'pk': jt.pk})
+    response = post(url, {
+        'disassociate': True,
+        'id': credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 204
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 0
+
+
+@pytest.mark.django_db
+def test_attach_extra_credential_wrong_kind_xfail(get, post, organization_factory, job_template_factory, machine_credential):
+    """Extra credentials only allow net + cloud credentials"""
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+
+    url = reverse('api:job_template_extra_credentials_list', kwargs={'version': 'v2', 'pk': jt.pk})
+    response = post(url, {
+        'associate': True,
+        'id': machine_credential.id,
+    }, objs.superusers.admin)
+    assert response.status_code == 400
+
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('count') == 0
+
+
+@pytest.mark.django_db
+def test_v1_extra_credentials_detail(get, organization_factory, job_template_factory, credential, net_credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    jt.extra_credentials.add(credential)
+    jt.extra_credentials.add(net_credential)
+    jt.save()
+
+    url = reverse('api:job_template_detail', kwargs={'version': 'v1', 'pk': jt.pk})
+    response = get(url, user=objs.superusers.admin)
+    assert response.data.get('cloud_credential') == credential.pk
+    assert response.data.get('network_credential') == net_credential.pk
+
+
+@pytest.mark.django_db
+def test_v1_set_extra_credentials(get, patch, organization_factory, job_template_factory, credential, net_credential):
+    objs = organization_factory("org", superusers=['admin'])
+    jt = job_template_factory("jt", organization=objs.organization,
+                              inventory='test_inv', project='test_proj').job_template
+    jt.save()
+
+    url = reverse('api:job_template_detail', kwargs={'version': 'v1', 'pk': jt.pk})
+    response = patch(url, {
+        'cloud_credential': credential.pk,
+        'network_credential': net_credential.pk
+    }, objs.superusers.admin)
+    assert response.status_code == 200
+
+    url = reverse('api:job_template_detail', kwargs={'version': 'v1', 'pk': jt.pk})
+    response = get(url, user=objs.superusers.admin)
+    assert response.status_code == 200
+    assert response.data.get('cloud_credential') == credential.pk
+    assert response.data.get('network_credential') == net_credential.pk
+
+    url = reverse('api:job_template_detail', kwargs={'version': 'v1', 'pk': jt.pk})
+    response = patch(url, {
+        'cloud_credential': None,
+        'network_credential': None,
+    }, objs.superusers.admin)
+    assert response.status_code == 200
+
+    url = reverse('api:job_template_detail', kwargs={'version': 'v1', 'pk': jt.pk})
+    response = get(url, user=objs.superusers.admin)
+    assert response.status_code == 200
+    assert response.data.get('cloud_credential') is None
+    assert response.data.get('network_credential') is None
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize(
     "grant_project, grant_credential, grant_inventory, expect", [

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

@@ -316,7 +316,6 @@ def test_prefetch_jt_copy_capability(job_template, project, inventory, machine_c
     qs = JobTemplate.objects.all()
     cache_list_capabilities(qs, [{'copy': [
         'project.use', 'inventory.use', 'credential.use',
-        'cloud_credential.use', 'network_credential.use'
     ]}], JobTemplate, rando)
     assert qs[0].capabilities_cache == {'copy': False}
 
@@ -326,7 +325,6 @@ def test_prefetch_jt_copy_capability(job_template, project, inventory, machine_c
 
     cache_list_capabilities(qs, [{'copy': [
         'project.use', 'inventory.use', 'credential.use',
-        'cloud_credential.use', 'network_credential.use'
     ]}], JobTemplate, rando)
     assert qs[0].capabilities_cache == {'copy': True}
 

awx/main/tests/functional/conftest.py

@@ -215,6 +215,12 @@ def credential(credentialtype_aws):
                                      inputs={'username': 'something', 'password': 'secret'})
 
 
+@pytest.fixture
+def net_credential(credentialtype_net):
+    return Credential.objects.create(credential_type=credentialtype_net, name='test-cred',
+                                     inputs={'username': 'something', 'password': 'secret'})
+
+
 @pytest.fixture
 def machine_credential(credentialtype_ssh):
     return Credential.objects.create(credential_type=credentialtype_ssh, name='machine-cred',

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

@@ -1,7 +1,15 @@
 import pytest
 
+import json
+
 # AWX models
-from awx.main.models import ActivityStream, Organization, JobTemplate
+from awx.main.models import (
+    ActivityStream,
+    Organization,
+    JobTemplate,
+    Credential,
+    CredentialType
+)
 
 
 
@@ -80,3 +88,46 @@ class TestRolesAssociationEntries:
         proj2.use_role.parents.add(proj1.admin_role)
         assert ActivityStream.objects.filter(role=proj1.admin_role, project=proj2).count() == 1
 
+
+
+@pytest.fixture
+def somecloud_type():
+    return CredentialType.objects.create(
+        kind='cloud',
+        name='SomeCloud',
+        managed_by_tower=False,
+        inputs={
+            'fields': [{
+                'id': 'api_token',
+                'label': 'API Token',
+                'type': 'string',
+                'secret': True
+            }]
+        },
+        injectors={
+            'env': {
+                'MY_CLOUD_API_TOKEN': '{{api_token.foo()}}'
+            }
+        }
+    )
+
+
+@pytest.mark.django_db
+class TestCredentialModels:
+    '''
+    Assure that core elements of activity stream feature are working
+    '''
+
+    def test_create_credential_type(self, somecloud_type):
+        assert ActivityStream.objects.filter(credential_type=somecloud_type).count() == 1
+        entry = ActivityStream.objects.filter(credential_type=somecloud_type)[0]
+        assert entry.operation == 'create'
+
+    def test_credential_hidden_information(self, somecloud_type):
+        cred = Credential.objects.create(
+            credential_type=somecloud_type,
+            inputs = {'api_token': 'ABC123'}
+        )
+        entry = ActivityStream.objects.filter(credential=cred)[0]
+        assert entry.operation == 'create'
+        assert json.loads(entry.changes)['inputs'] == 'hidden'

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

@@ -0,0 +1,60 @@
+import pytest
+
+from django.core.exceptions import ValidationError
+from awx.main.models import Credential
+
+
+@pytest.mark.django_db
+def test_clean_credential_with_ssh_type(credentialtype_ssh, job_template):
+    credential = Credential(
+        name='My Credential',
+        credential_type=credentialtype_ssh
+    )
+    credential.save()
+
+    job_template.credential = credential
+    job_template.full_clean()
+
+
+@pytest.mark.django_db
+def test_clean_credential_with_invalid_type_xfail(credentialtype_aws, job_template):
+    credential = Credential(
+        name='My Credential',
+        credential_type=credentialtype_aws
+    )
+    credential.save()
+
+    with pytest.raises(ValidationError):
+        job_template.credential = credential
+        job_template.full_clean()
+
+
+@pytest.mark.django_db
+def test_clean_credential_with_custom_types(credentialtype_aws, credentialtype_net, job_template):
+    aws = Credential(
+        name='AWS Credential',
+        credential_type=credentialtype_aws
+    )
+    aws.save()
+    net = Credential(
+        name='Net Credential',
+        credential_type=credentialtype_net
+    )
+    net.save()
+
+    job_template.extra_credentials.add(aws)
+    job_template.extra_credentials.add(net)
+    job_template.full_clean()
+
+
+@pytest.mark.django_db
+def test_clean_credential_with_custom_types_xfail(credentialtype_ssh, job_template):
+    ssh = Credential(
+        name='SSH Credential',
+        credential_type=credentialtype_ssh
+    )
+    ssh.save()
+
+    with pytest.raises(ValidationError):
+        job_template.extra_credentials.add(ssh)
+        job_template.full_clean()