From 9f9b356a7c75e7e1130f0691a4bf4a9e6ad1f2ba Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Thu, 14 Nov 2024 20:32:43 +0000 Subject: [PATCH] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/awx_plugins/inventory/plugins.py | 270 ++++++++++++++------------- 1 file changed, 137 insertions(+), 133 deletions(-) diff --git a/src/awx_plugins/inventory/plugins.py b/src/awx_plugins/inventory/plugins.py index 1d69300270..ebead91d08 100644 --- a/src/awx_plugins/inventory/plugins.py +++ b/src/awx_plugins/inventory/plugins.py @@ -23,7 +23,7 @@ class PluginFileInjector: # every source should have collection, these are for the collection name namespace = None collection = None - collection_migration = "2.9" # Starting with this version, we use collections + collection_migration = '2.9' # Starting with this version, we use collections use_fqcn = False # plugin: name versus plugin: namespace.collection.name # TODO: delete this method and update unit tests @@ -31,7 +31,11 @@ class PluginFileInjector: def get_proper_name(cls) -> str | None: if cls.plugin_name is None: return None - return f"{cls.namespace}.{cls.collection}.{cls.plugin_name}.{cls.plugin_description}" + return f"{ + cls.namespace}.{ + cls.collection}.{ + cls.plugin_name}.{ + cls.plugin_description}" @property def filename(self) -> str: @@ -58,14 +62,14 @@ def inventory_as_dict(self, inventory_update, private_data_dir): Note that a plugin value of '' should still be overridden. """ if self.plugin_name is not None: - source_vars["plugin"] = ( + source_vars['plugin'] = ( f"""{ - self.namespace - }.{ - self.collection - }.{ - self.plugin_name - }""" + self.namespace + }.{ + self.collection + }.{ + self.plugin_name + }""" if self.use_fqcn else self.plugin_name ) @@ -86,7 +90,7 @@ def build_env( env.update(injector_env) # All CLOUD_PROVIDERS sources implement as inventory plugin from # collection - env["ANSIBLE_INVENTORY_ENABLED"] = "auto" + env['ANSIBLE_INVENTORY_ENABLED'] = 'auto' return env def _get_shared_env( @@ -96,7 +100,7 @@ def _get_shared_env( private_data_files, ): # noqa: DAR101, DAR201; FIXME """By default, we will apply the standard managed injectors.""" - if self.base_injector not in {"managed", "template"}: + if self.base_injector not in {'managed', 'template'}: return {} credential = inventory_update.get_cloud_credential() @@ -105,14 +109,14 @@ def _get_shared_env( return {} injected_env = { - "INVENTORY_UPDATE_ID": str( + 'INVENTORY_UPDATE_ID': str( inventory_update.pk, ), # so injector knows this is inventory } - if self.base_injector == "managed": + if self.base_injector == 'managed': from awx_plugins.credentials import injectors as builtin_injectors - cred_kind = inventory_update.source.replace("ec2", "aws") + cred_kind = inventory_update.source.replace('ec2', 'aws') if cred_kind in dir(builtin_injectors): getattr( builtin_injectors, @@ -122,7 +126,7 @@ def _get_shared_env( injected_env, private_data_dir, ) - elif self.base_injector == "template": + elif self.base_injector == 'template': safe_env = injected_env.copy() args = [] credential.credential_type.inject_credential( @@ -161,83 +165,83 @@ def build_plugin_private_data(self, inventory_update, private_data_dir): class azure_rm(PluginFileInjector): - plugin_name = "azure_rm" - plugin_description = "Microsoft Azure Resource Manager" - base_injector = "managed" - namespace = "azure" - collection = "azcollection" + plugin_name = 'azure_rm' + plugin_description = 'Microsoft Azure Resource Manager' + base_injector = 'managed' + namespace = 'azure' + collection = 'azcollection' def get_plugin_env(self, *args, **kwargs): ret = super().get_plugin_env(*args, **kwargs) # We need native jinja2 types so that tags can give JSON null value - ret["ANSIBLE_JINJA2_NATIVE"] = str(True) + ret['ANSIBLE_JINJA2_NATIVE'] = str(True) return ret class ec2(PluginFileInjector): - plugin_name = "aws_ec2" - plugin_description = "Amazon EC2" - base_injector = "managed" - namespace = "amazon" - collection = "aws" + plugin_name = 'aws_ec2' + plugin_description = 'Amazon EC2' + base_injector = 'managed' + namespace = 'amazon' + collection = 'aws' def get_plugin_env(self, *args, **kwargs): ret = super().get_plugin_env(*args, **kwargs) # We need native jinja2 types so that ec2_state_code will give integer - ret["ANSIBLE_JINJA2_NATIVE"] = str(True) + ret['ANSIBLE_JINJA2_NATIVE'] = str(True) return ret class gce(PluginFileInjector): - plugin_name = "gcp_compute" - plugin_description = "Google Compute Engine" - base_injector = "managed" - namespace = "google" - collection = "cloud" + plugin_name = 'gcp_compute' + plugin_description = 'Google Compute Engine' + base_injector = 'managed' + namespace = 'google' + collection = 'cloud' def get_plugin_env(self, *args, **kwargs): ret = super().get_plugin_env(*args, **kwargs) # We need native jinja2 types so that ip addresses can give JSON null # value - ret["ANSIBLE_JINJA2_NATIVE"] = str(True) + ret['ANSIBLE_JINJA2_NATIVE'] = str(True) return ret def inventory_as_dict(self, inventory_update, private_data_dir): ret = super().inventory_as_dict(inventory_update, private_data_dir) credential = inventory_update.get_cloud_credential() # InventorySource.source_vars take precedence over ENV vars - if "projects" not in ret: - ret["projects"] = [credential.get_input("project", default="")] + if 'projects' not in ret: + ret['projects'] = [credential.get_input('project', default='')] return ret class vmware(PluginFileInjector): - plugin_name = "vmware_vm_inventory" - plugin_description = "VMware vCenter" - base_injector = "managed" - namespace = "community" - collection = "vmware" + plugin_name = 'vmware_vm_inventory' + plugin_description = 'VMware vCenter' + base_injector = 'managed' + namespace = 'community' + collection = 'vmware' class openstack(PluginFileInjector): - plugin_name = "openstack" - plugin_description = "OpenStack" - namespace = "openstack" - collection = "cloud" + plugin_name = 'openstack' + plugin_description = 'OpenStack' + namespace = 'openstack' + collection = 'cloud' def _get_clouds_dict(self, inventory_update, cred, private_data_dir): openstack_data = _openstack_data(cred) - openstack_data["clouds"]["devstack"]["private"] = ( + openstack_data['clouds']['devstack']['private'] = ( inventory_update.source_vars_dict.get( - "private", + 'private', True, ) ) ansible_variables = { - "use_hostnames": True, - "expand_hostvars": False, - "fail_on_errors": True, + 'use_hostnames': True, + 'expand_hostvars': False, + 'fail_on_errors': True, } provided_count = 0 for var_name in ansible_variables: @@ -249,19 +253,19 @@ def _get_clouds_dict(self, inventory_update, cred, private_data_dir): if provided_count: # Must we provide all 3 because the user provides any 1 of these?? # this probably results in some incorrect mangling of the defaults - openstack_data["ansible"] = ansible_variables + openstack_data['ansible'] = ansible_variables return openstack_data def build_plugin_private_data(self, inventory_update, private_data_dir): credential = inventory_update.get_cloud_credential() - private_data = {"credentials": {}} + private_data = {'credentials': {}} openstack_data = self._get_clouds_dict( inventory_update, credential, private_data_dir, ) - private_data["credentials"][credential] = yaml.safe_dump( + private_data['credentials'][credential] = yaml.safe_dump( openstack_data, default_flow_style=False, allow_unicode=True, @@ -280,8 +284,8 @@ def get_plugin_env( private_data_files, ) credential = inventory_update.get_cloud_credential() - cred_data = private_data_files["credentials"] - env["OS_CLIENT_CONFIG_FILE"] = get_incontainer_path( + cred_data = private_data_files['credentials'] + env['OS_CLIENT_CONFIG_FILE'] = get_incontainer_path( cred_data[credential], private_data_dir, ) @@ -291,32 +295,32 @@ def get_plugin_env( class rhv(PluginFileInjector): """Ovirt uses the custom credential templating, and that is all.""" - plugin_name = "ovirt" - plugin_description = "Red Hat Virtualization" - base_injector = "template" - initial_version = "2.9" - namespace = "ovirt" - collection = "ovirt" + plugin_name = 'ovirt' + plugin_description = 'Red Hat Virtualization' + base_injector = 'template' + initial_version = '2.9' + namespace = 'ovirt' + collection = 'ovirt' use_fqcn = True class rhv_supported(PluginFileInjector): """Ovirt uses the custom credential templating, and that is all.""" - plugin_name = "ovirt" - plugin_description = "Supported Red Hat Virtualization" - base_injector = "template" - initial_version = "2.9" - namespace = "redhat" - collection = "rhv" + plugin_name = 'ovirt' + plugin_description = 'Supported Red Hat Virtualization' + base_injector = 'template' + initial_version = '2.9' + namespace = 'redhat' + collection = 'rhv' use_fqcn = True class satellite6(PluginFileInjector): - plugin_name = "foreman" - plugin_description = "Red Hat Satellite 6" - namespace = "theforeman" - collection = "foreman" + plugin_name = 'foreman' + plugin_description = 'Red Hat Satellite 6' + namespace = 'theforeman' + collection = 'foreman' use_fqcn = True def get_plugin_env( @@ -334,20 +338,20 @@ def get_plugin_env( private_data_files, ) if credential: - ret["FOREMAN_SERVER"] = credential.get_input("host", default="") - ret["FOREMAN_USER"] = credential.get_input("username", default="") - ret["FOREMAN_PASSWORD"] = credential.get_input( - "password", - default="", + ret['FOREMAN_SERVER'] = credential.get_input('host', default='') + ret['FOREMAN_USER'] = credential.get_input('username', default='') + ret['FOREMAN_PASSWORD'] = credential.get_input( + 'password', + default='', ) return ret class satellite6_supported(PluginFileInjector): - plugin_name = "foreman" - plugin_description = "Supported Red Hat Satellite 6" - namespace = "redhat" - collection = "satellite" + plugin_name = 'foreman' + plugin_description = 'Supported Red Hat Satellite 6' + namespace = 'redhat' + collection = 'satellite' use_fqcn = True def get_plugin_env( @@ -365,34 +369,34 @@ def get_plugin_env( private_data_files, ) if credential: - ret["FOREMAN_SERVER"] = credential.get_input("host", default="") - ret["FOREMAN_USER"] = credential.get_input("username", default="") - ret["FOREMAN_PASSWORD"] = credential.get_input( - "password", - default="", + ret['FOREMAN_SERVER'] = credential.get_input('host', default='') + ret['FOREMAN_USER'] = credential.get_input('username', default='') + ret['FOREMAN_PASSWORD'] = credential.get_input( + 'password', + default='', ) return ret class terraform(PluginFileInjector): - plugin_name = "terraform_state" - plugin_description = "Terraform State" - namespace = "cloud" - collection = "terraform" + plugin_name = 'terraform_state' + plugin_description = 'Terraform State' + namespace = 'cloud' + collection = 'terraform' use_fqcn = True def inventory_as_dict(self, inventory_update, private_data_dir): ret = super().inventory_as_dict(inventory_update, private_data_dir) credential = inventory_update.get_cloud_credential() - config_cred = credential.get_input("configuration") + config_cred = credential.get_input('configuration') if config_cred: handle, path = tempfile.mkstemp( - dir=os.path.join(private_data_dir, "env"), + dir=os.path.join(private_data_dir, 'env'), ) - with os.fdopen(handle, "w") as f: + with os.fdopen(handle, 'w') as f: os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) f.write(config_cred) - ret["backend_config_files"] = get_incontainer_path( + ret['backend_config_files'] = get_incontainer_path( path, private_data_dir, ) @@ -401,10 +405,10 @@ def inventory_as_dict(self, inventory_update, private_data_dir): def build_plugin_private_data(self, inventory_update, private_data_dir): credential = inventory_update.get_cloud_credential() - private_data = {"credentials": {}} - gce_cred = credential.get_input("gce_credentials", default=None) + private_data = {'credentials': {}} + gce_cred = credential.get_input('gce_credentials', default=None) if gce_cred: - private_data["credentials"][credential] = gce_cred + private_data['credentials'][credential] = gce_cred return private_data def get_plugin_env( @@ -419,9 +423,9 @@ def get_plugin_env( private_data_files, ) credential = inventory_update.get_cloud_credential() - cred_data = private_data_files["credentials"] + cred_data = private_data_files['credentials'] if credential in cred_data: - env["GOOGLE_BACKEND_CREDENTIALS"] = get_incontainer_path( + env['GOOGLE_BACKEND_CREDENTIALS'] = get_incontainer_path( cred_data[credential], private_data_dir, ) @@ -431,69 +435,69 @@ def get_plugin_env( class controller(PluginFileInjector): # TODO: relying on routing for now, update after EEs pick up revised # collection - plugin_name = "tower" - plugin_description = "Red Hat Ansible Automation Platform" - base_injector = "template" - namespace = "awx" - collection = "awx" + plugin_name = 'tower' + plugin_description = 'Red Hat Ansible Automation Platform' + base_injector = 'template' + namespace = 'awx' + collection = 'awx' class controller_supported(PluginFileInjector): # TODO: relying on routing for now, update after EEs pick up revised # collection - plugin_name = "tower" - plugin_description = "Supported Red Hat Ansible Automation Platform" - base_injector = "template" - namespace = "ansible" - collection = "controller" + plugin_name = 'tower' + plugin_description = 'Supported Red Hat Ansible Automation Platform' + base_injector = 'template' + namespace = 'ansible' + collection = 'controller' class insights(PluginFileInjector): - plugin_name = "insights" - plugin_description = "Red Hat Insights" - base_injector = "template" - namespace = "redhatinsights" - collection = "insights" + plugin_name = 'insights' + plugin_description = 'Red Hat Insights' + base_injector = 'template' + namespace = 'redhatinsights' + collection = 'insights' use_fqcn = True class insights_supported(PluginFileInjector): - plugin_name = "insights" - plugin_description = "Supported Red Hat Insights" - base_injector = "template" - namespace = "redhat" - collection = "insights" + plugin_name = 'insights' + plugin_description = 'Supported Red Hat Insights' + base_injector = 'template' + namespace = 'redhat' + collection = 'insights' use_fqcn = True class openshift_virtualization(PluginFileInjector): - plugin_name = "kubevirt" - plugin_description = "OpenShift Virtualization" - base_injector = "template" - namespace = "kubevirt" - collection = "core" + plugin_name = 'kubevirt' + plugin_description = 'OpenShift Virtualization' + base_injector = 'template' + namespace = 'kubevirt' + collection = 'core' use_fqcn = True class openshift_virtualization_supported(PluginFileInjector): - plugin_name = "kubevirt" - plugin_description = "Supported OpenShift Virtualization" - base_injector = "template" - namespace = "redhat" - collection = "openshift_virtualization" + plugin_name = 'kubevirt' + plugin_description = 'Supported OpenShift Virtualization' + base_injector = 'template' + namespace = 'redhat' + collection = 'openshift_virtualization' use_fqcn = True class constructed(PluginFileInjector): - plugin_name = "constructed" - plugin_description = "Template additional groups and hostvars at runtime" - namespace = "ansible" - collection = "builtin" + plugin_name = 'constructed' + plugin_description = 'Template additional groups and hostvars at runtime' + namespace = 'ansible' + collection = 'builtin' def build_env(self, *args, **kwargs): env = super().build_env(*args, **kwargs) # Enable script inventory plugin so we pick up the script files from # source inventories - env["ANSIBLE_INVENTORY_ENABLED"] += ",script" - env["ANSIBLE_INVENTORY_ANY_UNPARSED_IS_FAILED"] = "True" + env['ANSIBLE_INVENTORY_ENABLED'] += ',script' + env['ANSIBLE_INVENTORY_ANY_UNPARSED_IS_FAILED'] = 'True' return env