diff options
author | Chris Meyers <chris.meyers.fsu@gmail.com> | 2024-12-19 16:06:01 +0100 |
---|---|---|
committer | Chris Meyers <chrismeyersfsu@users.noreply.github.com> | 2024-12-19 20:03:27 +0100 |
commit | cf9e6796ea3ad91ddd7e40474e878d7ded7e0c32 (patch) | |
tree | 324017053d332a983769d40e23cc4f76cc87d9e1 | |
parent | Remove inject_credential from awx (diff) | |
download | awx-cf9e6796ea3ad91ddd7e40474e878d7ded7e0c32.tar.xz awx-cf9e6796ea3ad91ddd7e40474e878d7ded7e0c32.zip |
Move cred type unite tests to awx-plugins
-rw-r--r-- | awx/main/tests/unit/test_tasks.py | 262 |
1 files changed, 0 insertions, 262 deletions
diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py index b960c80d34..9ac278b568 100644 --- a/awx/main/tests/unit/test_tasks.py +++ b/awx/main/tests/unit/test_tasks.py @@ -1,5 +1,4 @@ # -*- coding: utf-8 -*- -import configparser import json import os import shutil @@ -856,205 +855,6 @@ class TestJobCredentials(TestJobExecution): assert '--vault-id dev@prompt' in ' '.join(args) assert '--vault-id prod@prompt' in ' '.join(args) - @pytest.mark.parametrize("verify", (True, False)) - def test_k8s_credential(self, job, private_data_dir, verify, mock_me): - k8s = CredentialType.defaults['kubernetes_bearer_token']() - inputs = { - 'host': 'https://example.org/', - 'bearer_token': 'token123', - } - if verify: - inputs['verify_ssl'] = True - inputs['ssl_ca_cert'] = 'CERTDATA' - credential = Credential( - pk=1, - credential_type=k8s, - inputs=inputs, - ) - 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' - - if verify: - assert env['K8S_AUTH_VERIFY_SSL'] == 'True' - local_path = to_host_path(env['K8S_AUTH_SSL_CA_CERT'], private_data_dir) - with open(local_path, 'r') as f: - cert = f.read() - assert cert == 'CERTDATA' - else: - assert env['K8S_AUTH_VERIFY_SSL'] == 'False' - assert 'K8S_AUTH_SSL_CA_CERT' not in env - - assert safe_env['K8S_AUTH_API_KEY'] == HIDDEN_PASSWORD - - def test_aws_cloud_credential(self, job, private_data_dir, mock_me): - aws = CredentialType.defaults['aws']() - credential = Credential(pk=1, credential_type=aws, inputs={'username': 'bob', 'password': 'secret'}) - credential.inputs['password'] = encrypt_field(credential, 'password') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - assert env['AWS_ACCESS_KEY_ID'] == 'bob' - assert env['AWS_SECRET_ACCESS_KEY'] == 'secret' - assert 'AWS_SECURITY_TOKEN' not in env - assert safe_env['AWS_SECRET_ACCESS_KEY'] == HIDDEN_PASSWORD - - def test_aws_cloud_credential_with_sts_token(self, private_data_dir, job, mock_me): - aws = CredentialType.defaults['aws']() - credential = Credential(pk=1, credential_type=aws, inputs={'username': 'bob', 'password': 'secret', 'security_token': 'token'}) - for key in ('password', 'security_token'): - credential.inputs[key] = encrypt_field(credential, key) - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - assert env['AWS_ACCESS_KEY_ID'] == 'bob' - assert env['AWS_SECRET_ACCESS_KEY'] == 'secret' - assert env['AWS_SECURITY_TOKEN'] == 'token' - assert safe_env['AWS_SECRET_ACCESS_KEY'] == HIDDEN_PASSWORD - - @pytest.mark.parametrize("cred_env_var", ['GCE_CREDENTIALS_FILE_PATH', 'GOOGLE_APPLICATION_CREDENTIALS']) - def test_gce_credentials(self, cred_env_var, private_data_dir, job, mock_me): - gce = CredentialType.defaults['gce']() - credential = Credential(pk=1, credential_type=gce, inputs={'username': 'bob', 'project': 'some-project', 'ssh_key_data': self.EXAMPLE_PRIVATE_KEY}) - credential.inputs['ssh_key_data'] = encrypt_field(credential, 'ssh_key_data') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - runner_path = env[cred_env_var] - local_path = to_host_path(runner_path, private_data_dir) - with open(local_path, 'rb') as f_host: - json_data = json.load(f_host) - assert json_data['type'] == 'service_account' - assert json_data['private_key'] == self.EXAMPLE_PRIVATE_KEY - assert json_data['client_email'] == 'bob' - assert json_data['project_id'] == 'some-project' - - def test_azure_rm_with_tenant(self, private_data_dir, job, mock_me): - azure = CredentialType.defaults['azure_rm']() - credential = Credential( - pk=1, credential_type=azure, inputs={'client': 'some-client', 'secret': 'some-secret', 'tenant': 'some-tenant', 'subscription': 'some-subscription'} - ) - credential.inputs['secret'] = encrypt_field(credential, 'secret') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - assert env['AZURE_CLIENT_ID'] == 'some-client' - assert env['AZURE_SECRET'] == 'some-secret' - assert env['AZURE_TENANT'] == 'some-tenant' - assert env['AZURE_SUBSCRIPTION_ID'] == 'some-subscription' - assert safe_env['AZURE_SECRET'] == HIDDEN_PASSWORD - - def test_azure_rm_with_password(self, private_data_dir, job, mock_me): - azure = CredentialType.defaults['azure_rm']() - credential = Credential( - pk=1, credential_type=azure, inputs={'subscription': 'some-subscription', 'username': 'bob', 'password': 'secret', 'cloud_environment': 'foobar'} - ) - credential.inputs['password'] = encrypt_field(credential, 'password') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - assert env['AZURE_SUBSCRIPTION_ID'] == 'some-subscription' - assert env['AZURE_AD_USER'] == 'bob' - assert env['AZURE_PASSWORD'] == 'secret' - assert env['AZURE_CLOUD_ENVIRONMENT'] == 'foobar' - assert safe_env['AZURE_PASSWORD'] == HIDDEN_PASSWORD - - def test_vmware_credentials(self, private_data_dir, job, mock_me): - vmware = CredentialType.defaults['vmware']() - credential = Credential(pk=1, credential_type=vmware, inputs={'username': 'bob', 'password': 'secret', 'host': 'https://example.org'}) - credential.inputs['password'] = encrypt_field(credential, 'password') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - assert env['VMWARE_USER'] == 'bob' - assert env['VMWARE_PASSWORD'] == 'secret' - assert env['VMWARE_HOST'] == 'https://example.org' - assert safe_env['VMWARE_PASSWORD'] == HIDDEN_PASSWORD - - def test_openstack_credentials(self, private_data_dir, job, mock_me): - task = jobs.RunJob() - task.instance = job - openstack = CredentialType.defaults['openstack']() - credential = Credential( - pk=1, credential_type=openstack, inputs={'username': 'bob', 'password': 'secret', 'project': 'tenant-name', 'host': 'https://keystone.example.org'} - ) - credential.inputs['password'] = encrypt_field(credential, 'password') - job.credentials.add(credential) - - private_data_files, ssh_key_data = task.build_private_data_files(job, private_data_dir) - env = task.build_env(job, private_data_dir, private_data_files=private_data_files) - credential.credential_type.inject_credential(credential, env, {}, [], private_data_dir) - - config_loc = to_host_path(env['OS_CLIENT_CONFIG_FILE'], private_data_dir) - with open(config_loc, 'r') as f: - shade_config = f.read() - assert shade_config == '\n'.join( - [ - 'clouds:', - ' devstack:', - ' auth:', - ' auth_url: https://keystone.example.org', - ' password: secret', - ' project_name: tenant-name', - ' username: bob', - ' verify: true', - '', - ] - ) - - @pytest.mark.parametrize("ca_file", [None, '/path/to/some/file']) - def test_rhv_credentials(self, private_data_dir, job, ca_file, mock_me): - rhv = CredentialType.defaults['rhv']() - inputs = { - 'host': 'some-ovirt-host.example.org', - 'username': 'bob', - 'password': 'some-pass', - } - if ca_file: - inputs['ca_file'] = ca_file - credential = Credential(pk=1, credential_type=rhv, inputs=inputs) - credential.inputs['password'] = encrypt_field(credential, 'password') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - config = configparser.ConfigParser() - host_path = to_host_path(env['OVIRT_INI_PATH'], private_data_dir) - config.read(host_path) - assert config.get('ovirt', 'ovirt_url') == 'some-ovirt-host.example.org' - assert config.get('ovirt', 'ovirt_username') == 'bob' - assert config.get('ovirt', 'ovirt_password') == 'some-pass' - if ca_file: - assert config.get('ovirt', 'ovirt_ca_file') == ca_file - else: - with pytest.raises(configparser.NoOptionError): - config.get('ovirt', 'ovirt_ca_file') - @pytest.mark.parametrize( 'authorize, expected_authorize', [ @@ -1089,68 +889,6 @@ class TestJobCredentials(TestJobExecution): assert f.read() == self.EXAMPLE_PRIVATE_KEY assert safe_env['ANSIBLE_NET_PASSWORD'] == HIDDEN_PASSWORD - def test_terraform_cloud_credentials(self, job, private_data_dir, mock_me): - terraform = CredentialType.defaults['terraform']() - hcl_config = ''' - backend "s3" { - bucket = "s3_sample_bucket" - key = "/tf_state/" - region = "us-east-1" - } - ''' - credential = Credential(pk=1, credential_type=terraform, inputs={'configuration': hcl_config}) - credential.inputs['configuration'] = encrypt_field(credential, 'configuration') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - local_path = to_host_path(env['TF_BACKEND_CONFIG_FILE'], private_data_dir) - with open(local_path, 'r') as f: - config = f.read() - assert config == hcl_config - - def test_terraform_gcs_backend_credentials(self, job, private_data_dir, mock_me): - terraform = CredentialType.defaults['terraform']() - hcl_config = ''' - backend "gcs" { - bucket = "gce_storage" - } - ''' - gce_backend_credentials = ''' - { - "type": "service_account", - "project_id": "sample", - "private_key_id": "eeeeeeeeeeeeeeeeeeeeeeeeeee", - "private_key": "-----BEGIN PRIVATE KEY-----\naaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n-----END PRIVATE KEY-----\n", - "client_email": "sample@sample.iam.gserviceaccount.com", - "client_id": "0123456789", - "auth_uri": "https://accounts.google.com/o/oauth2/auth", - "token_uri": "https://oauth2.googleapis.com/token", - "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", - "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/cloud-content-robot%40sample.iam.gserviceaccount.com", - } - ''' - credential = Credential(pk=1, credential_type=terraform, inputs={'configuration': hcl_config, 'gce_credentials': gce_backend_credentials}) - credential.inputs['configuration'] = encrypt_field(credential, 'configuration') - credential.inputs['gce_credentials'] = encrypt_field(credential, 'gce_credentials') - job.credentials.add(credential) - - env = {} - safe_env = {} - credential.credential_type.inject_credential(credential, env, safe_env, [], private_data_dir) - - local_path = to_host_path(env['TF_BACKEND_CONFIG_FILE'], private_data_dir) - with open(local_path, 'r') as f: - config = f.read() - assert config == hcl_config - - credentials_path = to_host_path(env['GOOGLE_BACKEND_CREDENTIALS'], private_data_dir) - with open(credentials_path, 'r') as f: - credentials = f.read() - assert credentials == gce_backend_credentials - def test_multi_cloud(self, private_data_dir, mock_me): gce = CredentialType.defaults['gce']() gce_credential = Credential(pk=1, credential_type=gce, inputs={'username': 'bob', 'project': 'some-project', 'ssh_key_data': self.EXAMPLE_PRIVATE_KEY}) |