diff --git a/awx/main/tests/functional/api/test_credential_type.py b/awx/main/tests/functional/api/test_credential_type.py index 45b5e79994..c8f87f0c57 100644 --- a/awx/main/tests/functional/api/test_credential_type.py +++ b/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, diff --git a/awx/main/tests/functional/api/test_job_runtime_params.py b/awx/main/tests/functional/api/test_job_runtime_params.py index d792ec656d..e628623cf1 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/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 diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py index f8c8094ac0..01a91091be 100644 --- a/awx/main/tests/unit/test_tasks.py +++ b/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(