file_utils.py

"""Utility functions related to file operations."""
from __future__ import annotations

import copy
import logging
import os
import sys
from collections import defaultdict
from contextlib import contextmanager
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING, Any, cast

import pathspec
import wcmatch.pathlib
import wcmatch.wcmatch
from yaml.error import YAMLError

from ansiblelint.config import BASE_KINDS, Options, options
from ansiblelint.constants import CONFIG_FILENAMES, FileType, States

if TYPE_CHECKING:
    from collections.abc import Iterator, Sequence


_logger = logging.getLogger(__package__)


def abspath(path: str, base_dir: str) -> str:
    """Make relative path absolute relative to given directory.

    path (str): the path to make absolute
    base_dir (str): the directory from which make relative paths absolute.
    """
    if not os.path.isabs(path):
        # Don't use abspath as it assumes path is relative to cwd.
        # We want it relative to base_dir.
        path = os.path.join(base_dir, path)

    return os.path.normpath(path)


def normpath(path: str | Path) -> str:
    """Normalize a path in order to provide a more consistent output.

    Currently it generates a relative path but in the future we may want to
    make this user configurable.
    """
    # prevent possible ValueError with relpath(), when input is an empty string
    if not path:
        path = "."
    # conversion to string in order to allow receiving non string objects
    relpath = os.path.relpath(str(path))
    path_absolute = os.path.abspath(str(path))
    if path_absolute.startswith(os.getcwd()):
        return relpath
    if path_absolute.startswith(os.path.expanduser("~")):
        return path_absolute.replace(os.path.expanduser("~"), "~")
    # we avoid returning relative paths that end-up at root level
    if path_absolute in relpath:
        return path_absolute
    if relpath.startswith("../"):
        return path_absolute
    return relpath


# That is needed for compatibility with py38, later was added to Path class
def is_relative_to(path: Path, *other: Any) -> bool:
    """Return True if the path is relative to another path or False."""
    try:
        path.resolve().absolute().relative_to(*other)
        return True
    except ValueError:
        return False


def normpath_path(path: str | Path) -> Path:
    """Normalize a path in order to provide a more consistent output.

    - Any symlinks are resolved.
    - Any paths outside the CWD are resolved to their absolute path.
    - Any absolute path within current user home directory is compressed to
      make use of '~', so it is easier to read and more portable.
    """
    if not isinstance(path, Path):
        path = Path(path)

    is_relative = is_relative_to(path, path.cwd())
    path = path.resolve()
    if is_relative:
        path = path.relative_to(path.cwd())

    # Compress any absolute path within current user home directory
    if path.is_absolute():
        home = Path.home()
        if is_relative_to(path, home):
            path = Path("~") / path.relative_to(home)

    return path


@contextmanager
def cwd(path: Path) -> Iterator[None]:
    """Context manager for temporary changing current working directory."""
    old_pwd = Path.cwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old_pwd)


def expand_path_vars(path: str) -> str:
    """Expand the environment or ~ variables in a path string."""
    # It may be possible for function to be called with a Path object
    path = str(path).strip()
    path = os.path.expanduser(path)
    path = os.path.expandvars(path)
    return path


def expand_paths_vars(paths: list[str]) -> list[str]:
    """Expand the environment or ~ variables in a list."""
    paths = [expand_path_vars(p) for p in paths]
    return paths


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.
    """
    # pathlib.Path.match patterns are very limited, they do not support *a*.yml
    # glob.glob supports **/foo.yml but not multiple extensions
    pathex = wcmatch.pathlib.PurePath(str(path.absolute().resolve()))
    kinds = options.kinds if not base else BASE_KINDS
    for entry in kinds:
        for k, v in entry.items():
            if pathex.globmatch(
                v,
                flags=(
                    wcmatch.pathlib.GLOBSTAR
                    | wcmatch.pathlib.BRACE
                    | wcmatch.pathlib.DOTGLOB
                ),
            ):
                return str(k)  # type: ignore[return-value]

    if base:
        # Unknown base file type is default
        return ""

    if path.is_dir():
        known_role_subfolders = ("tasks", "meta", "vars", "defaults", "handlers")
        for filename in known_role_subfolders:
            if (path / filename).is_dir():
                return "role"
        _logger.debug(
            "Folder `%s` does not look like a role due to missing any of the common subfolders such: %s.",
            path,
            ", ".join(known_role_subfolders),
        )

    if str(path) == "/dev/stdin":
        return "playbook"

    # Unknown file types report a empty string (evaluated as False)
    return ""


# pylint: disable=too-many-instance-attributes
class Lintable:
    """Defines a file/folder that can be linted.

    Providing file content when creating the object allow creation of in-memory
    instances that do not need files to be present on disk.

    When symlinks are given, they will always be resolved to their target.
    """

    # pylint: disable=too-many-arguments
    def __init__(
        self,
        name: str | Path,
        content: str | None = None,
        kind: FileType | None = None,
        base_kind: str = "",
        parent: Lintable | None = None,
    ):
        """Create a Lintable instance."""
        self.dir: str = ""
        self.kind: FileType | None = None
        self.stop_processing = False  # Set to stop other rules from running
        self.state: Any = States.NOT_LOADED
        self.line_skips: dict[int, set[str]] = defaultdict(set)
        self.exc: Exception | None = None  # Stores data loading exceptions
        self.parent = parent
        self.explicit = False  # Indicates if the file was explicitly provided or was indirectly included.

        if isinstance(name, str):
            name = Path(name)
        is_relative = is_relative_to(name, str(name.cwd()))
        name = name.resolve()
        if is_relative:
            name = name.relative_to(name.cwd())
        name = normpath_path(name)
        # we need to be sure that we expanduser() because otherwise a simple
        # test like .path.exists() will return unexpected results.
        self.path = name.expanduser()
        # Filename is effective file on disk, for stdin is a namedtempfile
        self.name = self.filename = str(name)

        self._content = self._original_content = content
        self.updated = False

        # if the lintable is part of a role, we save role folder name
        self.role = ""
        parts = self.path.parent.parts
        if "roles" in parts:
            role = self.path
            while role.parent.name != "roles" and role.name:
                role = role.parent
            if role.exists():
                self.role = role.name

        if str(self.path) in ["/dev/stdin", "-"]:
            # pylint: disable=consider-using-with
            self.file = NamedTemporaryFile(mode="w+", suffix="playbook.yml")
            self.filename = self.file.name
            self._content = sys.stdin.read()
            self.file.write(self._content)
            self.file.flush()
            self.path = Path(self.file.name)
            self.name = "stdin"
            self.kind = "playbook"
            self.dir = "/"
        else:
            self.kind = kind or kind_from_path(self.path)
        # We store absolute directory in dir
        if not self.dir:
            if self.kind == "role":
                self.dir = str(self.path.resolve())
            else:
                self.dir = str(self.path.parent.resolve())

        # determine base file kind (yaml, xml, ini, ...)
        self.base_kind = base_kind or kind_from_path(self.path, base=True)
        self.abspath = self.path.expanduser().absolute()

        if self.kind == "tasks":
            self.parent = _guess_parent(self)

        if self.kind == "yaml":
            _ = self.data  # pylint: disable=pointless-statement

    def _guess_kind(self) -> None:
        if self.kind == "yaml":
            if (
                isinstance(self.data, list)
                and len(self.data) > 0
                and (
                    "hosts" in self.data[0]
                    or "import_playbook" in self.data[0]
                    or "ansible.builtin.import_playbook" in self.data[0]
                )
            ):
                if "rules" not in self.data[0]:
                    self.kind = "playbook"
                else:
                    self.kind = "rulebook"
            # we we failed to guess the more specific kind, we warn user
            if self.kind == "yaml":
                _logger.debug(
                    "Passed '%s' positional argument was identified as generic '%s' file kind.",
                    self.name,
                    self.kind,
                )

    def __getitem__(self, key: Any) -> Any:
        """Provide compatibility subscriptable support."""
        if key == "path":
            return str(self.path)
        if key == "type":
            return str(self.kind)
        raise NotImplementedError

    def get(self, key: Any, default: Any = None) -> Any:
        """Provide compatibility subscriptable support."""
        try:
            return self[key]
        except NotImplementedError:
            return default

    def _populate_content_cache_from_disk(self) -> None:
        # Can raise UnicodeDecodeError
        self._content = self.path.expanduser().resolve().read_text(encoding="utf-8")

        if self._original_content is None:
            self._original_content = self._content

    @property
    def content(self) -> str:
        """Retrieve file content, from internal cache or disk."""
        if self._content is None:
            self._populate_content_cache_from_disk()
        return cast(str, self._content)

    @content.setter
    def content(self, value: str) -> None:
        """Update ``content`` and calculate ``updated``.

        To calculate ``updated`` this will read the file from disk if the cache
        has not already been populated.
        """
        if not isinstance(value, str):
            msg = f"Expected str but got {type(value)}"
            raise TypeError(msg)
        if self._original_content is None:
            if self._content is not None:
                self._original_content = self._content
            elif self.path.exists():
                self._populate_content_cache_from_disk()
            else:
                # new file
                self._original_content = ""
        self.updated = self._original_content != value
        self._content = value

    @content.deleter
    def content(self) -> None:
        """Reset the internal content cache."""
        self._content = None

    def write(self, *, force: bool = False) -> None:
        """Write the value of ``Lintable.content`` to disk.

        This only writes to disk if the content has been updated (``Lintable.updated``).
        For example, you can update the content, and then write it to disk like this:

        .. code:: python

            lintable.content = new_content
            lintable.write()

        Use ``force=True`` when you want to force a content rewrite even if the
        content has not changed. For example:

        .. code:: python

            lintable.write(force=True)
        """
        if not force and not self.updated:
            # No changes to write.
            return
        self.path.expanduser().resolve().write_text(
            self._content or "",
            encoding="utf-8",
        )

    def __hash__(self) -> int:
        """Return a hash value of the lintables."""
        return hash((self.name, self.kind, self.abspath))

    def __eq__(self, other: object) -> bool:
        """Identify whether the other object represents the same rule match."""
        if isinstance(other, Lintable):
            return bool(self.name == other.name and self.kind == other.kind)
        return False

    def __repr__(self) -> str:
        """Return user friendly representation of a lintable."""
        return f"{self.name} ({self.kind})"

    @property
    def data(self) -> Any:
        """Return loaded data representation for current file, if possible."""
        if self.state == States.NOT_LOADED:
            if self.path.is_dir():
                self.state = None
                return self.state
            try:
                if str(self.base_kind) == "text/yaml":
                    from ansiblelint.utils import (  # pylint: disable=import-outside-toplevel
                        parse_yaml_linenumbers,
                    )

                    self.state = parse_yaml_linenumbers(self)
                    # now that _data is not empty, we can try guessing if playbook or rulebook
                    # it has to be done before append_skipped_rules() call as it's relying
                    # on self.kind.
                    if self.kind == "yaml":
                        self._guess_kind()
                    # Lazy import to avoid delays and cyclic-imports
                    if "append_skipped_rules" not in globals():
                        # pylint: disable=import-outside-toplevel
                        from ansiblelint.skip_utils import append_skipped_rules

                    self.state = append_skipped_rules(self.state, self)
                else:
                    logging.debug(
                        "data set to None for %s due to being '%s' (%s) kind.",
                        self.path,
                        self.kind,
                        self.base_kind or "unknown",
                    )
                    self.state = States.UNKNOWN_DATA

            except (
                RuntimeError,
                FileNotFoundError,
                YAMLError,
                UnicodeDecodeError,
            ) as exc:
                self.state = States.LOAD_FAILED
                self.exc = exc
        return self.state


# pylint: disable=redefined-outer-name
def discover_lintables(options: Options) -> list[str]:
    """Find all files that we know how to lint.

    Return format is normalized, relative for stuff below cwd, ~/ for content
    under current user and absolute for everything else.
    """
    if not options.lintables:
        options.lintables = ["."]

    return [
        str(filename)
        for filename in get_all_files(
            *[Path(s) for s in options.lintables],
            exclude_paths=options.exclude_paths,
        )
    ]


def strip_dotslash_prefix(fname: str) -> str:
    """Remove ./ leading from filenames."""
    return fname[2:] if fname.startswith("./") else fname


def find_project_root(
    srcs: Sequence[str],
    config_file: str | None = None,
) -> tuple[Path, str]:
    """Return a directory containing .git or ansible-lint config files.

    That directory will be a common parent of all files and directories
    passed in `srcs`.

    If no directory in the tree contains a marker that would specify it's the
    project root, the root of the file system is returned.

    Returns a two-tuple with the first element as the project root path and
    the second element as a string describing the method by which the
    project root was discovered.
    """
    directory = None
    if not srcs:
        srcs = [str(Path.cwd().resolve().absolute())]
    path_srcs = [Path(Path.cwd(), src).resolve() for src in srcs]

    cfg_files = [config_file] if config_file else CONFIG_FILENAMES

    # A list of lists of parents for each 'src'. 'src' is included as a
    # "parent" of itself if it is a directory
    src_parents = [
        list(path.parents) + ([path] if path.is_dir() else []) for path in path_srcs
    ]

    common_base = max(
        set.intersection(*(set(parents) for parents in src_parents)),
        key=lambda path: path.parts,
    )

    for directory in (common_base, *common_base.parents):
        if (directory / ".git").exists():
            return directory, ".git directory"

        if (directory / ".hg").is_dir():
            return directory, ".hg directory"

        for cfg_file in cfg_files:
            # note that if cfg_file is already absolute, 'directory' is ignored
            resolved_cfg_path = directory / cfg_file
            if resolved_cfg_path.is_file():
                if os.path.isabs(cfg_file):
                    directory = Path(cfg_file).parent
                    if directory.name == ".config":
                        directory = directory.parent
                return directory, f"config file {resolved_cfg_path}"

    if not directory:
        return Path.cwd(), "current working directory"
    return directory, "file system root"


def expand_dirs_in_lintables(lintables: set[Lintable]) -> None:
    """Return all recognized lintables within given directory."""
    should_expand = False

    for item in lintables:
        if item.path.is_dir():
            should_expand = True
            break

    if should_expand:
        # this relies on git and we do not want to call unless needed
        all_files = discover_lintables(options)

        for item in copy.copy(lintables):
            if item.path.is_dir():
                for filename in all_files:
                    if filename.startswith(str(item.path)):
                        lintables.add(Lintable(filename))


def _guess_parent(lintable: Lintable) -> Lintable | None:
    """Return a parent directory for a lintable."""
    try:
        if lintable.path.parents[2].name == "roles":
            # role_name = lintable.parents[1].name
            return Lintable(lintable.path.parents[1], kind="role")
    except IndexError:
        pass
    return None


def get_all_files(
    *paths: Path,
    exclude_paths: list[str] | None = None,
) -> list[Path]:
    """Recursively retrieve all files from given folders."""
    all_files: list[Path] = []
    exclude_paths = [] if exclude_paths is None else exclude_paths

    def is_excluded(path_to_check: Path) -> bool:
        """Check if a file is exclude by current specs."""
        return any(spec.match_file(str(path_to_check)) for spec in pathspecs)

    for path in paths:
        pathspecs = [
            pathspec.GitIgnoreSpec.from_lines(
                [
                    ".git",
                    ".tox",
                    ".mypy_cache",
                    "__pycache__",
                    ".DS_Store",
                    ".coverage",
                    ".pytest_cache",
                    ".ruff_cache",
                    *exclude_paths,
                ],
            ),
        ]
        gitignore = path / ".gitignore"
        if gitignore.exists():
            with gitignore.open(encoding="UTF-8") as f:
                _logger.info("Loading ignores from %s", gitignore)
                pathspecs.append(
                    pathspec.GitIgnoreSpec.from_lines(f.read().splitlines()),
                )

        # Iterate over all items in the directory
        if path.is_file():
            all_files.append(path)
        else:
            for item in sorted(path.iterdir()):
                if is_excluded(item):
                    _logger.info("Excluded: %s", item)
                    continue
                if item.is_file():
                    all_files.append(item)
                # If it's a directory, recursively call the function
                elif item.is_dir():
                    all_files.extend(get_all_files(item, exclude_paths=exclude_paths))

    return all_files