Convert more code to pathlib

pyproject.toml

@@ -114,7 +114,7 @@ disable = [
 filterwarnings = ["error"]
 
 [tool.ruff]
-select = ["PT"]
+select = ["PT", "PTH"]
 ignore = [
   "E501", # we use black
 

src/ansible_compat/loaders.py

@@ -1,7 +1,7 @@
 """Utilities for loading various files."""
 from __future__ import annotations
 
-import os
+from pathlib import Path
 from typing import Any
 
 import yaml
@@ -11,15 +11,16 @@
 
 def yaml_from_file(filepath: str) -> Any:
     """Return a loaded YAML file."""
-    with open(filepath, encoding="utf-8") as content:
+    path = Path(filepath)
+    with path.open(encoding="utf-8") as content:
         return yaml.load(content, Loader=yaml.FullLoader)
 
 
 def colpath_from_path(filepath: str) -> str | None:
     """Return a FQCN from a path."""
-    galaxy_file = f"{filepath}/galaxy.yml"
-    if os.path.exists(galaxy_file):
-        galaxy = yaml_from_file(galaxy_file)
+    galaxy_file = Path(f"{filepath}/galaxy.yml")
+    if galaxy_file.exists():
+        galaxy = yaml_from_file(str(galaxy_file))
         for k in ("namespace", "name"):
             if k not in galaxy:
                 raise InvalidPrerequisiteError(

src/ansible_compat/prerun.py

@@ -10,12 +10,12 @@ def get_cache_dir(project_dir: str) -> Path:
     # we would use the same key regardless the location of the user home
     # directory or where the project is clones (as long the project folder uses
     # the same name).
-    basename = os.path.basename(os.path.abspath(project_dir)).encode(encoding="utf-8")
+    basename = Path(project_dir).resolve().name.encode(encoding="utf-8")
     # 6 chars of entropy should be enough
     cache_key = hashlib.sha256(basename).hexdigest()[:6]
     cache_dir = (
-        os.getenv("XDG_CACHE_HOME", os.path.expanduser("~/.cache"))
-        + "/ansible-compat/"
-        + cache_key
+        Path(os.getenv("XDG_CACHE_HOME", "~/.cache")).expanduser()
+        / "ansible-compat"
+        / cache_key
     )
-    return Path(cache_dir)
+    return cache_dir

src/ansible_compat/runtime.py

@@ -10,6 +10,7 @@
 import subprocess
 import tempfile
 import warnings
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 import packaging
@@ -87,7 +88,7 @@ def __init__(
         :param environ: Environment dictionary to use, if undefined
                         ``os.environ`` will be copied and used.
         """
-        self.project_dir = project_dir or os.getcwd()
+        self.project_dir = project_dir or str(Path.cwd())
         self.isolated = isolated
         self.max_retries = max_retries
         self.environ = environ or os.environ.copy()
@@ -159,7 +160,7 @@ def _ensure_module_available(self) -> None:
                 # noinspection PyProtectedMember
                 # pylint: disable=protected-access
                 col_path += self.config.collections_paths
-                col_path += os.path.dirname(
+                col_path += os.path.dirname(  # noqa: PTH120
                     os.environ.get(ansible_collections_path(), "."),
                 ).split(":")
                 _AnsibleCollectionFinder(
@@ -278,7 +279,7 @@ def install_collection(
             cpaths.insert(0, str(destination))
         cmd.append(f"{collection}")
 
-        _logger.info("Running from %s : %s", os.getcwd(), " ".join(cmd))
+        _logger.info("Running from %s : %s", Path.cwd(), " ".join(cmd))
         run = self.exec(
             cmd,
             retry=True,
@@ -315,7 +316,7 @@ def install_collection_from_disk(
                 raise AnsibleCommandError(run)
             for archive_file in os.listdir(tmp_dir):
                 self.install_collection(
-                    os.path.join(tmp_dir, archive_file),
+                    str(Path(tmp_dir) / archive_file),
                     destination=destination,
                     force=True,
                 )
@@ -333,7 +334,7 @@ def install_requirements(
         :param retry: retry network operations on failures
         :param offline: bypass installation, may fail if requirements are not met.
         """
-        if not os.path.exists(requirement):
+        if not Path(requirement).exists():
             return
         reqs_yaml = yaml_from_file(requirement)
         if not isinstance(reqs_yaml, (dict, list)):
@@ -432,26 +433,29 @@ def prepare_environment(
         if not install_local:
             return
 
-        if os.path.exists("galaxy.yml"):
+        if Path("galaxy.yml").exists():
             if destination:
                 # while function can return None, that would not break the logic
-                colpath = f"{destination}/ansible_collections/{colpath_from_path(os.getcwd())}"
-                if os.path.islink(colpath):
-                    if os.path.realpath(colpath) == os.getcwd():
+                colpath = Path(
+                    f"{destination}/ansible_collections/{colpath_from_path(str(Path.cwd()))}"
+                )
+                if colpath.is_symlink():
+                    if os.path.realpath(colpath) == Path.cwd():
                         _logger.warning(
                             "Found symlinked collection, skipping its installation.",
                         )
                         return
                     _logger.warning(
                         "Collection is symlinked, but not pointing to %s directory, so we will remove it.",
-                        os.getcwd(),
+                        Path.cwd(),
                     )
-                    os.unlink(colpath)
+                    colpath.unlink()
 
             # molecule scenario within a collection
             self.install_collection_from_disk(".", destination=destination)
-        elif pathlib.Path().resolve().parent.name == "roles" and os.path.exists(
-            "../../galaxy.yml",
+        elif (
+            pathlib.Path().resolve().parent.name == "roles"
+            and Path("../../galaxy.yml").exists()
         ):
             # molecule scenario located within roles/<role-name>/molecule inside
             # a collection
@@ -495,17 +499,15 @@ def require_collection(
             paths.insert(0, f"{self.cache_dir}/collections")  # pylint: disable=E1101
 
         for path in paths:
-            collpath = os.path.expanduser(
-                os.path.join(path, "ansible_collections", ns, coll),
-            )
-            if os.path.exists(collpath):
-                mpath = os.path.join(collpath, "MANIFEST.json")
-                if not os.path.exists(mpath):
+            collpath = pathlib.Path(path) / "ansible_collections" / ns / coll
+            if collpath.exists():
+                mpath = collpath / "MANIFEST.json"
+                if not mpath.exists():
                     msg = f"Found collection at '{collpath}' but missing MANIFEST.json, cannot get info."
                     _logger.fatal(msg)
                     raise InvalidPrerequisiteError(msg)
 
-                with open(mpath, encoding="utf-8") as f:
+                with mpath.open(encoding="utf-8") as f:
                     manifest = json.loads(f.read())
                     found_version = packaging.version.parse(
                         manifest["collection_info"]["version"],
@@ -552,13 +554,14 @@ def _prepare_ansible_paths(self) -> None:
             else [],
         )
 
-        for path_list, path, must_be_present in alterations_list:
-            if not os.path.exists(path):
+        for path_list, path_, must_be_present in alterations_list:
+            path = pathlib.Path(path_)
+            if not path.exists():
                 if must_be_present:
                     continue
-                os.makedirs(path, exist_ok=True)
+                path.mkdir(parents=True, exist_ok=True)
             if path not in path_list:
-                path_list.insert(0, path)
+                path_list.insert(0, str(path))
 
         if library_paths != self.config.DEFAULT_MODULE_PATH:
             self._update_env("ANSIBLE_LIBRARY", library_paths)
@@ -578,7 +581,7 @@ def _get_roles_path(self) -> pathlib.Path:
         if self.cache_dir:
             path = pathlib.Path(f"{self.cache_dir}/roles")
         else:
-            path = pathlib.Path(os.path.expanduser(self.config.default_roles_path[0]))
+            path = pathlib.Path(self.config.default_roles_path[0]).expanduser()
         return path
 
     def _install_galaxy_role(
@@ -603,13 +606,13 @@ def _install_galaxy_role(
         """
         yaml = None
         galaxy_info = {}
-        meta_filename = os.path.join(project_dir, "meta", "main.yml")
+        meta_filename = pathlib.Path(project_dir) / "meta" / "main.yml"
 
-        if not os.path.exists(meta_filename):
+        if not meta_filename.exists():
             if ignore_errors:
                 return
         else:
-            yaml = yaml_from_file(meta_filename)
+            yaml = yaml_from_file(str(meta_filename))
 
         if yaml and "galaxy_info" in yaml:
             galaxy_info = yaml["galaxy_info"]
@@ -639,7 +642,7 @@ def _install_galaxy_role(
         # it appears that is_dir() reports true instead, so we rely on exists().
         target = pathlib.Path(project_dir).absolute()
         if not link_path.exists() or (
-            link_path.is_symlink() and os.readlink(link_path) != str(target)
+            link_path.is_symlink() and link_path.readlink() != target
         ):
             # must call unlink before checking exists because a broken
             # link reports as not existing and we want to repair it