28891854405

Committed 07 Jul 2026 07:11PM UTC coverage: 76.693%. Remained the same

Build # 28891854405

Build Type

Pull #192

github

Committed by

web-flow

Commit Message

Merge e748cc93b into 7c6a7cf97

Pull Request Pull Request #192: Remove redundant linters, relax over-aggressive checks

Coverage Stats

39 of 39 branches covered (100.0%)

Branch coverage included in aggregate %.

65 of 68 new or added lines in 18 files covered. (95.59%)

2929 of 3831 relevant lines covered (76.46%)

1.53 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 (
    EnvVarsType,
    ManagedCredentialType,
)
from awx_plugins.interfaces._temporary_private_container_api import (
    CONTAINER_ROOT,
)
from awx_plugins.interfaces._temporary_private_credential_api import (
    Credential,
    CredentialInputType,
)
from awx_plugins.interfaces._temporary_private_inject_api import (
    HIDDEN_PASSWORD,
    inject_credential,
)

import yaml

from awx_plugins.credentials._managed_types.terraform import hcp_terraform
from awx_plugins.credentials.plugins import (
    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 because
    MappingProxy did not work 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',
    }


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

ansible / awx-plugins / 28891854405

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