Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
---
repos:
- repo: https://github.com/ansible-network/collection_prep
rev: da78082ea59b03ad16cd7dbee267f53e8ff50bb5
rev: 1.1.2
hooks:
- id: update-docs

Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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:
Expand Down
36 changes: 18 additions & 18 deletions plugins/action/git_publish.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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")
Expand All @@ -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"])
Expand All @@ -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"])
Expand All @@ -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"])
Expand All @@ -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)
Expand All @@ -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")
Expand All @@ -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"])
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down
42 changes: 21 additions & 21 deletions plugins/action/git_retrieve.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 = ""

Expand All @@ -62,7 +62,7 @@ class ActionModule(GitBase):
_requires_connection = False

def __init__( # noqa: PLR0913
self: T,
self,
connection: Connection,
loader: DataLoader,
play_context: PlayContext,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand All @@ -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 ""
Expand All @@ -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 ""
Expand Down Expand Up @@ -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"])
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down
16 changes: 8 additions & 8 deletions plugins/plugin_utils/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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.
Expand Down
Loading