fix: address SonarCloud new code violations (#5116)

Author: sudhirverma

.config/dictionary.txt

@@ -6,6 +6,7 @@ FQCNs
 Fimport
 LIBYAML
 LINEINFILE
+LINTABLE
 Lineinfile
 Lintable
 MYSQLD

src/ansiblelint/cli.py

@@ -62,6 +62,30 @@ def expand_to_normalized_paths(
         config[paths_var] = normalized_paths
 
 
+def _fatal_config_error(msg: str) -> None:
+    """Log a fatal configuration error and terminate the process."""
+    if any(
+        isinstance(h, logging.FileHandler)
+        and h.baseFilename != os.path.abspath(os.devnull)
+        for h in logging.root.handlers
+    ):
+        _logger.error(msg)
+    console_stderr.print(f"[error]{msg}[/]")
+    sys.exit(RC.INVALID_CONFIG)
+
+
+def _resolve_config_project_dir(config: dict[str, Any], config_path: str) -> None:
+    """Resolve a relative project_dir in config against the config file location."""
+    if (
+        "project_dir" in config
+        and config["project_dir"]
+        and config["project_dir"][0] not in ("/", "~")
+    ):
+        config["project_dir"] = (
+            (Path(config_path).parent / config["project_dir"]).resolve().as_posix()
+        )
+
+
 def load_config(
     config_file: str | None = None, project_path: str | None = None
 ) -> tuple[dict[Any, Any], str | None]:
@@ -75,15 +99,7 @@ def load_config(
     if config_file:
         config_path = os.path.abspath(config_file)
         if not os.path.exists(config_path):
-            msg = f"Config file not found '{config_path}'"
-            if any(
-                isinstance(h, logging.FileHandler)
-                and h.baseFilename != os.path.abspath(os.devnull)
-                for h in logging.root.handlers
-            ):
-                _logger.error(msg)
-            console_stderr.print(f"[error]{msg}[/]")
-            sys.exit(RC.INVALID_CONFIG)
+            _fatal_config_error(f"Config file not found '{config_path}'")
     config_path = config_path or get_config_path(None, project_path=project_path)
     if not config_path or not os.path.exists(config_path):
         # a missing default config file should not trigger an error
@@ -98,15 +114,7 @@ def load_config(
     )
 
     for error in validate_file_schema(config_lintable):
-        msg = f"Invalid configuration file {config_path}. {error}"
-        if any(
-            isinstance(h, logging.FileHandler)
-            and h.baseFilename != os.path.abspath(os.devnull)
-            for h in logging.root.handlers
-        ):
-            _logger.error(msg)
-        console_stderr.print(f"[error]{msg}[/]")
-        sys.exit(RC.INVALID_CONFIG)
+        _fatal_config_error(f"Invalid configuration file {config_path}. {error}")
 
     config = clean_json(config_lintable.data)
     if not isinstance(config, dict):
@@ -116,14 +124,7 @@ def load_config(
     config_dir = os.path.dirname(config_path)
     expand_to_normalized_paths(config, config_dir)
 
-    if (
-        "project_dir" in config
-        and config["project_dir"]
-        and config["project_dir"][0] not in ("/", "~")
-    ):
-        config["project_dir"] = (
-            (Path(config_path).parent / config["project_dir"]).resolve().as_posix()
-        )
+    _resolve_config_project_dir(config, config_path)
 
     return config, config_path
 

src/ansiblelint/config.py

@@ -14,7 +14,7 @@
 from importlib.metadata import PackageNotFoundError, distribution, version
 from pathlib import Path
 from typing import Any
-from urllib.error import HTTPError, URLError
+from urllib.error import URLError
 
 from packaging.version import Version
 
@@ -297,6 +297,57 @@ def get_deps_versions() -> dict[str, Version | None]:
     return result
 
 
+def _version_cache_needs_refresh(cache_file: str) -> bool:
+    if not os.path.exists(cache_file):
+        return True
+    age = time.time() - os.path.getmtime(cache_file)
+    return age >= 24 * 60 * 60
+
+
+def _load_version_cache(cache_file: str) -> dict[str, Any]:
+    with open(cache_file, encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _fetch_latest_release(cache_file: str) -> dict[str, Any]:
+    release_url = "https://api.github.com/repos/ansible/ansible-lint/releases/latest"
+    if not release_url.startswith("https://"):  # pragma: no cover (ruff compatibility)
+        msg = "release_url must start with https://"
+        raise ValueError(msg)
+    try:
+        with urllib.request.urlopen(release_url) as url:
+            data = json.load(url)
+            with open(cache_file, "w", encoding="utf-8") as f:
+                json.dump(data, f)
+    except (URLError, HTTPException) as exc:  # pragma: no cover
+        _logger.debug(
+            "Unable to fetch latest version from %s due to: %s",
+            release_url,
+            exc,
+        )
+        return {}
+    else:
+        return data
+
+
+def _format_version_upgrade_message(
+    current_version: Version,
+    data: dict[str, Any],
+    pip: str,
+) -> str:
+    if not data:
+        return ""
+    html_url = data["html_url"]
+    new_version = Version(data["tag_name"][1:])  # removing v prefix from tag
+
+    if current_version > new_version:
+        return "[dim]You are using a pre-release version of ansible-lint.[/]"
+    if current_version < new_version:
+        msg = f"""[warning]A new release of ansible-lint is available: [warning]{current_version}[/] → [success][link={html_url}]{new_version}[/link][/][/]"""
+        return f"{msg} Upgrade by running: [info]{pip}[/]"
+    return ""
+
+
 def get_version_warning() -> str:
     """Display warning if current version is outdated."""
     # 0.1dev1 is special fallback version
@@ -308,51 +359,20 @@ def get_version_warning() -> str:
     if not pip:
         return ""
 
-    msg = ""
-    data = {}
     current_version = Version(__version__)
 
     if not os.path.exists(CACHE_DIR):  # pragma: no cover
         os.makedirs(CACHE_DIR)
     cache_file = f"{CACHE_DIR}/latest.json"
-    refresh = True
+    refresh = _version_cache_needs_refresh(cache_file)
+    data: dict[str, Any] = {}
     if os.path.exists(cache_file):
-        age = time.time() - os.path.getmtime(cache_file)
-        if age < 24 * 60 * 60:
-            refresh = False
-        with open(cache_file, encoding="utf-8") as f:
-            data = json.load(f)
+        data = _load_version_cache(cache_file)
 
     if not options.offline and (refresh or not data):
-        release_url = (
-            "https://api.github.com/repos/ansible/ansible-lint/releases/latest"
-        )
-        if not release_url.startswith(
-            "https://"
-        ):  # pragma: no cover (ruff compatibility)
-            msg = "release_url must start with https://"
-            raise ValueError(msg)
-        try:
-            with urllib.request.urlopen(release_url) as url:
-                data = json.load(url)
-                with open(cache_file, "w", encoding="utf-8") as f:
-                    json.dump(data, f)
-        except (URLError, HTTPError, HTTPException) as exc:  # pragma: no cover
-            _logger.debug(
-                "Unable to fetch latest version from %s due to: %s",
-                release_url,
-                exc,
-            )
+        fetched = _fetch_latest_release(cache_file)
+        if not fetched:
             return ""
+        data = fetched
 
-    if data:
-        html_url = data["html_url"]
-        new_version = Version(data["tag_name"][1:])  # removing v prefix from tag
-
-        if current_version > new_version:
-            msg = "[dim]You are using a pre-release version of ansible-lint.[/]"
-        elif current_version < new_version:
-            msg = f"""[warning]A new release of ansible-lint is available: [warning]{current_version}[/] → [success][link={html_url}]{new_version}[/link][/][/]"""
-            msg += f" Upgrade by running: [info]{pip}[/]"
-
-    return msg
+    return _format_version_upgrade_message(current_version, data, pip)

src/ansiblelint/file_utils.py

@@ -8,6 +8,7 @@
 import sys
 from collections import defaultdict
 from contextlib import contextmanager
+from functools import lru_cache
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING, Any, cast
@@ -149,49 +150,58 @@ def expand_paths_vars(paths: list[str]) -> list[str]:
     return paths
 
 
-def kind_from_path(path: Path, *, base: bool = False) -> FileType:
-    """Determine the file kind based on its name.
+@lru_cache(maxsize=128)
+def _get_project_root_for_dir(dir_path: str) -> Path:
+    """Cache project-root discovery per directory to avoid repeated FS walks."""
+    project_root, _ = find_project_root([dir_path])
+    return project_root
 
-    When called with base=True, it will return the base file type instead
-    of the explicit one. That is expected to return 'yaml' for any yaml files.
-    """
+
+def _resolve_kind_glob_path(path: Path) -> wcmatch.pathlib.PurePath:
+    """Return a PurePath usable for glob matching against kind patterns."""
     # We attempt to use a relative path to the project root for glob matching.
     # This prevents parent directory names (like 'tasks') from triggering
     # false positives in kind discovery. See #4763.
     try:
-        project_root, _ = find_project_root([str(path)])
+        # Project root is shared by siblings; cache by directory to cut FS walks.
+        lookup_path = path if path.is_dir() else path.parent
+        project_root = _get_project_root_for_dir(str(lookup_path.resolve()))
         # .resolve() ensures we handle symlinks and double-dots correctly
         rel_path = path.resolve().relative_to(project_root.resolve())
-        pathex = wcmatch.pathlib.PurePath(str(rel_path))
+        return wcmatch.pathlib.PurePath(str(rel_path))
     except (ValueError, RuntimeError):
         # Fallback to absolute if the file is outside the project root or can't be found
-        pathex = wcmatch.pathlib.PurePath(str(path.absolute().resolve()))
-    kinds = options.kinds if not base else BASE_KINDS
+        return wcmatch.pathlib.PurePath(str(path.absolute().resolve()))
+
+
+def _match_kind_from_globs(
+    pathex: wcmatch.pathlib.PurePath,
+    kinds: list[dict[str, str]],
+    path: Path,
+) -> FileType | None:
+    """Return the first kind whose glob pattern matches path, if any."""
     for entry in kinds:
         for k, v in entry.items():
-            if pathex.globmatch(
+            if not pathex.globmatch(
                 v,
                 flags=(
                     wcmatch.pathlib.GLOBSTAR
                     | wcmatch.pathlib.BRACE
                     | wcmatch.pathlib.DOTGLOB
                 ),
             ):
-                matched_kind = str(k)
-                # Namespace folders under roles/ can match **/roles/*/ without
-                # being role roots themselves. See #5079.
-                if (
-                    matched_kind == "role"
-                    and path.is_dir()
-                    and not _has_role_subdirs(path)
-                ):
-                    continue
-                return matched_kind  # type: ignore[return-value]
+                continue
+            matched_kind = str(k)
+            # Namespace folders under roles/ can match **/roles/*/ without
+            # being role roots themselves. See #5079.
+            if matched_kind == "role" and path.is_dir() and not _has_role_subdirs(path):
+                continue
+            return matched_kind  # type: ignore[return-value]
+    return None
 
-    if base:
-        # Unknown base file type is default
-        return ""
 
+def _fallback_kind_from_path(path: Path) -> FileType:
+    """Determine a fallback kind for a path not matched by any glob pattern."""
     if path.is_dir() and _has_role_subdirs(path):
         return "role"
 
@@ -213,6 +223,25 @@ def kind_from_path(path: Path, *, base: bool = False) -> FileType:
     return ""
 
 
+def kind_from_path(path: Path, *, base: bool = False) -> FileType:
+    """Determine the file kind based on its name.
+
+    When called with base=True, it will return the base file type instead
+    of the explicit one. That is expected to return 'yaml' for any yaml files.
+    """
+    pathex = _resolve_kind_glob_path(path)
+    kinds = options.kinds if not base else BASE_KINDS
+    matched_kind = _match_kind_from_globs(pathex, kinds, path)
+    if matched_kind is not None:
+        return matched_kind
+
+    if base:
+        # Unknown base file type is default
+        return ""
+
+    return _fallback_kind_from_path(path)
+
+
 # pylint: disable=too-many-instance-attributes
 class Lintable:
     """Defines a file/folder that can be linted.

src/ansiblelint/loaders.py

@@ -67,10 +67,8 @@ def get_ignore_rule(rule: str, qualifiers: str) -> IgnoreRule:
     return IgnoreRule(rule, frozenset(s))
 
 
-def load_ignore_txt(filepath: Path | None = None) -> dict[str, set[IgnoreRule]]:
-    """Return a list of rules to ignore."""
-    result = defaultdict(set)
-
+def _resolve_ignore_file(filepath: Path | None) -> str | None:
+    """Return the path to the ignore file to use, if any."""
     ignore_file = None
 
     if filepath:
@@ -83,21 +81,38 @@ def load_ignore_txt(filepath: Path | None = None) -> dict[str, set[IgnoreRule]]:
     elif os.path.isfile(IGNORE_FILE.alternative):
         ignore_file = IGNORE_FILE.alternative
 
+    return ignore_file
+
+
+def _parse_ignore_file(
+    ignore_file: str,
+    result: defaultdict[str, set[IgnoreRule]],
+) -> None:
+    """Parse an ignore file, populating result in place."""
+    with open(ignore_file, encoding="utf-8") as ignore_file_h:
+        _logger.debug("Loading ignores from '%s'", ignore_file)
+        for line in ignore_file_h:
+            entry = line.split("#")[0].rstrip()
+            if not entry:
+                continue
+            try:
+                fields = entry.split()
+                path = fields[0]
+                rule = fields[1]
+                qualifiers = fields[2] if len(fields) == 3 else ""
+                result[path].add(get_ignore_rule(rule, qualifiers))
+            except ValueError as exc:  # pragma: no cover
+                msg = f"Unable to parse line '{line}' from {ignore_file} file."
+                raise RuntimeError(msg) from exc
+
+
+def load_ignore_txt(filepath: Path | None = None) -> dict[str, set[IgnoreRule]]:
+    """Return a list of rules to ignore."""
+    result: defaultdict[str, set[IgnoreRule]] = defaultdict(set)
+
+    ignore_file = _resolve_ignore_file(filepath)
     if ignore_file:
-        with open(ignore_file, encoding="utf-8") as ignore_file_h:
-            _logger.debug("Loading ignores from '%s'", ignore_file)
-            for line in ignore_file_h:
-                entry = line.split("#")[0].rstrip()
-                if entry:
-                    try:
-                        fields = entry.split()
-                        path = fields[0]
-                        rule = fields[1]
-                        qualifiers = fields[2] if len(fields) == 3 else ""
-                        result[path].add(get_ignore_rule(rule, qualifiers))
-                    except ValueError as exc:  # pragma: no cover
-                        msg = f"Unable to parse line '{line}' from {ignore_file} file."
-                        raise RuntimeError(msg) from exc
+        _parse_ignore_file(ignore_file, result)
     return result