diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ae7460ed..14e60d7b 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,7 +1,7 @@ --- repos: - repo: https://github.com/ansible-network/collection_prep - rev: da78082ea59b03ad16cd7dbee267f53e8ff50bb5 + rev: 1.1.2 hooks: - id: update-docs @@ -17,7 +17,7 @@ repos: - id: trailing-whitespace - repo: https://github.com/asottile/add-trailing-comma.git - rev: v3.1.0 + rev: v4.0.0 hooks: - id: add-trailing-comma args: @@ -53,20 +53,20 @@ repos: - prettier - prettier-plugin-toml - - repo: https://github.com/psf/black - rev: 24.8.0 + - repo: https://github.com/psf/black-pre-commit-mirror + rev: 25.11.0 hooks: - id: black - repo: https://github.com/astral-sh/ruff-pre-commit - rev: "v0.6.1" + rev: "v0.14.5" hooks: - id: ruff args: - "--exit-non-zero-on-fix" - repo: https://github.com/streetsidesoftware/cspell-cli - rev: v8.13.3 + rev: v9.3.3 hooks: - id: cspell name: Spell check with cspell @@ -77,7 +77,7 @@ repos: - id: darglint - repo: https://github.com/pycqa/pylint.git - rev: v3.2.6 + rev: v4.0.3 hooks: - id: pylint args: @@ -88,7 +88,7 @@ repos: - pytest-ansible - repo: https://github.com/pre-commit/mirrors-mypy.git - rev: v1.11.1 + rev: v1.18.2 hooks: - id: mypy additional_dependencies: diff --git a/plugins/action/git_publish.py b/plugins/action/git_publish.py index 7465d3eb..83451bc1 100644 --- a/plugins/action/git_publish.py +++ b/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 diff --git a/plugins/action/git_retrieve.py b/plugins/action/git_retrieve.py index 4aea3943..9af3fa9b 100644 --- a/plugins/action/git_retrieve.py +++ b/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,7 +37,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) @@ -45,7 +45,7 @@ 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 diff --git a/plugins/plugin_utils/command.py b/plugins/plugin_utils/command.py index be83ba63..c41a1a8c 100644 --- a/plugins/plugin_utils/command.py +++ b/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,19 +27,19 @@ 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. @@ -47,7 +47,7 @@ def command(self: T) -> str: 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. diff --git a/plugins/plugin_utils/git_base.py b/plugins/plugin_utils/git_base.py index 34fceced..9621b482 100644 --- a/plugins/plugin_utils/git_base.py +++ b/plugins/plugin_utils/git_base.py @@ -12,7 +12,7 @@ from dataclasses import dataclass, field, fields from types import ModuleType -from typing import Dict, List, Tuple, TypeVar, Union +from typing import TypeVar, Union from ansible.parsing.dataloader import DataLoader from ansible.playbook.play_context import PlayContext @@ -26,7 +26,7 @@ # 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 T = TypeVar("T", bound="ActionInit") # pylint: disable=invalid-name, useless-suppression @@ -44,8 +44,8 @@ class ActionInit: @property def asdict( - self: T, - ) -> Dict[str, Union[Connection, DataLoader, PlayContext, ModuleType, Task, Templar]]: + self, + ) -> dict[str, Connection | DataLoader | PlayContext | ModuleType | Task | Templar]: """Create a dictionary, avoiding the deepcopy with dataclass.asdict. :return: A dictionary of the keyword arguments. @@ -60,7 +60,7 @@ class ResultBase: changed: bool = True failed: bool = False msg: str = "" - output: List[Dict[str, Union[int, Dict[str, str], List[str], str]]] = field( + output: list[dict[str, int | dict[str, str] | list[str] | str]] = field( default_factory=list, ) @@ -71,7 +71,7 @@ class ResultBase: class GitBase(ActionBase): # type: ignore[misc] # parent has type Any """Base class for the git paction plugins.""" - def __init__(self: U, action_init: ActionInit) -> None: + def __init__(self, action_init: ActionInit) -> None: """Initialize the action plugin. :param action_init: The keyword arguments for action base @@ -81,7 +81,7 @@ def __init__(self: U, action_init: ActionInit) -> None: self._timeout: int @staticmethod - def _git_auth_header(token: str) -> Tuple[str, List[str]]: + def _git_auth_header(token: str) -> tuple[str, list[str]]: """Create the authorization header. helpful: https://github.com/actions/checkout/blob/main/src/git-auth-helper.ts#L56 @@ -97,7 +97,7 @@ def _git_auth_header(token: str) -> Tuple[str, List[str]]: ] return basic_encoded, cli_parameters - def _run_command(self: U, command: Command, ignore_errors: bool = False) -> None: + def _run_command(self, command: Command, ignore_errors: bool = False) -> None: """Run a command and append the command result to the results. :param command: The command to run diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 29b863a6..e5fb532b 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -20,7 +20,7 @@ from .definitions import ActionModuleInit -@pytest.fixture() +@pytest.fixture def action_init() -> ActionModuleInit: """Provide a fixture for action initialization. diff --git a/tests/unit/definitions.py b/tests/unit/definitions.py index 775aec62..670808a1 100644 --- a/tests/unit/definitions.py +++ b/tests/unit/definitions.py @@ -9,8 +9,6 @@ import types -from typing import Dict, Union - from ansible.parsing.dataloader import DataLoader from ansible.playbook.play_context import PlayContext from ansible.playbook.task import Task @@ -18,7 +16,7 @@ from ansible.template import Templar -ActionModuleInit = Dict[ +ActionModuleInit = dict[ str, - Union[Connection, PlayContext, DataLoader, Task, types.ModuleType, Templar], + Connection | PlayContext | DataLoader | Task | types.ModuleType | Templar, ] diff --git a/tests/unit/test_smoke.py b/tests/unit/test_smoke.py index d5b3255a..c98e21da 100644 --- a/tests/unit/test_smoke.py +++ b/tests/unit/test_smoke.py @@ -7,7 +7,6 @@ __metaclass__ = type # pylint: enable=invalid-name -from typing import Union import pytest @@ -31,7 +30,7 @@ ) def test_fail_argspec( action_init: ActionModuleInit, - module: Union[GitPublishActionModule, GitRetrieveActionModule], + module: GitPublishActionModule | GitRetrieveActionModule, ) -> None: """Test an argspec failure.