flake8 and pylint

Author: chrismeyersfsu

.flake8

@@ -110,6 +110,8 @@ per-file-ignores =
   # additionally test docstrings don't need param lists (DAR, DCO020):
   tests/**.py: DAR, DCO020, S101, S105, S108, S404, S603, WPS202, WPS210, WPS430, WPS436, WPS441, WPS442, WPS450
 
+  tests/_temporary_private_inject_api_test.py: DAR, DCO020, S101, S105, S108, S404, S603, WPS202, WPS210, WPS226, WPS430, WPS436, WPS441, WPS442, WPS450, WWPS202
+
   src/awx_plugins/interfaces/_temporary_private_inject_api.py: ANN001,ANN201,B950,C901,CCR001,D103,E800,LN001,LN002,Q003,WPS110,WPS111,WPS118,WPS125,WPS204,WPS210,WPS211,WPS213,WPS221,WPS226,WPS231,WPS232,WPS319,WPS323,WPS336,WPS337,WPS361,WPS421,WPS429,WPS430,WPS431,WPS436,WPS442,WPS503,WPS507,WPS516,WPS226
 
 

.pylintrc.toml

@@ -422,6 +422,7 @@ disable = [
   "useless-import-alias",  # MyPy requires the opposite
   "wrong-import-order",  # isort-handled: https://github.com/pylint-dev/pylint/issues/9977
   "wrong-import-position",  # isort-handled: https://github.com/pylint-dev/pylint/issues/9977
+  "relative-beyond-top-level",  # Developer preference
 ]
 
 # Enable the message, report, category or checker with the given id(s). You can

src/awx_plugins/interfaces/_temporary_private_api.py

@@ -12,9 +12,9 @@
 )
 
 
-InputSchemaValueType = list[dict[str, str | bool]]
-InputSchemaType = dict[str, InputSchemaValueType]
-
+InputDefinitionValueType = list[dict[str, str | bool]]
+InputDefinitionType = dict[str, InputDefinitionValueType]
+InjectorDefinitionType = dict[str, dict[str, str]]
 
 try:
     # pylint: disable-next=unused-import
@@ -37,10 +37,10 @@ class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
         kind: str
         """Plugin category."""
 
-        inputs: InputSchemaType
+        inputs: InputDefinitionType
         """UI input fields schema."""
 
-        injectors: dict[str, dict[str, str]] | None = None
+        injectors: InjectorDefinitionType | None = None
         """Injector hook parameters."""
 
         managed: bool = False
@@ -54,6 +54,7 @@ class ManagedCredentialType:  # type: ignore[no-redef]  # noqa: WPS440
         ] | None = None
         """Function to call as an alternative to the templated injection."""
 
+        # pylint: disable-next=R0913,R0917
         def inject_credential(  # noqa: WPS211
             self: 'ManagedCredentialType',
             credential: Credential,
@@ -69,7 +70,8 @@ def inject_credential(  # noqa: WPS211
             inject_credential is called directly. Once the above awx
             import hack is removed the duplication can be removed.
             """
-            from ._temporary_private_inject_api import (  # noqa: WPS433, WPS436
+            # pylint: disable-next=C0415,E0402
+            from ._temporary_private_inject_api import (  # noqa: WPS433,WPS436
                 inject_credential,
             )
             inject_credential(

src/awx_plugins/interfaces/_temporary_private_inject_api.py

@@ -1,4 +1,4 @@
-"""Injection makes use of plugins."""
+"""Injectors exercise plugins."""
 
 import os
 import re
@@ -18,6 +18,15 @@
 )
 
 
+# pylint: disable-next=too-few-public-methods
+class TowerNamespace:
+    """Dummy class."""
+
+
+TowerNamespaceValueType = TowerNamespace | GenericOptionalPrimitiveType
+ExtraVarsType = dict[str, str] | list[str] | str
+
+
 HIDDEN_PASSWORD = '*' * 10
 SENSITIVE_ENV_VAR_NAMES = 'API|TOKEN|KEY|SECRET|PASS'
 
@@ -61,7 +70,7 @@ def build_safe_env(
     :returns: Sanitized environment variables.
     """
     safe_env = dict(env)
-    for env_k, env_val in safe_env.items():
+    for env_k, env_v in safe_env.items():
         is_special = (
             env_k == 'AWS_ACCESS_KEY_ID'
             or (
@@ -72,23 +81,29 @@ def build_safe_env(
         )
         if is_special:
             continue
-        elif HIDDEN_PASSWORD_RE.search(env_k):
+        if HIDDEN_PASSWORD_RE.search(env_k):
             safe_env[env_k] = HIDDEN_PASSWORD
-        elif isinstance(env_val, str) and HIDDEN_URL_PASSWORD_RE.match(env_val):
+        elif isinstance(env_v, str) and HIDDEN_URL_PASSWORD_RE.match(env_v):
             safe_env[env_k] = HIDDEN_URL_PASSWORD_RE.sub(
-                HIDDEN_PASSWORD, env_val,
+                HIDDEN_PASSWORD, env_v,
             )
     return safe_env
 
 
 def secret_fields(cred_type: ManagedCredentialType) -> list[str]:
+    """List of fields that are sensitive from the credential type.
+
+    :param cred_type: Where the secret field descriptions live
+    :return: list of secret field names
+    """
     return [
         str(field['id'])
         for field in cred_type.inputs.get('fields', [])
         if field.get('secret', False) is True
     ]
 
 
+# pylint: disable-next=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-branches,too-many-statements
 def inject_credential(
     cred_type: ManagedCredentialType,
     credential: Credential,
@@ -97,6 +112,7 @@ def inject_credential(
     args: list[GenericOptionalPrimitiveType],
     private_data_dir: str,
 ) -> None:
+    # pylint: disable=unidiomatic-typecheck
     """Inject credential data.
 
     Inject credential data into the environment variables and
@@ -133,21 +149,19 @@ def inject_credential(
             safe_env.update(build_safe_env(injected_env))
         return
 
-    class TowerNamespace:
-        """Dummy class."""
-
     tower_namespace = TowerNamespace()
 
     # maintain a normal namespace for building the ansible-playbook
     # arguments (env and args)
-    namespace: dict[str, TowerNamespace | GenericOptionalPrimitiveType] = {
+    namespace: dict[str, TowerNamespaceValueType] = {
         'tower': tower_namespace,
     }
 
     # maintain a sanitized namespace for building the DB-stored arguments
     # (safe_env)
-    safe_namespace: dict[str, TowerNamespace | GenericOptionalPrimitiveType] = {
-        'tower': tower_namespace, }
+    safe_namespace: dict[str, TowerNamespaceValueType] = {
+        'tower': tower_namespace,
+    }
 
     # build a normal namespace with secret values decrypted (for
     # ansible-playbook) and a safe namespace with secret values hidden (for
@@ -170,8 +184,9 @@ class TowerNamespace:
 
     for field in cred_type.inputs.get('fields', []):
         field_id = str(field['id'])
+        field_type_is_bool = field['type'] == 'boolean'
         # default missing boolean fields to False
-        if field['type'] == 'boolean' and field_id not in credential.get_input_keys():
+        if field_type_is_bool and field_id not in credential.get_input_keys():
             namespace[field_id] = False
             safe_namespace[field_id] = False
         # make sure private keys end with a \n
@@ -193,20 +208,24 @@ class TowerNamespace:
             **namespace,
         )  # type: ignore[misc]
         env_dir = os.path.join(private_data_dir, 'env')
-        _, path = tempfile.mkstemp(dir=env_dir)
-        with open(path, 'w') as f:
+        path = tempfile.mkstemp(dir=env_dir)[1]
+        with open(path, 'w') as f:  # pylint: disable=unspecified-encoding
             f.write(data)
         os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
         container_path = get_incontainer_path(path, private_data_dir)
 
         # determine if filename indicates single file or many
         if file_label.find('.') == -1:
-            tower_namespace.filename = container_path
+            tower_namespace.filename = container_path  # pylint: disable=attribute-defined-outside-init
         else:
             if not hasattr(tower_namespace, 'filename'):
-                tower_namespace.filename = TowerNamespace()
+                tower_namespace.filename = TowerNamespace(
+                )  # pylint: disable=attribute-defined-outside-init
             file_label = file_label.split('.')[1]
-            setattr(tower_namespace.filename, file_label, container_path)
+            setattr(
+                tower_namespace.filename,
+                file_label,
+                container_path)  # pylint: disable=attribute-defined-outside-init
 
     for env_var, tmpl in cred_type.injectors.get('env', {}).items():
         if env_var in ENV_BLOCKLIST:
@@ -220,23 +239,25 @@ class TowerNamespace:
         # awx-manage inventory_update does not support extra_vars via -e
         def build_extra_vars(
             node: dict[str, str | list[str]] | list[str] | str,
-        ) -> dict[str, str] | list[str] | str:
+        ) -> ExtraVarsType:
             if isinstance(node, dict):
                 return {
-                    build_extra_vars(k): build_extra_vars(v) for k,
+                    build_extra_vars(entry): build_extra_vars(v) for entry,
                     v in node.items()
                 }
-            elif isinstance(node, list):
-                return [build_extra_vars(x) for x in node]
-            else:
-                return sandbox_env.from_string(node).render(**namespace)
-
-        def build_extra_vars_file(vars, private_dir: str) -> str:
+            if isinstance(node, list):
+                return [build_extra_vars(entry) for entry in node]
+            return sandbox_env.from_string(node).render(**namespace)
+
+        def build_extra_vars_file(
+            extra_vars: ExtraVarsType,
+            private_dir: str,
+        ) -> str:
             handle, path = tempfile.mkstemp(
                 dir=os.path.join(private_dir, 'env'),
             )
             f = os.fdopen(handle, 'w')
-            f.write(yaml_safe_dump(vars))
+            f.write(yaml_safe_dump(extra_vars))
             f.close()
             os.chmod(path, stat.S_IRUSR)
             return path
@@ -249,4 +270,5 @@ def build_extra_vars_file(vars, private_dir: str) -> str:
         if extra_vars:
             path = build_extra_vars_file(extra_vars, private_data_dir)
             container_path = get_incontainer_path(path, private_data_dir)
-            args.extend(['-e', '@%s' % container_path])
+            args.extend(['-e', '@%s' % container_path]
+                        )  # pylint: disable=consider-using-f-string

tests/_temporary_private_api_test.py

@@ -1,11 +1,8 @@
 """Tests for the temporarily hosted private helpers."""
 
-from awx_plugins.interfaces._temporary_private_api import (
-    ManagedCredentialType,
-)
+from awx_plugins.interfaces._temporary_private_api import ManagedCredentialType
 
 
 def test_managed_credential_type_instantiation() -> None:
     """Check that managed credential type can be instantiated."""
-
     assert ManagedCredentialType('', '', '', {})