19658985666

Committed 20 Nov 2025 03:23PM UTC coverage: 71.764% (+0.2%) from 71.523%

Build # 19658985666

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #140 from shvenkat-rh/add-tfec-credential-type

Add a managed credential type for HCP Terraform

Coverage Stats

37 of 37 branches covered (100.0%)

Branch coverage included in aggregate %.

56 of 67 new or added lines in 4 files covered. (83.58%)

2286 of 3200 relevant lines covered (71.44%)

2.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.67

/tests/managed_credential_plugins_test.py

"""Individual ManagedCredentialType plugin tests."""

import configparser
import json
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Generic, TypeVar

import pytest
from pytest_subtests import SubTests

from awx_plugins.interfaces._temporary_private_api import (  # noqa: WPS436
    EnvVarsType,
    ManagedCredentialType,
)
from awx_plugins.interfaces._temporary_private_container_api import (
    CONTAINER_ROOT,
)
from awx_plugins.interfaces._temporary_private_credential_api import (  # noqa: WPS436
    Credential,
    CredentialInputType,
)
from awx_plugins.interfaces._temporary_private_inject_api import (  # noqa: WPS436
    HIDDEN_PASSWORD,
    inject_credential,
)

import yaml

from awx_plugins.credentials._managed_types.terraform import hcp_terraform
from awx_plugins.credentials.plugins import (  # noqa: WPS235
    aws,
    azure_rm,
    gce,
    kubernetes_bearer_token as k8s,
    net,
    openstack,
    rhv,
    terraform,
    vmware,
)


EXAMPLE_PRIVATE_KEY = 'foo'
HCL_S3_FILE = """
backend "s3" {
    bucket = "s3_sample_bucket"
    key    = "/tf_state/"
    region = "us-east-1"
}
"""
HCL_GCS_FILE = """
backend "gcs" {
    bucket = "gce_storage"
}
"""


def get_gce_creds() -> dict[str, str]:
    """GCE plugin expected backend credentials.

    This is a function instead of a module variable to work around
    WPS407. MappingProxy did not work to fix WPS407 due to the type not
    being json serializable errors during pytest fixture gathering.

    :returns: sample gce backend credential data.
    """
    return {
        'type': 'service_account',
        'project_id': 'sample',
        'private_key_id': 'eeeeeeeeeeeeeeeeeeeeeeeeeee',
        'private_key': """-----BEGIN PRIVATE KEY-----
    aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
    -----END PRIVATE KEY-----
    """,
        'client_email': 'sample@sample.iam.gserviceaccount.com',
        'client_id': '6123456789',
        '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',  # noqa: B950, WPS323
    }


# pylint: disable=redefined-outer-name
def to_host_path(path: str, private_data_dir: str) -> str:
    """Convert container path to host path.

    Given a path inside of the EE container, this gives the absolute
    path on the host machine within the private_data_dir.

    :param path: container path
    :param private_data_dir: runtime directory
    :return: Absolute path of private_data_dir on the container host
    """
    return path.replace(CONTAINER_ROOT, private_data_dir, 1)


@pytest.fixture
def private_data_dir(tmp_path: Path) -> Path:
    """Simulate ansible-runner directory backed runtime parameters.

    :yield: runtime directory
    """
    private_data_directory = tmp_path / 'aws-private-data-dir'
    (private_data_directory / 'env').mkdir(parents=True)
    (private_data_directory / 'inventory').mkdir()
    return private_data_directory


ExpectedDataType = TypeVar('ExpectedDataType')
IniEntryDataType = dict[str, dict[str, object]]


@dataclass(frozen=True)
class BaseEnvFile(Generic[ExpectedDataType]):
    """Base class for file type."""

    env_key: str
    expected_data: ExpectedDataType

    def check(
        self: 'BaseEnvFile[ExpectedDataType]',
        env: EnvVarsType,
        private_data_dir: str,
    ) -> None:
        """Override to ensure file contains expected data.

        :param env: environment to get the filename from
        :param private_data_dir: directory to read the file from
        :returns: None
        """
        raise NotImplementedError('Define me.')


class JsonEnvFile(BaseEnvFile[dict[str, str] | MappingProxyType[str, str]]):
    """Handle json files generated by credentials."""

    def check(
        self: 'JsonEnvFile',
        env: EnvVarsType,
        private_data_dir: str,
    ) -> None:
        """Ensure the json file contains expected data.

        :param env: environment to get the filename from
        :param private_data_dir: directory to read the file from
        :returns: None
        """
        with open(
            to_host_path(str(env[self.env_key]), private_data_dir),
        ) as fh:
            assert self.expected_data.items() == json.load(fh).items()


class StringEnvFile(BaseEnvFile[str]):
    """Handle basic non-structured files generated by credentials."""

    def check(
        self: 'StringEnvFile',
        env: EnvVarsType,
        private_data_dir: str,
    ) -> None:
        """Ensure file contains expected data.

        :param env: environment to get the filename from
        :param private_data_dir: directory to read the file from
        :returns: None
        """
        with open(
            to_host_path(str(env[self.env_key]), private_data_dir),
        ) as fh:
            assert self.expected_data == fh.read()


class IniEnvFile(BaseEnvFile[dict[str, dict[str, str]]]):
    """Handle ini files generated by credentials."""

    def check(
        self: 'IniEnvFile',
        env: EnvVarsType,
        private_data_dir: str,
    ) -> None:
        """Ensure ini file contains expected data.

        :param env: environment to get the filename from
        :param private_data_dir: directory to read the file from
        :returns: None
        """
        config = configparser.ConfigParser()
        config.read(to_host_path(str(env[self.env_key]), private_data_dir))
        for ini_section, expected_kv in self.expected_data.items():
            for expected_k, expected_v in expected_kv.items():
                assert config.get(ini_section, expected_k) == expected_v


class YamlEnvFile(BaseEnvFile[dict[str, IniEntryDataType]]):
    """Handle yaml files generated by credentials."""

    def check(
        self: 'YamlEnvFile',
        env: EnvVarsType,
        private_data_dir: str,
    ) -> None:
        """Ensure yaml file contains expected data.

        :param env: environment to get the filename from
        :param private_data_dir: directory to read the file from
        :returns: None
        """
        with open(
            to_host_path(str(env[self.env_key]), private_data_dir),
        ) as fh:
            assert self.expected_data.items() == yaml.safe_load(fh).items()


@pytest.mark.parametrize(
    (
        'cred_type',
        'inputs',
        'expected_env',
        'expected_safe_env',
        'expected_data_file',
    ),
    (
        pytest.param(
            aws,
            {
                'username': 'naomi',
                'password': 'naomis-secret',
                'security_token': 'token',
            },
            {
                'AWS_ACCESS_KEY_ID': 'naomi',
                'AWS_SECRET_ACCESS_KEY': 'naomis-secret',
                'AWS_SECURITY_TOKEN': 'token',
            },
            {'AWS_SECRET_ACCESS_KEY': HIDDEN_PASSWORD},
            [],
            id='aws',
        ),
        pytest.param(
            k8s,
            {
                'host': 'https://jill.example.org/',
                'bearer_token': 'jills-token',
            },
            {
                'K8S_AUTH_HOST': 'https://jill.example.org/',
                'K8S_AUTH_API_KEY': 'jills-token',
                'K8S_AUTH_VERIFY_SSL': 'False',
            },
            {'K8S_AUTH_API_KEY': HIDDEN_PASSWORD},
            [],
            id='k8s-verify-ssl-false',
        ),
        pytest.param(
            k8s,
            {
                'host': 'https://wilber.example.org/',
                'bearer_token': 'wilbers-token',
                'verify_ssl': True,
                'ssl_ca_cert': 'CERTDATA',
            },
            {
                'K8S_AUTH_HOST': 'https://wilber.example.org/',
                'K8S_AUTH_API_KEY': 'wilbers-token',
                'K8S_AUTH_VERIFY_SSL': 'True',
            },
            {'K8S_AUTH_API_KEY': HIDDEN_PASSWORD},
            [StringEnvFile('K8S_AUTH_SSL_CA_CERT', 'CERTDATA')],
            id='k8s-verify-ssl-true',
        ),
        pytest.param(
            gce,
            {
                'username': 'tony',
                'project': 'tonys-project',
                'ssh_key_data': EXAMPLE_PRIVATE_KEY,
            },
            {},
            {},
            [
                JsonEnvFile(
                    'GCE_CREDENTIALS_FILE_PATH',
                    {
                        'type': 'service_account',
                        'private_key': EXAMPLE_PRIVATE_KEY,
                        'client_email': 'tony',
                        'project_id': 'tonys-project',
                        'token_uri': 'https://oauth2.googleapis.com/token',
                    },
                ),
            ],
            id='gce',
        ),
        pytest.param(
            azure_rm,
            {
                'client': 'susie-client',
                'secret': 'susies-secret',
                'tenant': 'susies-tenant',
                'subscription': 'susies-subscription',
            },
            {
                'AZURE_CLIENT_ID': 'susie-client',
                'AZURE_SECRET': 'susies-secret',
                'AZURE_TENANT': 'susies-tenant',
                'AZURE_SUBSCRIPTION_ID': 'susies-subscription',
            },
            {'AZURE_SECRET': HIDDEN_PASSWORD},
            [],
            id='azure-tenant',
        ),
        pytest.param(
            azure_rm,
            {
                'subscription': 'bobs-subscription',
                'username': 'bob',
                'password': 'bobs-secret',
                'cloud_environment': 'bobs-cloud',
            },
            {
                'AZURE_AD_USER': 'bob',
                'AZURE_PASSWORD': 'bobs-secret',
                'AZURE_SUBSCRIPTION_ID': 'bobs-subscription',
                'AZURE_CLOUD_ENVIRONMENT': 'bobs-cloud',
            },
            {'AZURE_PASSWORD': HIDDEN_PASSWORD},
            [],
            id='azure-password',
        ),
        pytest.param(
            vmware,
            {
                'username': 'sam',
                'password': 'sams-secret',
                'host': 'https://sam.example.org',
            },
            {
                'VMWARE_USER': 'sam',
                'VMWARE_PASSWORD': 'sams-secret',
                'VMWARE_HOST': 'https://sam.example.org',
                'VMWARE_VALIDATE_CERTS': 'True',
            },
            {'VMWARE_PASSWORD': HIDDEN_PASSWORD},
            [],
            id='vmware',
        ),
        pytest.param(
            openstack,
            {
                'username': 'newman',
                'password': 'newmans-secret',
                'project': 'newmans-project',
                'host': 'https://newman.example.org',
            },
            {},
            {},
            [
                YamlEnvFile(
                    'OS_CLIENT_CONFIG_FILE',
                    {
                        'clouds': {
                            'devstack': {
                                'auth': {
                                    'auth_url': 'https://newman.example.org',
                                    'password': 'newmans-secret',
                                    'project_name': 'newmans-project',
                                    'username': 'newman',
                                },
                                'verify': True,
                            },
                        },
                    },
                ),
            ],
            id='openstack',
        ),
        pytest.param(
            rhv,
            {
                'host': 'rick.example.org',
                'username': 'rick',
                'password': 'ricks-pass',
            },
            {},
            {},
            [
                IniEnvFile(
                    'OVIRT_INI_PATH',
                    {
                        'ovirt': {
                            'ovirt_url': 'rick.example.org',
                            'ovirt_username': 'rick',
                            'ovirt_password': 'ricks-pass',
                        },
                    },
                ),
            ],
            id='rhv',
        ),
        pytest.param(
            rhv,
            {
                'host': 'rick.example.org',
                'username': 'rick',
                'password': 'ricks-pass',
                'ca_file': '/path/to/ricks/ca-file',
            },
            {},
            {},
            [
                IniEnvFile(
                    'OVIRT_INI_PATH',
                    {
                        'ovirt': {
                            'ovirt_url': 'rick.example.org',
                            'ovirt_username': 'rick',
                            'ovirt_password': 'ricks-pass',
                            'ovirt_ca_file': '/path/to/ricks/ca-file',
                        },
                    },
                ),
            ],
            id='rhv-ca-file',
        ),
        pytest.param(
            terraform,
            {'configuration': HCL_S3_FILE},
            {},
            {},
            [StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_S3_FILE)],
            id='terraform',
        ),
        pytest.param(
            terraform,
            {'configuration': HCL_S3_FILE},
            {},
            {},
            [StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_S3_FILE)],
            id='terraform-s3',
        ),
        pytest.param(
            terraform,
            {
                'configuration': HCL_GCS_FILE,
                'gce_credentials': json.dumps(get_gce_creds()),
            },
            {},
            {},
            [
                StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_GCS_FILE),
                JsonEnvFile('GOOGLE_BACKEND_CREDENTIALS', get_gce_creds()),
            ],
            id='terraform-gcs',
        ),
        pytest.param(
            hcp_terraform,
            {
                'token': 'hcp-token-123',
            },
            {
                'TF_TOKEN': 'hcp-token-123',
                'TF_HOSTNAME': 'app.terraform.io',
            },
            {'TF_TOKEN': HIDDEN_PASSWORD},
            [],
            id='hcp-terraform-default-hostname',
        ),
        pytest.param(
            hcp_terraform,
            {
                'hostname': 'terraform.example.com',
                'token': 'hcp-custom-token-456',
            },
            {
                'TF_TOKEN': 'hcp-custom-token-456',
                'TF_HOSTNAME': 'terraform.example.com',
            },
            {'TF_TOKEN': HIDDEN_PASSWORD},
            [],
            id='hcp-terraform-custom-hostname',
        ),
        pytest.param(
            hcp_terraform,
            {
                'hostname': 'app.terraform.io',
                'token': 'hcp-explicit-default-789',
            },
            {
                'TF_TOKEN': 'hcp-explicit-default-789',
                'TF_HOSTNAME': 'app.terraform.io',
            },
            {'TF_TOKEN': HIDDEN_PASSWORD},
            [],
            id='hcp-terraform-explicit-default-hostname',
        ),
        pytest.param(
            net,
            {
                'username': 'froto',
                'password': 'froto-pass',
                'ssh_key_data': EXAMPLE_PRIVATE_KEY,
                'authorize_password': 'frotos-authorize-pass',
                'authorize': True,
            },
            {
                'ANSIBLE_NET_USERNAME': 'froto',
                'ANSIBLE_NET_PASSWORD': 'froto-pass',
                'ANSIBLE_NET_AUTHORIZE': '1',
                'ANSIBLE_NET_AUTH_PASS': 'frotos-authorize-pass',
            },
            {'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},
            [StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],
            id='net-authorize',
            marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),
        ),
        pytest.param(
            net,
            {
                'username': 'shire',
                'password': 'shires-pass',
                'ssh_key_data': EXAMPLE_PRIVATE_KEY,
                'authorize_password': 'shires-authorize-pass',
                'authorize': False,
            },
            {
                'ANSIBLE_NET_USERNAME': 'froto',
                'ANSIBLE_NET_PASSWORD': 'frotos-pass',
                'ANSIBLE_NET_AUTHORIZE': '0',
            },
            {'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},
            [StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],
            id='net-authorize-false',
            marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),
        ),
        pytest.param(
            net,
            {
                'username': 'bilbo',
                'password': 'bilbos-pass',
                'ssh_key_data': EXAMPLE_PRIVATE_KEY,
                'authorize_password': 'bilbos-authorize-pass',
                'authorize': None,
            },
            {
                'ANSIBLE_NET_USERNAME': 'bilbo',
                'ANSIBLE_NET_PASSWORD': 'bilbos-pass',
                'ANSIBLE_NET_AUTHORIZE': '0',
            },
            {'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},
            [StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],
            id='net-authorize-none',
            marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),
        ),
    ),
)
# FIXME: Type ignore below due to ManagedCredentialType type not found.
# pylint: disable-next=too-many-arguments, too-many-positional-arguments
def test_credential_plugins(  # type: ignore[no-any-unimported]  # noqa: WPS211
    cred_type: ManagedCredentialType,
    inputs: CredentialInputType,
    expected_env: dict[str, str],
    expected_safe_env: dict[str, str],
    expected_data_file: list[BaseEnvFile[ExpectedDataType]],
    private_data_dir: Path,
    subtests: SubTests,
) -> None:
    """Check that the aws sts token credentials are injected."""
    env: EnvVarsType = {}
    safe_env: EnvVarsType = {}
    inject_credential(
        cred_type,
        Credential(inputs=inputs),
        env,
        safe_env,
        [],
        str(private_data_dir),
    )
    assert expected_env.items() <= env.items()
    assert expected_safe_env.items() <= safe_env.items()

    for env_file_type in expected_data_file:
        with subtests.test(env_file_type.__class__.__name__):
            env_file_type.check(env, str(private_data_dir))

1	"""Individual ManagedCredentialType plugin tests."""
2
3	import configparser	3✔
4	import json	3✔
5	from dataclasses import dataclass	3✔
6	from pathlib import Path	3✔
7	from types import MappingProxyType	3✔
8	from typing import Generic, TypeVar	3✔
9
10	import pytest	3✔
11	from pytest_subtests import SubTests	3✔
12
13	from awx_plugins.interfaces._temporary_private_api import ( # noqa: WPS436	3✔
14	EnvVarsType,
15	ManagedCredentialType,
16	)
17	from awx_plugins.interfaces._temporary_private_container_api import (	3✔
18	CONTAINER_ROOT,
19	)
20	from awx_plugins.interfaces._temporary_private_credential_api import ( # noqa: WPS436	3✔
21	Credential,
22	CredentialInputType,
23	)
24	from awx_plugins.interfaces._temporary_private_inject_api import ( # noqa: WPS436	3✔
25	HIDDEN_PASSWORD,
26	inject_credential,
27	)
28
29	import yaml	3✔
30
31	from awx_plugins.credentials._managed_types.terraform import hcp_terraform	3✔
32	from awx_plugins.credentials.plugins import ( # noqa: WPS235	3✔
33	aws,
34	azure_rm,
35	gce,
36	kubernetes_bearer_token as k8s,
37	net,
38	openstack,
39	rhv,
40	terraform,
41	vmware,
42	)
43
44
45	EXAMPLE_PRIVATE_KEY = 'foo'	3✔
46	HCL_S3_FILE = """	3✔
47	backend "s3" {
48	bucket = "s3_sample_bucket"
49	key = "/tf_state/"
50	region = "us-east-1"
51	}
52	"""
53	HCL_GCS_FILE = """	3✔
54	backend "gcs" {
55	bucket = "gce_storage"
56	}
57	"""
58
59
60	def get_gce_creds() -> dict[str, str]:	3✔
61	"""GCE plugin expected backend credentials.
62
63	This is a function instead of a module variable to work around
64	WPS407. MappingProxy did not work to fix WPS407 due to the type not
65	being json serializable errors during pytest fixture gathering.
66
67	:returns: sample gce backend credential data.
68	"""
69	return {
70	'type': 'service_account',	3✔
71	'project_id': 'sample',	3✔
72	'private_key_id': 'eeeeeeeeeeeeeeeeeeeeeeeeeee',	3✔
73	'private_key': """-----BEGIN PRIVATE KEY-----	3✔
74	aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
75	-----END PRIVATE KEY-----
76	""",
77	'client_email': 'sample@sample.iam.gserviceaccount.com',	3✔
78	'client_id': '6123456789',	3✔
79	'auth_uri': 'https://accounts.google.com/o/oauth2/auth',	3✔
80	'token_uri': 'https://oauth2.googleapis.com/token',	3✔
81	'auth_provider_x509_cert_url': 'https://www.googleapis.com/oauth2/v1/certs',	3✔
82	'client_x509_cert_url': 'https://www.googleapis.com/robot/v1/metadata/x509/cloud-content-robot%40sample.iam.gserviceaccount.com', # noqa: B950, WPS323	3✔
83	}
84
85
86	# pylint: disable=redefined-outer-name
87	def to_host_path(path: str, private_data_dir: str) -> str:	3✔
88	"""Convert container path to host path.
89
90	Given a path inside of the EE container, this gives the absolute
91	path on the host machine within the private_data_dir.
92
93	:param path: container path
94	:param private_data_dir: runtime directory
95	:return: Absolute path of private_data_dir on the container host
96	"""
97	return path.replace(CONTAINER_ROOT, private_data_dir, 1)	3✔
98
99
100	@pytest.fixture	3✔
101	def private_data_dir(tmp_path: Path) -> Path:	3✔
102	"""Simulate ansible-runner directory backed runtime parameters.
103
104	:yield: runtime directory
105	"""
106	private_data_directory = tmp_path / 'aws-private-data-dir'	3✔
107	(private_data_directory / 'env').mkdir(parents=True)	3✔
108	(private_data_directory / 'inventory').mkdir()	3✔
109	return private_data_directory	3✔
110
111
112	ExpectedDataType = TypeVar('ExpectedDataType')
113	IniEntryDataType = dict[str, dict[str, object]]	3✔
114
115
116	@dataclass(frozen=True)	3✔
117	class BaseEnvFile(Generic[ExpectedDataType]):	3✔
118	"""Base class for file type."""
119
120	env_key: str	3✔
121	expected_data: ExpectedDataType	3✔
122
123	def check(	3✔
124	self: 'BaseEnvFile[ExpectedDataType]',
125	env: EnvVarsType,
126	private_data_dir: str,
127	) -> None:
128	"""Override to ensure file contains expected data.
129
130	:param env: environment to get the filename from
131	:param private_data_dir: directory to read the file from
132	:returns: None
133	"""
134	raise NotImplementedError('Define me.')	3✔
135
136
137	class JsonEnvFile(BaseEnvFile[dict[str, str] \| MappingProxyType[str, str]]):	3✔
138	"""Handle json files generated by credentials."""
139
140	def check(	3✔
141	self: 'JsonEnvFile',
142	env: EnvVarsType,
143	private_data_dir: str,
144	) -> None:
145	"""Ensure the json file contains expected data.
146
147	:param env: environment to get the filename from
148	:param private_data_dir: directory to read the file from
149	:returns: None
150	"""
151	with open(	3✔
152	to_host_path(str(env[self.env_key]), private_data_dir),	3✔
153	) as fh:	3✔
154	assert self.expected_data.items() == json.load(fh).items()	×
155
156
157	class StringEnvFile(BaseEnvFile[str]):	3✔
158	"""Handle basic non-structured files generated by credentials."""
159
160	def check(	3✔
161	self: 'StringEnvFile',
162	env: EnvVarsType,
163	private_data_dir: str,
164	) -> None:
165	"""Ensure file contains expected data.
166
167	:param env: environment to get the filename from
168	:param private_data_dir: directory to read the file from
169	:returns: None
170	"""
171	with open(	3✔
172	to_host_path(str(env[self.env_key]), private_data_dir),	3✔
173	) as fh:	3✔
174	assert self.expected_data == fh.read()	3✔
175
176
177	class IniEnvFile(BaseEnvFile[dict[str, dict[str, str]]]):	3✔
178	"""Handle ini files generated by credentials."""
179
180	def check(	3✔
181	self: 'IniEnvFile',
182	env: EnvVarsType,
183	private_data_dir: str,
184	) -> None:
185	"""Ensure ini file contains expected data.
186
187	:param env: environment to get the filename from
188	:param private_data_dir: directory to read the file from
189	:returns: None
190	"""
191	config = configparser.ConfigParser()	3✔
192	config.read(to_host_path(str(env[self.env_key]), private_data_dir))	3✔
193	for ini_section, expected_kv in self.expected_data.items():	3✔
194	for expected_k, expected_v in expected_kv.items():	3✔
195	assert config.get(ini_section, expected_k) == expected_v	3✔
196
197
198	class YamlEnvFile(BaseEnvFile[dict[str, IniEntryDataType]]):	3✔
199	"""Handle yaml files generated by credentials."""
200
201	def check(	3✔
202	self: 'YamlEnvFile',
203	env: EnvVarsType,
204	private_data_dir: str,
205	) -> None:
206	"""Ensure yaml file contains expected data.
207
208	:param env: environment to get the filename from
209	:param private_data_dir: directory to read the file from
210	:returns: None
211	"""
212	with open(	3✔
213	to_host_path(str(env[self.env_key]), private_data_dir),	3✔
214	) as fh:	3✔
215	assert self.expected_data.items() == yaml.safe_load(fh).items()	×
216
217
218	@pytest.mark.parametrize(	3✔
219	(
220	'cred_type',	3✔
221	'inputs',	3✔
222	'expected_env',	3✔
223	'expected_safe_env',	3✔
224	'expected_data_file',	3✔
225	),
226	(
227	pytest.param(	3✔
228	aws,	×
229	{
230	'username': 'naomi',	3✔
231	'password': 'naomis-secret',	3✔
232	'security_token': 'token',	3✔
233	},
234	{
235	'AWS_ACCESS_KEY_ID': 'naomi',	3✔
236	'AWS_SECRET_ACCESS_KEY': 'naomis-secret',	3✔
237	'AWS_SECURITY_TOKEN': 'token',	3✔
238	},
239	{'AWS_SECRET_ACCESS_KEY': HIDDEN_PASSWORD},	3✔
240	[],
241	id='aws',	3✔
242	),
243	pytest.param(	3✔
244	k8s,	×
245	{
246	'host': 'https://jill.example.org/',	3✔
247	'bearer_token': 'jills-token',	3✔
248	},
249	{
250	'K8S_AUTH_HOST': 'https://jill.example.org/',	3✔
251	'K8S_AUTH_API_KEY': 'jills-token',	3✔
252	'K8S_AUTH_VERIFY_SSL': 'False',	3✔
253	},
254	{'K8S_AUTH_API_KEY': HIDDEN_PASSWORD},	3✔
255	[],
256	id='k8s-verify-ssl-false',	3✔
257	),
258	pytest.param(	3✔
259	k8s,	×
260	{
261	'host': 'https://wilber.example.org/',	3✔
262	'bearer_token': 'wilbers-token',	3✔
263	'verify_ssl': True,	3✔
264	'ssl_ca_cert': 'CERTDATA',	3✔
265	},
266	{
267	'K8S_AUTH_HOST': 'https://wilber.example.org/',	3✔
268	'K8S_AUTH_API_KEY': 'wilbers-token',	3✔
269	'K8S_AUTH_VERIFY_SSL': 'True',	3✔
270	},
271	{'K8S_AUTH_API_KEY': HIDDEN_PASSWORD},	3✔
272	[StringEnvFile('K8S_AUTH_SSL_CA_CERT', 'CERTDATA')],	3✔
273	id='k8s-verify-ssl-true',	3✔
274	),
275	pytest.param(	3✔
276	gce,	×
277	{
278	'username': 'tony',	3✔
279	'project': 'tonys-project',	3✔
280	'ssh_key_data': EXAMPLE_PRIVATE_KEY,	3✔
281	},
282	{},
283	{},
284	[
285	JsonEnvFile(	3✔
286	'GCE_CREDENTIALS_FILE_PATH',	3✔
287	{
288	'type': 'service_account',	3✔
289	'private_key': EXAMPLE_PRIVATE_KEY,	3✔
290	'client_email': 'tony',	3✔
291	'project_id': 'tonys-project',	3✔
292	'token_uri': 'https://oauth2.googleapis.com/token',	3✔
293	},
294	),
295	],
296	id='gce',	3✔
297	),
298	pytest.param(	3✔
299	azure_rm,	×
300	{
301	'client': 'susie-client',	3✔
302	'secret': 'susies-secret',	3✔
303	'tenant': 'susies-tenant',	3✔
304	'subscription': 'susies-subscription',	3✔
305	},
306	{
307	'AZURE_CLIENT_ID': 'susie-client',	3✔
308	'AZURE_SECRET': 'susies-secret',	3✔
309	'AZURE_TENANT': 'susies-tenant',	3✔
310	'AZURE_SUBSCRIPTION_ID': 'susies-subscription',	3✔
311	},
312	{'AZURE_SECRET': HIDDEN_PASSWORD},	3✔
313	[],
314	id='azure-tenant',	3✔
315	),
316	pytest.param(	3✔
317	azure_rm,	×
318	{
319	'subscription': 'bobs-subscription',	3✔
320	'username': 'bob',	3✔
321	'password': 'bobs-secret',	3✔
322	'cloud_environment': 'bobs-cloud',	3✔
323	},
324	{
325	'AZURE_AD_USER': 'bob',	3✔
326	'AZURE_PASSWORD': 'bobs-secret',	3✔
327	'AZURE_SUBSCRIPTION_ID': 'bobs-subscription',	3✔
328	'AZURE_CLOUD_ENVIRONMENT': 'bobs-cloud',	3✔
329	},
330	{'AZURE_PASSWORD': HIDDEN_PASSWORD},	3✔
331	[],
332	id='azure-password',	3✔
333	),
334	pytest.param(	3✔
335	vmware,	×
336	{
337	'username': 'sam',	3✔
338	'password': 'sams-secret',	3✔
339	'host': 'https://sam.example.org',	3✔
340	},
341	{
342	'VMWARE_USER': 'sam',	3✔
343	'VMWARE_PASSWORD': 'sams-secret',	3✔
344	'VMWARE_HOST': 'https://sam.example.org',	3✔
345	'VMWARE_VALIDATE_CERTS': 'True',	3✔
346	},
347	{'VMWARE_PASSWORD': HIDDEN_PASSWORD},	3✔
348	[],
349	id='vmware',	3✔
350	),
351	pytest.param(	3✔
352	openstack,	×
353	{
354	'username': 'newman',	3✔
355	'password': 'newmans-secret',	3✔
356	'project': 'newmans-project',	3✔
357	'host': 'https://newman.example.org',	3✔
358	},
359	{},
360	{},
361	[
362	YamlEnvFile(	3✔
363	'OS_CLIENT_CONFIG_FILE',	3✔
364	{
365	'clouds': {	3✔
366	'devstack': {	3✔
367	'auth': {	3✔
368	'auth_url': 'https://newman.example.org',	3✔
369	'password': 'newmans-secret',	3✔
370	'project_name': 'newmans-project',	3✔
371	'username': 'newman',	3✔
372	},
373	'verify': True,	3✔
374	},
375	},
376	},
377	),
378	],
379	id='openstack',	3✔
380	),
381	pytest.param(	3✔
382	rhv,	×
383	{
384	'host': 'rick.example.org',	3✔
385	'username': 'rick',	3✔
386	'password': 'ricks-pass',	3✔
387	},
388	{},
389	{},
390	[
391	IniEnvFile(	3✔
392	'OVIRT_INI_PATH',	3✔
393	{
394	'ovirt': {	3✔
395	'ovirt_url': 'rick.example.org',	3✔
396	'ovirt_username': 'rick',	3✔
397	'ovirt_password': 'ricks-pass',	3✔
398	},
399	},
400	),
401	],
402	id='rhv',	3✔
403	),
404	pytest.param(	3✔
405	rhv,	×
406	{
407	'host': 'rick.example.org',	3✔
408	'username': 'rick',	3✔
409	'password': 'ricks-pass',	3✔
410	'ca_file': '/path/to/ricks/ca-file',	3✔
411	},
412	{},
413	{},
414	[
415	IniEnvFile(	3✔
416	'OVIRT_INI_PATH',	3✔
417	{
418	'ovirt': {	3✔
419	'ovirt_url': 'rick.example.org',	3✔
420	'ovirt_username': 'rick',	3✔
421	'ovirt_password': 'ricks-pass',	3✔
422	'ovirt_ca_file': '/path/to/ricks/ca-file',	3✔
423	},
424	},
425	),
426	],
427	id='rhv-ca-file',	3✔
428	),
429	pytest.param(	3✔
430	terraform,	×
431	{'configuration': HCL_S3_FILE},	3✔
432	{},
433	{},
434	[StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_S3_FILE)],	3✔
435	id='terraform',	3✔
436	),
437	pytest.param(	3✔
438	terraform,	×
439	{'configuration': HCL_S3_FILE},	3✔
440	{},
441	{},
442	[StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_S3_FILE)],	3✔
443	id='terraform-s3',	3✔
444	),
445	pytest.param(	3✔
446	terraform,	×
447	{
448	'configuration': HCL_GCS_FILE,	3✔
449	'gce_credentials': json.dumps(get_gce_creds()),	×
450	},
451	{},
452	{},
453	[
454	StringEnvFile('TF_BACKEND_CONFIG_FILE', HCL_GCS_FILE),	3✔
455	JsonEnvFile('GOOGLE_BACKEND_CREDENTIALS', get_gce_creds()),	3✔
456	],
457	id='terraform-gcs',	3✔
458	),
459	pytest.param(	3✔
NEW 460	hcp_terraform,	×
461	{
462	'token': 'hcp-token-123',	3✔
463	},
464	{
465	'TF_TOKEN': 'hcp-token-123',	3✔
466	'TF_HOSTNAME': 'app.terraform.io',	3✔
467	},
468	{'TF_TOKEN': HIDDEN_PASSWORD},	3✔
469	[],
470	id='hcp-terraform-default-hostname',	3✔
471	),
472	pytest.param(	3✔
NEW 473	hcp_terraform,	×
474	{
475	'hostname': 'terraform.example.com',	3✔
476	'token': 'hcp-custom-token-456',	3✔
477	},
478	{
479	'TF_TOKEN': 'hcp-custom-token-456',	3✔
480	'TF_HOSTNAME': 'terraform.example.com',	3✔
481	},
482	{'TF_TOKEN': HIDDEN_PASSWORD},	3✔
483	[],
484	id='hcp-terraform-custom-hostname',	3✔
485	),
486	pytest.param(	3✔
NEW 487	hcp_terraform,	×
488	{
489	'hostname': 'app.terraform.io',	3✔
490	'token': 'hcp-explicit-default-789',	3✔
491	},
492	{
493	'TF_TOKEN': 'hcp-explicit-default-789',	3✔
494	'TF_HOSTNAME': 'app.terraform.io',	3✔
495	},
496	{'TF_TOKEN': HIDDEN_PASSWORD},	3✔
497	[],
498	id='hcp-terraform-explicit-default-hostname',	3✔
499	),
500	pytest.param(	3✔
501	net,	×
502	{
503	'username': 'froto',	3✔
504	'password': 'froto-pass',	3✔
505	'ssh_key_data': EXAMPLE_PRIVATE_KEY,	3✔
506	'authorize_password': 'frotos-authorize-pass',	3✔
507	'authorize': True,	3✔
508	},
509	{
510	'ANSIBLE_NET_USERNAME': 'froto',	3✔
511	'ANSIBLE_NET_PASSWORD': 'froto-pass',	3✔
512	'ANSIBLE_NET_AUTHORIZE': '1',	3✔
513	'ANSIBLE_NET_AUTH_PASS': 'frotos-authorize-pass',	3✔
514	},
515	{'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},	3✔
516	[StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],	3✔
517	id='net-authorize',	3✔
518	marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),	3✔
519	),
520	pytest.param(	3✔
521	net,	×
522	{
523	'username': 'shire',	3✔
524	'password': 'shires-pass',	3✔
525	'ssh_key_data': EXAMPLE_PRIVATE_KEY,	3✔
526	'authorize_password': 'shires-authorize-pass',	3✔
527	'authorize': False,	3✔
528	},
529	{
530	'ANSIBLE_NET_USERNAME': 'froto',	3✔
531	'ANSIBLE_NET_PASSWORD': 'frotos-pass',	3✔
532	'ANSIBLE_NET_AUTHORIZE': '0',	3✔
533	},
534	{'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},	3✔
535	[StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],	3✔
536	id='net-authorize-false',	3✔
537	marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),	3✔
538	),
539	pytest.param(	3✔
540	net,	×
541	{
542	'username': 'bilbo',	3✔
543	'password': 'bilbos-pass',	3✔
544	'ssh_key_data': EXAMPLE_PRIVATE_KEY,	3✔
545	'authorize_password': 'bilbos-authorize-pass',	3✔
546	'authorize': None,	3✔
547	},
548	{
549	'ANSIBLE_NET_USERNAME': 'bilbo',	3✔
550	'ANSIBLE_NET_PASSWORD': 'bilbos-pass',	3✔
551	'ANSIBLE_NET_AUTHORIZE': '0',	3✔
552	},
553	{'ANSIBLE_NET_PASSWORD': HIDDEN_PASSWORD},	3✔
554	[StringEnvFile('ANSIBLE_NET_SSH_KEYFILE', EXAMPLE_PRIVATE_KEY)],	3✔
555	id='net-authorize-none',	3✔
556	marks=pytest.mark.xfail(reason='net plugin lives in AWX.'),	3✔
557	),
558	),
559	)
560	# FIXME: Type ignore below due to ManagedCredentialType type not found.
561	# pylint: disable-next=too-many-arguments, too-many-positional-arguments
562	def test_credential_plugins( # type: ignore[no-any-unimported] # noqa: WPS211	×
563	cred_type: ManagedCredentialType,
564	inputs: CredentialInputType,
565	expected_env: dict[str, str],
566	expected_safe_env: dict[str, str],
567	expected_data_file: list[BaseEnvFile[ExpectedDataType]],
568	private_data_dir: Path,
569	subtests: SubTests,
570	) -> None:
571	"""Check that the aws sts token credentials are injected."""
572	env: EnvVarsType = {}	3✔
573	safe_env: EnvVarsType = {}	3✔
574	inject_credential(	×
575	cred_type,	×
576	Credential(inputs=inputs),	3✔
577	env,	3✔
578	safe_env,	3✔
579	[],
580	str(private_data_dir),	3✔
581	)
582	assert expected_env.items() <= env.items()	3✔
583	assert expected_safe_env.items() <= safe_env.items()	3✔
584
585	for env_file_type in expected_data_file:	3✔
586	with subtests.test(env_file_type.__class__.__name__):	3✔
587	env_file_type.check(env, str(private_data_dir))	3✔

ansible / awx-plugins / 19658985666

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous