More pathlib conversion (#3396)

pyproject.toml

@@ -257,7 +257,6 @@ known-first-party = ["ansiblelint"]
 "src/ansiblelint/rules/*.py" = ["S"]
 "src/ansiblelint/testing/*.py" = ["S"]
 # Temporary disabled until we fix them:
-"src/ansiblelint/{testing,schemas,rules}/*.py" = ["PTH"]
 "src/ansiblelint/{utils,file_utils,runner,loaders,constants,config,cli,_mockings,__main__}.py" = [
   "PTH",
 ]

src/ansiblelint/cli.py

@@ -572,8 +572,8 @@ def get_config(arguments: list[str]) -> Options:
     config = merge_config(file_config, options)
 
     options.rulesdirs = get_rules_dirs(
-        [str(r) for r in options.rulesdir],
-        options.use_default_rules,
+        options.rulesdir,
+        use_default=options.use_default_rules,
     )
 
     if not options.project_dir:
@@ -603,7 +603,7 @@ def print_help(file: Any = sys.stdout) -> None:
     get_cli_parser().print_help(file=file)
 
 
-def get_rules_dirs(rulesdir: list[str], use_default: bool = True) -> list[str]:
+def get_rules_dirs(rulesdir: list[Path], *, use_default: bool = True) -> list[Path]:
     """Return a list of rules dirs."""
     default_ruledirs = [DEFAULT_RULESDIR]
     default_custom_rulesdir = os.environ.get(
@@ -616,7 +616,11 @@ def get_rules_dirs(rulesdir: list[str], use_default: bool = True) -> list[str]:
         if x.is_dir() and (x / "__init__.py").exists()
     )
 
+    result: list[Any] = []
     if use_default:
-        return rulesdir + custom_ruledirs + default_ruledirs
-
-    return rulesdir or custom_ruledirs + default_ruledirs
+        result = rulesdir + custom_ruledirs + default_ruledirs
+    elif rulesdir:
+        result = rulesdir
+    else:
+        result = custom_ruledirs + default_ruledirs
+    return [Path(p) for p in result]

src/ansiblelint/config.py

@@ -116,7 +116,7 @@ class Options:  # pylint: disable=too-many-instance-attributes,too-few-public-me
     write_list: list[str] = field(default_factory=list)
     parseable: bool = False
     quiet: bool = False
-    rulesdirs: list[str] = field(default_factory=list)
+    rulesdirs: list[Path] = field(default_factory=list)
     skip_list: list[str] = field(default_factory=list)
     tags: list[str] = field(default_factory=list)
     verbosity: int = 0

src/ansiblelint/constants.py

@@ -1,10 +1,9 @@
 """Constants used by AnsibleLint."""
-import os.path
 from enum import Enum
 from pathlib import Path
 from typing import Literal
 
-DEFAULT_RULESDIR = os.path.join(os.path.dirname(__file__), "rules")
+DEFAULT_RULESDIR = Path(__file__).parent / "rules"
 CUSTOM_RULESDIR_ENVVAR = "ANSIBLE_LINT_CUSTOM_RULESDIR"
 RULE_DOC_URL = "https://ansible-lint.readthedocs.io/rules/"
 

src/ansiblelint/rules/command_instead_of_module.py

@@ -20,8 +20,8 @@
 # THE SOFTWARE.
 from __future__ import annotations
 
-import os
 import sys
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from ansiblelint.rules import AnsibleLintRule
@@ -87,7 +87,7 @@ def matchtask(
         if not first_cmd_arg:
             return False
 
-        executable = os.path.basename(first_cmd_arg)
+        executable = Path(first_cmd_arg).name
 
         if (
             second_cmd_arg

src/ansiblelint/rules/galaxy.py

@@ -1,7 +1,6 @@
 """Implementation of GalaxyRule."""
 from __future__ import annotations
 
-import os
 import sys
 from functools import total_ordering
 from typing import TYPE_CHECKING, Any
@@ -45,16 +44,16 @@ def matchplay(self, file: Lintable, data: dict[str, Any]) -> list[MatchError]:
 
         results = []
 
-        base_path = os.path.split(str(file.abspath))[0]
+        base_path = file.path.parent.resolve()
         changelog_found = 0
         changelog_paths = [
-            os.path.join(base_path, "changelogs", "changelog.yaml"),
-            os.path.join(base_path, "CHANGELOG.rst"),
-            os.path.join(base_path, "CHANGELOG.md"),
+            base_path / "changelogs" / "changelog.yaml",
+            base_path / "CHANGELOG.rst",
+            base_path / "CHANGELOG.md",
         ]
 
         for path in changelog_paths:
-            if os.path.isfile(path):
+            if path.is_file():
                 changelog_found = 1
 
         galaxy_tag_list = data.get("tags", None)
@@ -109,7 +108,7 @@ def matchplay(self, file: Lintable, data: dict[str, Any]) -> list[MatchError]:
                 ),
             )
 
-        if not os.path.isfile(os.path.join(base_path, "meta/runtime.yml")):
+        if not (base_path / "meta" / "runtime.yml").is_file():
             results.append(
                 self.create_matcherror(
                     message="meta/runtime.yml file not found.",

src/ansiblelint/rules/meta_no_tags.py

@@ -3,6 +3,7 @@
 
 import re
 import sys
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from ansiblelint.rules import AnsibleLintRule
@@ -104,7 +105,9 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
     )
     def test_valid_tag_rule(rule_runner: RunFromText) -> None:
         """Test rule matches."""
-        results = rule_runner.run("examples/roles/meta_no_tags_valid/meta/main.yml")
+        results = rule_runner.run(
+            Path("examples/roles/meta_no_tags_valid/meta/main.yml"),
+        )
         assert "Use 'galaxy_tags' rather than 'categories'" in str(results), results
         assert "Expected 'categories' to be a list" in str(results)
         assert "invalid: 'my s q l'" in str(results)

src/ansiblelint/rules/playbook_extension.py

@@ -3,7 +3,6 @@
 # Copyright (c) 2018, Ansible Project
 from __future__ import annotations
 
-import os
 import sys
 from typing import TYPE_CHECKING
 
@@ -30,8 +29,8 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
         if file.kind != "playbook":
             return result
         path = str(file.path)
-        ext = os.path.splitext(path)
-        if ext[1] not in [".yml", ".yaml"] and path not in self.done:
+        ext = file.path.suffix
+        if ext not in [".yml", ".yaml"] and path not in self.done:
             self.done.append(path)
             result.append(self.create_matcherror(filename=file))
         return result

src/ansiblelint/rules/risky_file_permissions.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import sys
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from ansiblelint.rules import AnsibleLintRule
@@ -154,7 +155,7 @@ def test_risky_file_permissions(file: str, expected: int) -> None:
         collection = RulesCollection()
         collection.register(MissingFilePermissionsRule())
         runner = RunFromText(collection)
-        results = runner.run(file)
+        results = runner.run(Path(file))
         assert len(results) == expected
         for result in results:
             assert result.tag == "risky-file-permissions"

src/ansiblelint/rules/sanity.py

@@ -61,7 +61,7 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
         if file.kind != "sanity-ignore-file":
             return []
 
-        with open(file.abspath, encoding="utf-8") as ignore_file:
+        with file.path.open(encoding="utf-8") as ignore_file:
             entries = ignore_file.read().splitlines()
 
             ignores = self.allowed_ignores_all

src/ansiblelint/rules/var_naming.py

@@ -17,6 +17,8 @@
 from ansiblelint.utils import parse_yaml_from_file
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from ansiblelint.errors import MatchError
 
 # Should raise var-naming at line [2, 6].
@@ -212,9 +214,12 @@ def test_invalid_var_name_playbook(file: str, expected: int) -> None:
         (VariableNamingRule,),
         indirect=["rule_runner"],
     )
-    def test_invalid_var_name_varsfile(rule_runner: RunFromText) -> None:
+    def test_invalid_var_name_varsfile(
+        rule_runner: RunFromText,
+        tmp_path: Path,
+    ) -> None:
         """Test rule matches."""
-        results = rule_runner.run_role_defaults_main(FAIL_VARS)
+        results = rule_runner.run_role_defaults_main(FAIL_VARS, tmp_path=tmp_path)
         assert len(results) == 2
         for result in results:
             assert result.rule.id == VariableNamingRule.id

src/ansiblelint/schemas/__main__.py

@@ -16,7 +16,7 @@
 # Maps kinds to JSON schemas
 # See https://www.schemastore.org/json/
 store_file = Path(f"{__file__}/../__store__.json").resolve()
-with open(store_file, encoding="utf-8") as json_file:
+with store_file.open(encoding="utf-8") as json_file:
     JSON_SCHEMAS = json.load(json_file)
 
 
@@ -33,8 +33,8 @@ def __missing__(self, key: str) -> Any:
 @cache
 def get_schema(kind: str) -> Any:
     """Return the schema for the given kind."""
-    schema_file = os.path.dirname(__file__) + "/" + kind + ".json"
-    with open(schema_file, encoding="utf-8") as f:
+    schema_file = Path(__file__).parent / f"{kind}.json"
+    with schema_file.open(encoding="utf-8") as f:
         return json.load(f)
 
 
@@ -66,7 +66,7 @@ def refresh_schemas(min_age_seconds: int = 3600 * 24) -> int:  # noqa: C901
         if "#" in url:
             msg = f"Schema URLs cannot contain # due to python-jsonschema limitation: {url}"
             raise RuntimeError(msg)
-        path = Path(f"{os.path.relpath(os.path.dirname(__file__))}/{kind}.json")
+        path = Path(__file__).parent.resolve() / f"{kind}.json"
         _logger.debug("Refreshing %s schema ...", kind)
         request = Request(url)
         etag = data.get("etag", "")
@@ -80,7 +80,7 @@ def refresh_schemas(min_age_seconds: int = 3600 * 24) -> int:  # noqa: C901
                     if etag != data.get("etag", ""):
                         JSON_SCHEMAS[kind]["etag"] = etag
                         changed += 1
-                    with open(f"{path}", "w", encoding="utf-8") as f_out:
+                    with path.open("w", encoding="utf-8") as f_out:
                         _logger.info("Schema %s was updated", kind)
                         f_out.write(content)
                         f_out.write("\n")  # prettier/editors
@@ -100,7 +100,7 @@ def refresh_schemas(min_age_seconds: int = 3600 * 24) -> int:  # noqa: C901
             _logger.debug("Skipped schema refresh due to unexpected exception: %s", exc)
             break
     if changed:  # pragma: no cover
-        with open(store_file, "w", encoding="utf-8") as f_out:
+        with store_file.open("w", encoding="utf-8") as f_out:
             # formatting should match our .prettierrc.yaml
             json.dump(JSON_SCHEMAS, f_out, indent=2, sort_keys=True)
             f_out.write("\n")  # prettier and editors in general

src/ansiblelint/schemas/__store__.json

@@ -4,7 +4,7 @@
     "url": "https://raw.githubusercontent.com/ansible/ansible-lint/main/src/ansiblelint/schemas/ansible-lint-config.json"
   },
   "ansible-navigator-config": {
-    "etag": "59cfdb89d07c929dbd8a0127fbc824faffdb0a094052d36935266770fcbd4bca",
+    "etag": "9095eec00b504fee85c5c7f445839e9823d25a15cde28270007dfd6cda575e81",
     "url": "https://raw.githubusercontent.com/ansible/ansible-navigator/main/src/ansible_navigator/data/ansible-navigator.json"
   },
   "changelog": {

src/ansiblelint/schemas/ansible-navigator-config.json

@@ -520,7 +520,7 @@
     "required": [
         "ansible-navigator"
     ],
-    "title": "ansible-navigator settings v3.2",
+    "title": "ansible-navigator settings v3",
     "type": "object",
-    "version": "3.2"
+    "version": "3"
 }

src/ansiblelint/testing/__init__.py

@@ -6,13 +6,12 @@
 import subprocess
 import sys
 import tempfile
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from ansiblelint.app import get_app
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     # https://github.com/PyCQA/pylint/issues/3240
     # pylint: disable=unsubscriptable-object
     CompletedProcess = subprocess.CompletedProcess[Any]
@@ -39,11 +38,11 @@ def __init__(self, collection: RulesCollection) -> None:
 
         self.collection = collection
 
-    def _call_runner(self, path: str) -> list[MatchError]:
+    def _call_runner(self, path: Path) -> list[MatchError]:
         runner = Runner(path, rules=self.collection)
         return runner.run()
 
-    def run(self, filename: str) -> list[MatchError]:
+    def run(self, filename: Path) -> list[MatchError]:
         """Lints received filename."""
         return self._call_runner(filename)
 
@@ -56,39 +55,51 @@ def run_playbook(
         with tempfile.NamedTemporaryFile(mode="w", suffix=".yml", prefix=prefix) as fh:
             fh.write(playbook_text)
             fh.flush()
-            results = self._call_runner(fh.name)
+            results = self._call_runner(Path(fh.name))
         return results
 
-    def run_role_tasks_main(self, tasks_main_text: str) -> list[MatchError]:
+    def run_role_tasks_main(
+        self,
+        tasks_main_text: str,
+        tmp_path: Path,
+    ) -> list[MatchError]:
         """Lints received text as tasks."""
-        role_path = tempfile.mkdtemp(prefix="role_")
-        tasks_path = os.path.join(role_path, "tasks")
-        os.makedirs(tasks_path)
-        with open(os.path.join(tasks_path, "main.yml"), "w", encoding="utf-8") as fh:
+        role_path = tmp_path
+        tasks_path = role_path / "tasks"
+        tasks_path.mkdir(parents=True, exist_ok=True)
+        with (tasks_path / "main.yml").open("w", encoding="utf-8") as fh:
             fh.write(tasks_main_text)
             fh.flush()
         results = self._call_runner(role_path)
         shutil.rmtree(role_path)
         return results
 
-    def run_role_meta_main(self, meta_main_text: str) -> list[MatchError]:
+    def run_role_meta_main(
+        self,
+        meta_main_text: str,
+        temp_path: Path,
+    ) -> list[MatchError]:
         """Lints received text as meta."""
-        role_path = tempfile.mkdtemp(prefix="role_")
-        meta_path = os.path.join(role_path, "meta")
-        os.makedirs(meta_path)
-        with open(os.path.join(meta_path, "main.yml"), "w", encoding="utf-8") as fh:
+        role_path = temp_path
+        meta_path = role_path / "meta"
+        meta_path.mkdir(parents=True, exist_ok=True)
+        with (meta_path / "main.yml").open("w", encoding="utf-8") as fh:
             fh.write(meta_main_text)
             fh.flush()
         results = self._call_runner(role_path)
         shutil.rmtree(role_path)
         return results
 
-    def run_role_defaults_main(self, defaults_main_text: str) -> list[MatchError]:
+    def run_role_defaults_main(
+        self,
+        defaults_main_text: str,
+        tmp_path: Path,
+    ) -> list[MatchError]:
         """Lints received text as vars file in defaults."""
-        role_path = tempfile.mkdtemp(prefix="role_")
-        defaults_path = os.path.join(role_path, "defaults")
-        os.makedirs(defaults_path)
-        with open(os.path.join(defaults_path, "main.yml"), "w", encoding="utf-8") as fh:
+        role_path = tmp_path
+        defaults_path = role_path / "defaults"
+        defaults_path.mkdir(parents=True, exist_ok=True)
+        with (defaults_path / "main.yml").open("w", encoding="utf-8") as fh:
             fh.write(defaults_main_text)
             fh.flush()
         results = self._call_runner(role_path)

src/ansiblelint/testing/fixtures.py

@@ -8,7 +8,6 @@
 from __future__ import annotations
 
 import copy
-import os
 from typing import TYPE_CHECKING
 
 import pytest
@@ -27,7 +26,7 @@
 @pytest.fixture(name="default_rules_collection")
 def fixture_default_rules_collection() -> RulesCollection:
     """Return default rule collection."""
-    assert os.path.isdir(DEFAULT_RULESDIR)
+    assert DEFAULT_RULESDIR.is_dir()
     # For testing we want to manually enable opt-in rules
     options.enable_list = ["no-same-owner"]
     return RulesCollection(rulesdirs=[DEFAULT_RULESDIR], options=options)