Merge branch 'main' into ruff_RUF005

pyproject.toml

@@ -277,7 +277,6 @@ ignore = [
   'C409', # [*] Unnecessary `list` literal passed to `tuple()` (rewrite as a `tuple` literal)
   'C414', # [*] Unnecessary `list` call within `sorted()`
   'C901', # `_params_row_for_entry` is too complex (11 > 10)
-  'COM812', # [*] Trailing comma missing
   'D100', # Missing docstring in public module
   'D101', # Missing docstring in public class
   'D102', # Missing docstring in public method
@@ -306,22 +305,16 @@ ignore = [
   'PT019', # Fixture `_mocked_func` without value is injected as parameter, use `@pytest.mark.usefixtures` instead
   'PT022', # [*] No teardown in fixture `cmd_in_tty`, use `return` instead of `yield`
   'PTH100', # `os.path.abspath()` should be replaced by `Path.resolve()`
-  'PTH101', # `os.chmod()` should be replaced by `Path.chmod()`
   'PTH103', # `os.makedirs()` should be replaced by `Path.mkdir(parents=True)`
-  'PTH107', # `os.remove()` should be replaced by `Path.unlink()`
   'PTH109', # `os.getcwd()` should be replaced by `Path.cwd()`
   'PTH110', # `os.path.exists()` should be replaced by `Path.exists()`
   'PTH111', # `os.path.expanduser()` should be replaced by `Path.expanduser()`
-  'PTH112', # `os.path.isdir()` should be replaced by `Path.is_dir()`
   'PTH113', # `os.path.isfile()` should be replaced by `Path.is_file()`
-  'PTH114', # `os.path.islink()` should be replaced by `Path.is_symlink()`
-  'PTH115', # `os.readlink()` should be replaced by `Path.readlink()`
   'PTH116', # `os.stat()` should be replaced by `Path.stat()`, `Path.owner()`, or `Path.group()`
   'PTH118', # `os.path.join()` should be replaced by `Path` with `/` operator
   'PTH119', # `os.path.basename()` should be replaced by `Path.name`
   'PTH120', # `os.path.dirname()` should be replaced by `Path.parent`
   'PTH122', # `os.path.splitext()` should be replaced by `Path.suffix`
-  'PTH204', # `os.path.getmtime` should be replaced by `Path.stat().st_mtime`
   'PTH207', # Replace `glob` with `Path.glob` or `Path.rglob`
   'RET505', # Unnecessary `else` after `return` statement
   'RUF012', # Mutable class attributes should be annotated with `typing.ClassVar`

src/ansible_navigator/action_base.py

@@ -34,7 +34,10 @@ class ActionBase:
     """Base class for actions."""
 
     def __init__(
-        self, args: ApplicationConfiguration, name: str, logger_name: str = __name__
+        self,
+        args: ApplicationConfiguration,
+        name: str,
+        logger_name: str = __name__,
     ) -> None:
         """Initialize the App class.
 

src/ansible_navigator/actions/collections.py

@@ -466,13 +466,15 @@ def _run_runner(self) -> None:
             mount_doc_cache = True
             # Determine if the doc_cache is relative to the cache directory
             if path_is_relative_to(
-                child=Path(self._args.collection_doc_cache_path), parent=(cache_path)
+                child=Path(self._args.collection_doc_cache_path),
+                parent=(cache_path),
             ):
                 mount_doc_cache = False
 
             # The playbook directory will be mounted as host_cwd, so don't duplicate
             if path_is_relative_to(
-                child=Path(self._args.collection_doc_cache_path), parent=(Path(playbook_dir))
+                child=Path(self._args.collection_doc_cache_path),
+                parent=(Path(playbook_dir)),
             ):
                 mount_doc_cache = False
 
@@ -603,7 +605,8 @@ def _parse(self, output: str) -> None:
         return
 
     def _get_collection_plugins_details(
-        self, selected_collection: dict[str, Any]
+        self,
+        selected_collection: dict[str, Any],
     ) -> dict[str, Any]:
         """Get plugin details for the given collection.
 

src/ansible_navigator/actions/images.py

@@ -97,7 +97,10 @@ def color_menu(self, colno: int, colname: str, entry: dict[str, Any]) -> tuple[i
         return 2, 0
 
     def generate_content_heading(
-        self, obj: dict[Any, Any], screen_w: int, name: str = ""
+        self,
+        obj: dict[Any, Any],
+        screen_w: int,
+        name: str = "",
     ) -> CursesLines:
         """Create a heading for image content.
 
@@ -469,7 +472,7 @@ def _collect_image_list(self) -> None:
                 config_label_check = False
 
             image["execution_environment"] = any(
-                (legacy_check, root_label_check, config_label_check)
+                (legacy_check, root_label_check, config_label_check),
             )
         self._images.value = sorted(images, key=lambda i: i["name"])
 
@@ -574,7 +577,9 @@ def _run_runner(self, image_name: str) -> tuple[str, str, int]:
             kwargs.update({"container_options": self._args.container_options})
 
         self._logger.debug(
-            "Invoke runner with executable_cmd: %s and kwargs: %s", python_exec_path, kwargs
+            "Invoke runner with executable_cmd: %s and kwargs: %s",
+            python_exec_path,
+            kwargs,
         )
         _runner = Command(executable_cmd=python_exec_path, **kwargs)
         output, error, return_code = _runner.run()

src/ansible_navigator/actions/inventory.py

@@ -163,15 +163,15 @@ def _set_inventories_mtime(self) -> None:
         """Record the inventory modification time."""
         modification_times = []
         for inventory in self._inventories:
-            if os.path.isdir(inventory):
+            if Path(inventory).is_dir():
                 modification_times.append(
                     max(
-                        os.path.getmtime(e)
+                        Path(e).stat().st_mtime
                         for e in glob.glob(os.path.join(inventory, "**"), recursive=True)
                     ),
                 )
             elif os.path.isfile(inventory):
-                modification_times.append(os.path.getmtime(inventory))
+                modification_times.append(Path(inventory).stat().st_mtime)
         if modification_times:
             self._inventories_mtime = max(modification_times)
         else:

src/ansible_navigator/command_runner/command_runner.py

@@ -79,7 +79,8 @@ def run_command(command: Command) -> None:
 
 
 def worker(
-    pending_queue: multiprocessing.Queue[Any], completed_queue: multiprocessing.Queue[Any]
+    pending_queue: multiprocessing.Queue[Any],
+    completed_queue: multiprocessing.Queue[Any],
 ) -> None:
     """Read pending, run, post process, and place in completed.
 

src/ansible_navigator/configuration_subsystem/definitions.py

@@ -472,7 +472,8 @@ class PaeChangeRequest:
 
 # A type used for the settings as a dictionary
 SettingsFileType = NewType(
-    "SettingsFileType", dict[str, bool | dict[Any, Any] | int | str | list[Any]]
+    "SettingsFileType",
+    dict[str, bool | dict[Any, Any] | int | str | list[Any]],
 )
 
 # A type used to describe a schema file for the settings

src/ansible_navigator/configuration_subsystem/navigator_post_processor.py

@@ -253,7 +253,9 @@ def display_color(
 
     @_post_processor
     def execution_environment(
-        self, entry: SettingsEntry, config: ApplicationConfiguration
+        self,
+        entry: SettingsEntry,
+        config: ApplicationConfiguration,
     ) -> PostProcessorReturn:
         # pylint: disable=too-many-locals
         """Post process execution_environment.
@@ -497,8 +499,9 @@ def _disable_pae_and_enforce_stdout(
             self._requested_mode.append(ModeChangeRequest(entry=entry.name, mode=mode))
             self._requested_pae.append(
                 PaeChangeRequest(
-                    entry=entry.name, playbook_artifact_enable=playbook_artifact_enable
-                )
+                    entry=entry.name,
+                    playbook_artifact_enable=playbook_artifact_enable,
+                ),
             )
             message = (
                 f"`{entry.name} requesting mode {mode.value} and pae as {playbook_artifact_enable}"
@@ -673,7 +676,7 @@ def lintables(
                 except subprocess.CalledProcessError:
                     exit_messages.append(
                         ExitMessage(
-                            message=("ansible-lint does not seem to be installed correctly.")
+                            message=("ansible-lint does not seem to be installed correctly."),
                         ),
                     )
                     exit_messages.append(
@@ -934,7 +937,8 @@ def playbook(entry: SettingsEntry, config: ApplicationConfiguration) -> PostProc
             exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT))
             return messages, exit_messages
         if check_playbook_type(entry.value.current) == "file" and isinstance(
-            entry.value.current, str
+            entry.value.current,
+            str,
         ):
             entry.value.current = abs_user_path(entry.value.current)
         return messages, exit_messages

src/ansible_navigator/data/catalog_collections.py

@@ -356,7 +356,8 @@ def add_pseudo_builtin(self) -> None:
 
 
 def worker(
-    pending_queue: multiprocessing.Queue[Any], completed_queue: multiprocessing.Queue[Any]
+    pending_queue: multiprocessing.Queue[Any],
+    completed_queue: multiprocessing.Queue[Any],
 ) -> None:
     """Extract the documentation from a plugin, place in completed queue.
 
@@ -419,7 +420,8 @@ def worker(
 
 
 def identify_missing(
-    collections: dict[Any, Any], collection_cache: KeyValueStore
+    collections: dict[Any, Any],
+    collection_cache: KeyValueStore,
 ) -> tuple[set[Any], list[Any], int]:
     """Identify plugins missing from the cache.
 

src/ansible_navigator/data/image_introspect.py

@@ -163,7 +163,10 @@ def re_partition(content: Any, separator: str) -> Any:
         return key, delim, content
 
     def splitter(
-        self, lines: list[str], line_split: str, section_delim: str | None = None
+        self,
+        lines: list[str],
+        line_split: str,
+        section_delim: str | None = None,
     ) -> list[dict[str, Any]] | dict[str, Any]:
         """Split lines given a delimiter.
 
@@ -294,7 +297,9 @@ def commands(self) -> list[Command]:
         :returns: The defined command
         """
         pre = Command(
-            id_="pip_freeze", command="/usr/bin/python3 -m pip freeze", parse=self.parse_freeze
+            id_="pip_freeze",
+            command="/usr/bin/python3 -m pip freeze",
+            parse=self.parse_freeze,
         )
         run_command(pre)
         pre.parse(pre)

src/ansible_navigator/initialization.py

@@ -9,6 +9,7 @@
 import os
 import sys
 
+from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import NoReturn
 
@@ -135,7 +136,7 @@ def get_and_check_collection_doc_cache(
         message = "Collection doc cache: version was empty or incorrect, rebuilding"
         messages.append(LogMessage(level=logging.INFO, message=message))
         collection_cache.close()
-        os.remove(collection_doc_cache_path)
+        Path(collection_doc_cache_path).unlink()
         collection_cache = KeyValueStore(collection_doc_cache_path)
         collection_cache["version"] = VERSION_CDC
         cache_version = collection_cache["version"]

src/ansible_navigator/logger.py

@@ -7,6 +7,7 @@
 import os
 import zoneinfo
 
+from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
 
@@ -57,7 +58,7 @@ def setup_logger(args: ApplicationConfiguration) -> None:
     :param args: The CLI args
     """
     if os.path.exists(args.log_file) and args.log_append is False:
-        os.remove(args.log_file)
+        Path(args.log_file).unlink()
     handler = logging.FileHandler(args.log_file)
 
     time_zone = args.entry("time_zone").value.current

src/ansible_navigator/runner/command_async.py

@@ -20,7 +20,11 @@ class CommandAsync(CommandBase):
     """A wrapper for the asynchronous runner."""
 
     def __init__(
-        self, executable_cmd: str, queue: Queue[Any], write_job_events: bool, **kwargs: Any
+        self,
+        executable_cmd: str,
+        queue: Queue[Any],
+        write_job_events: bool,
+        **kwargs: Any,
     ) -> None:
         """Initialize the arguments for the ``run_command_async`` interface of ``ansible-runner``.
 

src/ansible_navigator/ui_framework/form.py

@@ -341,7 +341,8 @@ def _generate_messages(form_field: FieldWorking) -> CursesLines:
         return CursesLines(lines)
 
     def _generate_prompt(
-        self, form_field: FieldText | FieldChecks | FieldRadio
+        self,
+        form_field: FieldText | FieldChecks | FieldRadio,
     ) -> list[CursesLinePart]:
         """Generate the prompt for a field.
 

src/ansible_navigator/ui_framework/form_handler_options.py

@@ -50,7 +50,9 @@ def populate(self, form_field: FieldText, active: int) -> None:
                 text = option.text
             clp_text = CursesLinePart(len(option_code) + 1, text, color, decoration)
             self._add_line(
-                window=self.win, lineno=idx, line=CursesLine((clp_option_code, clp_text))
+                window=self.win,
+                lineno=idx,
+                line=CursesLine((clp_option_code, clp_text)),
             )
 
     def handle(self, idx: int, form_fields: list[FieldText]) -> tuple[FieldText, int]:

src/ansible_navigator/ui_framework/ui.py

@@ -814,7 +814,10 @@ def _get_heading_menu_items(
         return menu_heading, menu_items
 
     def _show_menu(
-        self, current: Sequence[Any], columns: list[str], await_input: bool
+        self,
+        current: Sequence[Any],
+        columns: list[str],
+        await_input: bool,
     ) -> Interaction:
         """Show a menu on the screen.
 

src/ansible_navigator/utils/functions.py

@@ -503,7 +503,7 @@ def time_stamp_for_file(path: str, time_zone: str) -> tuple[float | None, str |
     :returns: The UNIX timestamp and an ISO 8601 string
     """
     try:
-        modified = os.path.getmtime(path)
+        modified = Path(path).stat().st_mtime
     except FileNotFoundError:
         # It may have been mounted to a different location in the execution environment
         modified = None

src/ansible_navigator/utils/print.py

@@ -112,7 +112,7 @@ def print_to_stdout(
                 content=content,
                 content_view=ContentView.NORMAL,
                 serialization_format=serialization_format,
-            )
+            ),
         )
         output = serialized
     else:

tests/conftest.py

@@ -119,9 +119,9 @@ def locked_directory(tmpdir: str) -> Generator[str, None, None]:
     :param tmpdir: Fixture for temporary directory
     :yields: The temporary directory
     """
-    os.chmod(tmpdir, 0o000)
+    Path(tmpdir).chmod(0o000)
     yield tmpdir
-    os.chmod(tmpdir, 0o777)
+    Path(tmpdir).chmod(0o777)
 
 
 @pytest.fixture(scope="session")
@@ -392,7 +392,7 @@ def pytest_configure(config: pytest.Config) -> None:
     ):
         pytest.exit(
             f"Please remove or empty the ansible config file '{config_file}' "
-            "before testing, as this will likely break the test results."
+            "before testing, as this will likely break the test results.",
         )
 
     # look for ansible-navigator settings file
@@ -417,7 +417,8 @@ def pytest_unconfigure(config: pytest.Config) -> None:
 
 @pytest.fixture()
 def skip_if_already_failed(  # noqa: PT004
-    request: pytest.FixtureRequest, failed: set[str] = set()
+    request: pytest.FixtureRequest,
+    failed: set[str] = set(),
 ) -> Generator[None, None, None]:
     """Fixture that stops parametrized tests running on first failure."""
     key = request.node.name.split("[")[0]

tests/integration/_common.py

@@ -217,10 +217,10 @@ def copytree(
         source_path = os.path.join(src, name)
         destination_path = os.path.join(dst, name)
         try:
-            if symlinks and os.path.islink(source_path):
-                source_link = os.readlink(source_path)
+            if symlinks and Path(source_path).is_symlink():
+                source_link = Path(source_path).readlink()
                 os.symlink(source_link, destination_path)
-            elif os.path.isdir(source_path):
+            elif Path(source_path).is_dir():
                 copytree(
                     source_path,
                     destination_path,

tests/integration/_tmux_session.py

@@ -108,7 +108,9 @@ def _build_tmux_session(self) -> None:
                 break
             except libtmux.exc.LibTmuxException as exc:
                 warnings.warn(
-                    f"tmux session failure #{count}: {exc!s}", RuntimeWarning, stacklevel=2
+                    f"tmux session failure #{count}: {exc!s}",
+                    RuntimeWarning,
+                    stacklevel=2,
                 )
                 if count == tries:
                     raise

tests/integration/actions/builder/base.py

@@ -29,7 +29,8 @@ class BaseClass:
 
     @pytest.fixture(scope="module", name="tmux_session")
     def fixture_tmux_session(
-        self, request: pytest.FixtureRequest
+        self,
+        request: pytest.FixtureRequest,
     ) -> Generator[TmuxSession, None, None]:
         """Generate a tmux fixture for this module.