[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

plugins/action/git_publish.py

@@ -13,7 +13,7 @@
 from contextlib import suppress
 from dataclasses import asdict, dataclass
 from pathlib import Path
-from typing import Dict, List, Optional, TypeVar, Union
+from typing import TypeVar, Union
 
 from ansible.errors import AnsibleActionFail
 from ansible.parsing.dataloader import DataLoader
@@ -38,7 +38,7 @@
 # pylint: enable=invalid-name
 
 # mypy disallow you from omitting parameters in generic types
-JSONTypes = Union[bool, int, str, Dict, List]  # type:ignore
+JSONTypes = Union[bool, int, str, dict, list]  # type:ignore
 
 
 @dataclass(frozen=False)
@@ -61,7 +61,7 @@ class ActionModule(GitBase):
 
     # pylint: disable=too-many-arguments
     def __init__(  # noqa: PLR0913
-        self: T,
+        self,
         connection: Connection,
         loader: DataLoader,
         play_context: PlayContext,
@@ -94,10 +94,10 @@ def __init__(  # noqa: PLR0913
         self._play_name: str = ""
         self._supports_async = True
         self._result: Result = Result()
-        self._env: Optional[Dict[str, str]] = None
-        self._temp_ssh_key_path: Optional[str] = None
+        self._env: dict[str, str] | None = None
+        self._temp_ssh_key_path: str | None = None
 
-    def _check_argspec(self: T) -> None:
+    def _check_argspec(self) -> None:
         """Check the argspec for the action plugin.
 
         :raises AnsibleActionFail: If the argspec is invalid
@@ -117,7 +117,7 @@ def _check_argspec(self: T) -> None:
             msg = "Parameters `ssh_key_file` and `ssh_key_content` are mutually exclusive."
             raise AnsibleActionFail(msg)
 
-    def _prepare_ssh_environment(self: T) -> None:
+    def _prepare_ssh_environment(self) -> None:
         """Prepare the environment for SSH key authentication."""
         key_content = self._task.args.get("ssh_key_content")
         key_file = self._task.args.get("ssh_key_file")
@@ -137,12 +137,12 @@ def _prepare_ssh_environment(self: T) -> None:
         ssh_command = f"ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no"
         self._env = {"GIT_SSH_COMMAND": ssh_command}
 
-    def _cleanup_ssh_key(self: T) -> None:
+    def _cleanup_ssh_key(self) -> None:
         """Remove the temporary SSH key file if it was created."""
         if self._temp_ssh_key_path:
             Path(self._temp_ssh_key_path).unlink()
 
-    def _configure_git_user_name(self: T) -> None:
+    def _configure_git_user_name(self) -> None:
         """Configure the git user name."""
         command_parts = list(self._base_command)
         command_parts.extend(["config", "--get", "user.name"])
@@ -167,7 +167,7 @@ def _configure_git_user_name(self: T) -> None:
         self._run_command(command=command)
         self._result.user_name = name
 
-    def _configure_git_user_email(self: T) -> None:
+    def _configure_git_user_email(self) -> None:
         """Configure the git user email."""
         command_parts = list(self._base_command)
         command_parts.extend(["config", "--get", "user.email"])
@@ -192,7 +192,7 @@ def _configure_git_user_email(self: T) -> None:
         self._run_command(command=command)
         self._result.user_email = email
 
-    def _add(self: T) -> None:
+    def _add(self) -> None:
         """Add files for the pending commit."""
         command_parts = list(self._base_command)
         files = " ".join(self._task.args["include"])
@@ -204,7 +204,7 @@ def _add(self: T) -> None:
         )
         self._run_command(command=command)
 
-    def _commit(self: T) -> None:
+    def _commit(self) -> None:
         """Perform a commit for the pending push."""
         command_parts = list(self._base_command)
         message = self._task.args["commit"]["message"].format(play_name=self._play_name)
@@ -217,7 +217,7 @@ def _commit(self: T) -> None:
         )
         self._run_command(command=command)
 
-    def _tag(self: T) -> None:
+    def _tag(self) -> None:
         """Create a tag object."""
         command_parts = list(self._base_command)
         message = self._task.args["tag"].get("message")
@@ -233,7 +233,7 @@ def _tag(self: T) -> None:
         )
         self._run_command(command=command)
 
-    def _push(self: T) -> None:
+    def _push(self) -> None:
         """Push the commit to the origin."""
         command_parts = list(self._base_command)
         command_parts.extend(["remote", "-v"])
@@ -278,7 +278,7 @@ def _push(self: T) -> None:
                 line for line in command.stderr.split("remote:") if "https" in line
             ).strip()
 
-    def _remove_repo(self: T) -> None:
+    def _remove_repo(self) -> None:
         """Remove the temporary directory."""
         if not self._task.args["remove"]:
             return
@@ -290,10 +290,10 @@ def _remove_repo(self: T) -> None:
             self._result.msg = "Failed to remove repository"
 
     def run(
-        self: T,
+        self,
         tmp: None = None,
-        task_vars: Optional[Dict[str, JSONTypes]] = None,
-    ) -> Dict[str, JSONTypes]:
+        task_vars: dict[str, JSONTypes] | None = None,
+    ) -> dict[str, JSONTypes]:
         """Run the action plugin.
 
         :param tmp: The temporary directory

plugins/action/git_retrieve.py

@@ -12,7 +12,7 @@
 
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple, TypeVar, Union
+from typing import TypeVar, Union
 
 from ansible.errors import AnsibleActionFail
 from ansible.parsing.dataloader import DataLoader
@@ -37,15 +37,15 @@
 # pylint: enable=invalid-name
 
 # mypy disallow you from omitting parameters in generic types
-JSONTypes = Union[bool, int, str, Dict, List]  # type:ignore
+JSONTypes = Union[bool, int, str, dict, list]  # type:ignore
 
 
 @dataclass(frozen=False)
 class Result(ResultBase):
     """Data structure for the task result."""
 
     branch_name: str = ""
-    branches: List[str] = field(default_factory=list)
+    branches: list[str] = field(default_factory=list)
     name: str = ""
     path: str = ""
 
@@ -62,7 +62,7 @@ class ActionModule(GitBase):
     _requires_connection = False
 
     def __init__(  # noqa: PLR0913
-        self: T,
+        self,
         connection: Connection,
         loader: DataLoader,
         play_context: PlayContext,
@@ -90,18 +90,18 @@ def __init__(  # noqa: PLR0913
             ),
         )
 
-        self._base_command: Tuple[str, ...]
-        self._branches: List[str]
+        self._base_command: tuple[str, ...]
+        self._branches: list[str]
         self._branch_name: str
         self._parent_directory: str
         self._repo_path: str
         self._play_name: str = ""
         self._supports_async = True
         self._result: Result = Result()
-        self._temp_ssh_key_path: Optional[str] = None
+        self._temp_ssh_key_path: str | None = None
         self._ssh_command_str: str = "ssh"
 
-    def _check_argspec(self: T) -> None:
+    def _check_argspec(self) -> None:
         """Check the argspec for the action plugin.
 
         :raises AnsibleActionFail: If the argspec is invalid
@@ -129,7 +129,7 @@ def _check_argspec(self: T) -> None:
             )
             raise AnsibleActionFail(msg)
 
-    def _prepare_ssh_environment(self: T) -> None:
+    def _prepare_ssh_environment(self) -> None:
         """Prepare the environment for SSH key authentication."""
         origin_args = self._task.args.get("origin", {})
         key_content = origin_args.get("ssh_key_content")
@@ -151,20 +151,20 @@ def _prepare_ssh_environment(self: T) -> None:
             f"ssh -i {key_path} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no"
         )
 
-    def _cleanup_ssh_key(self: T) -> None:
+    def _cleanup_ssh_key(self) -> None:
         """Remove the temporary SSH key file if it was created."""
         if self._temp_ssh_key_path:
             Path(self._temp_ssh_key_path).unlink()
 
     @property
-    def _branch_exists(self: T) -> bool:
+    def _branch_exists(self) -> bool:
         """Return True if the branch exists.
 
         :returns: True if the branch exists
         """
         return self._branch_name in self._branches
 
-    def _host_key_checking(self: T) -> None:
+    def _host_key_checking(self) -> None:
         """Configure host key checking."""
         origin = self._task.args["origin"]["url"]
         upstream = self._task.args["upstream"].get("url") or ""
@@ -188,7 +188,7 @@ def _host_key_checking(self: T) -> None:
         )
         self._run_command(command=command)
 
-    def _clone(self: T) -> None:
+    def _clone(self) -> None:
         """Clone the repository, creating a new subdirectory."""
         origin = self._task.args["origin"]["url"]
         upstream = self._task.args["upstream"].get("url") or ""
@@ -247,7 +247,7 @@ def _clone(self: T) -> None:
         self._base_command = ("git", "-C", self._repo_path)
         return
 
-    def _get_branches(self: T) -> None:
+    def _get_branches(self) -> None:
         """Get the branches."""
         command_parts = list(self._base_command)
         command_parts.extend(["branch", "-a"])
@@ -294,14 +294,14 @@ def _get_branches(self: T) -> None:
         self._result.branch_name = self._branch_name
         return
 
-    def _detect_duplicate_branch(self: T) -> None:
+    def _detect_duplicate_branch(self) -> None:
         """Detect duplicate branch."""
         duplicate_detection = self._task.args["branch"]["duplicate_detection"]
         if duplicate_detection and self._branch_exists:
             self._result.failed = True
             self._result.msg = f"Branch '{self._branch_name}' already exists"
 
-    def _switch_checkout(self: T) -> None:
+    def _switch_checkout(self) -> None:
         """Switch to or checkout the branch."""
         command_parts = list(self._base_command)
         branch = self._branch_name
@@ -321,7 +321,7 @@ def _switch_checkout(self: T) -> None:
         )
         self._run_command(command=command)
 
-    def _add_upstream_remote(self: T) -> None:
+    def _add_upstream_remote(self) -> None:
         """Add the upstream remote."""
         if not self._task.args["upstream"].get("url"):
             return
@@ -336,7 +336,7 @@ def _add_upstream_remote(self: T) -> None:
         self._run_command(command=command)
         return
 
-    def _pull_upstream(self: T) -> None:
+    def _pull_upstream(self) -> None:
         """Pull from upstream."""
         if not self._task.args["upstream"].get("url"):
             return
@@ -363,10 +363,10 @@ def _pull_upstream(self: T) -> None:
         return
 
     def run(
-        self: T,
+        self,
         tmp: None = None,
-        task_vars: Optional[Dict[str, JSONTypes]] = None,
-    ) -> Dict[str, JSONTypes]:
+        task_vars: dict[str, JSONTypes] | None = None,
+    ) -> dict[str, JSONTypes]:
         """Run the action plugin.
 
         :param tmp: The temporary directory

plugins/plugin_utils/command.py

@@ -11,7 +11,7 @@
 
 
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional, TypeVar, Union
+from typing import TypeVar
 
 
 T = TypeVar("T", bound="Command")  # pylint: disable=invalid-name, useless-suppression
@@ -27,27 +27,27 @@ class Command:
 
     # pylint: disable=too-many-instance-attributes
 
-    command_parts: List[str]
+    command_parts: list[str]
     fail_msg: str
 
-    env: Optional[Dict[str, str]] = None
-    no_log: Dict[str, str] = field(default_factory=dict)
+    env: dict[str, str] | None = None
+    no_log: dict[str, str] = field(default_factory=dict)
     return_code: int = -1
     stdout: str = ""
     stderr: str = ""
-    stdout_lines: List[str] = field(default_factory=list)
-    stderr_lines: List[str] = field(default_factory=list)
+    stdout_lines: list[str] = field(default_factory=list)
+    stderr_lines: list[str] = field(default_factory=list)
 
     @property
-    def command(self: T) -> str:
+    def command(self) -> str:
         """Return the command as a string.
 
         :return: The command as a string.
         """
         return shlex.join(self.command_parts)
 
     @property
-    def cleaned(self: T) -> Dict[str, Union[int, Dict[str, str], List[str], str]]:
+    def cleaned(self) -> dict[str, int | dict[str, str] | list[str] | str]:
         """Return the sanitized details of the command for the log.
 
         :return: The sanitized details of the command for the log.