add tests for k8s credential usage in Job Templates

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

@@ -220,7 +220,7 @@ def test_create_valid_kind(kind, get, post, admin):
 
 
 @pytest.mark.django_db
-@pytest.mark.parametrize('kind', ['ssh', 'vault', 'scm', 'insights'])
+@pytest.mark.parametrize('kind', ['ssh', 'vault', 'scm', 'insights', 'kubernetes'])
 def test_create_invalid_kind(kind, get, post, admin):
     response = post(reverse('api:credential_type_list'), {
         'kind': kind,

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

@@ -483,25 +483,26 @@ def test_job_launch_pass_with_prompted_vault_password(machine_credential, vault_
 
 
 @pytest.mark.django_db
-def test_job_launch_JT_with_credentials(machine_credential, credential, net_credential, deploy_jobtemplate):
+def test_job_launch_JT_with_credentials(machine_credential, credential, net_credential, kube_credential, deploy_jobtemplate):
     deploy_jobtemplate.ask_credential_on_launch = True
     deploy_jobtemplate.save()
 
-    kv = dict(credentials=[credential.pk, net_credential.pk, machine_credential.pk])
+    kv = dict(credentials=[credential.pk, net_credential.pk, machine_credential.pk, kube_credential.pk])
     serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
     validated = serializer.is_valid()
     assert validated, serializer.errors
 
-    kv['credentials'] = [credential, net_credential, machine_credential]  # convert to internal value
+    kv['credentials'] = [credential, net_credential, machine_credential, kube_credential]  # convert to internal value
     prompted_fields, ignored_fields, errors = deploy_jobtemplate._accept_or_ignore_job_kwargs(
         _exclude_errors=['required', 'prompts'], **kv)
     job_obj = deploy_jobtemplate.create_unified_job(**prompted_fields)
 
     creds = job_obj.credentials.all()
-    assert len(creds) == 3
+    assert len(creds) == 4
     assert credential in creds
     assert net_credential in creds
     assert machine_credential in creds
+    assert kube_credential in creds
 
 
 @pytest.mark.django_db

awx/main/tests/unit/test_tasks.py

@@ -1037,6 +1037,34 @@ class TestJobCredentials(TestJobExecution):
         assert '--vault-id dev@prompt' in ' '.join(args)
         assert '--vault-id prod@prompt' in ' '.join(args)
 
+    def test_k8s_credential(self, job, private_data_dir):
+        k8s = CredentialType.defaults['kubernetes_bearer_token']()
+        credential = Credential(
+            pk=1,
+            credential_type=k8s,
+            inputs = {
+                'host': 'https://example.org/',
+                'bearer_token': 'token123',
+                'verify_ssl': True,
+                'ssl_ca_cert': 'CERTDATA'
+            }
+        )
+        credential.inputs['bearer_token'] = encrypt_field(credential, 'bearer_token')
+        job.credentials.add(credential)
+
+        env = {}
+        safe_env = {}
+        credential.credential_type.inject_credential(
+            credential, env, safe_env, [], private_data_dir
+        )
+
+        assert env['K8S_AUTH_HOST'] == 'https://example.org/'
+        assert env['K8S_AUTH_API_KEY'] == 'token123'
+        assert env['K8S_AUTH_VERIFY_SSL'] == 'True'
+        cert = open(env['K8S_AUTH_SSL_CA_CERT'], 'r').read()
+        assert cert == 'CERTDATA'
+        assert safe_env['K8S_AUTH_API_KEY'] == tasks.HIDDEN_PASSWORD
+
     def test_aws_cloud_credential(self, job, private_data_dir):
         aws = CredentialType.defaults['aws']()
         credential = Credential(