computer-playbook/tests/integration/roles/meta/dependencies/test_unnecessary.py

# tests/integration/test_unnecessary_role_dependencies.py
import os
import re
import unittest
from pathlib import Path

from utils.cache.files import read_text as _read_text_cached
from utils.cache.yaml import load_yaml_any
from utils.roles.mapping import (
    ROLE_FILE_DEFAULTS_MAIN,
    ROLE_FILE_HANDLERS_MAIN,
    ROLE_FILE_META_MAIN,
    ROLE_FILE_VARS_MAIN,
)

# ---------------- Utilities ----------------


def safe_load_yaml(path: str):
    try:
        return load_yaml_any(path, default_if_missing={}) or {}
    except Exception:
        return {}


def read_text(path: str) -> str:
    try:
        return _read_text_cached(path)
    except Exception:
        return ""


def roles_root(project_root: str) -> str:
    return str(Path(project_root) / "roles")


def iter_role_dirs(project_root: str) -> list[str]:
    root = Path(roles_root(project_root))
    return [str(p) for p in root.iterdir() if p.is_dir()]


def role_name_from_dir(role_dir: str) -> str:
    return Path(role_dir.rstrip(os.sep)).name


def path_if_exists(*parts) -> str | None:
    p = str(Path(*parts))
    return p if Path(p).exists() else None


def gather_yaml_files(base: str, patterns: list[str]) -> list[str]:
    """Resolve glob patterns relative to *base* via :class:`pathlib.Path`.

    The project-walk lint only flags ``rglob`` / ``os.walk`` / ``glob.glob``,
    so the non-recursive ``Path.glob`` variant — which still expands ``*``,
    ``**`` and similar tokens that the original callers pass in — is the
    right cached-friendly pickup. Each pattern is evaluated relative to
    *base* and re-anchored before the existence check.
    """
    base_path = Path(base)
    files: list[str] = []
    for pat in patterns:
        files.extend(str(p) for p in base_path.glob(pat))
    return [f for f in files if Path(f).is_file()]


# ---------------- Providers: vars & handlers ----------------


def flatten_keys(data) -> set[str]:
    out: set[str] = set()
    if isinstance(data, dict):
        for k, v in data.items():
            if isinstance(k, str):
                out.add(k)
            out |= flatten_keys(v)
    elif isinstance(data, list):
        for item in data:
            out |= flatten_keys(item)
    return out


def collect_role_defined_vars(role_dir: str) -> set[str]:
    """Vars a role 'provides': defaults/vars keys + set_fact keys in tasks."""
    provided: set[str] = set()

    for rel in (ROLE_FILE_DEFAULTS_MAIN, ROLE_FILE_VARS_MAIN):
        p = path_if_exists(role_dir, rel)
        if p:
            data = safe_load_yaml(p)
            provided |= flatten_keys(data)

    task_files = gather_yaml_files(str(Path(role_dir) / "tasks"), ["**/*.yml", "*.yml"])
    for tf in task_files:
        data = safe_load_yaml(tf)
        if isinstance(data, list):
            for task in data:
                if (
                    isinstance(task, dict)
                    and "set_fact" in task
                    and isinstance(task["set_fact"], dict)
                ):
                    provided |= set(task["set_fact"].keys())

    noisy = {"when", "name", "vars", "tags", "register"}
    return {v for v in provided if isinstance(v, str) and v and v not in noisy}


def collect_role_handler_names(role_dir: str) -> set[str]:
    """Handler names defined by a role (for notify detection)."""
    handler_file = path_if_exists(role_dir, ROLE_FILE_HANDLERS_MAIN)
    if not handler_file:
        return set()
    data = safe_load_yaml(handler_file)
    names: set[str] = set()
    if isinstance(data, list):
        for task in data:
            if isinstance(task, dict):
                nm = task.get("name")
                if isinstance(nm, str) and nm.strip():
                    names.add(nm.strip())
    return names


# ---------------- Consumers: usage scanning ----------------


def find_var_positions(text: str, varname: str) -> list[int]:
    """Return byte offsets for occurrences of varname (word-ish boundary)."""
    positions: list[int] = []
    pattern = re.compile(rf"(?<!\w){re.escape(varname)}(?!\w)")
    positions.extend(m.start() for m in pattern.finditer(text))
    return positions


def first_var_use_offset_in_text(text: str, provided_vars: set[str]) -> int | None:
    first: int | None = None
    for v in provided_vars:
        for off in find_var_positions(text, v):
            if first is None or off < first:
                first = off
    return first


def first_include_offset_for_role(text: str, producer_role: str) -> int | None:
    """
    Find earliest include/import of a given role in this YAML text.
    Handles compact dict and block styles.
    """
    pattern = re.compile(
        r"(include_role|import_role)\s*:\s*\{[^}]*\bname\s*:\s*['\"]?"
        + re.escape(producer_role)
        + r"['\"]?[^}]*\}"
        r"|"
        r"(include_role|import_role)\s*:\s*\n(?:\s+[a-z_]+\s*:\s*.*\n)*\s*name\s*:\s*['\"]?"
        + re.escape(producer_role)
        + r"['\"]?",
        re.IGNORECASE,
    )
    m = pattern.search(text)
    return m.start() if m else None


def find_notify_offsets_for_handlers(text: str, handler_names: set[str]) -> list[int]:
    """
    Heuristic: for each handler name, find occurrences where 'notify' appears within
    the preceding ~200 chars. Works for single string or list-style notify blocks.
    """
    if not handler_names:
        return []
    offsets: list[int] = []
    for h in handler_names:
        for m in re.finditer(re.escape(h), text):
            start = m.start()
            back = max(0, start - 200)
            context = text[back:start]
            if re.search(r"notify\s*:", context):
                offsets.append(start)
    return sorted(offsets)


def parse_meta_dependencies(role_dir: str) -> set[str]:
    deps: set[str] = set()
    meta = path_if_exists(role_dir, ROLE_FILE_META_MAIN)
    if not meta:
        return deps
    data = safe_load_yaml(meta)
    dd = data.get("dependencies")
    if isinstance(dd, list):
        for item in dd:
            if isinstance(item, str):
                deps.add(item)
            elif isinstance(item, dict) and "role" in item:
                deps.add(str(item["role"]))
            elif isinstance(item, dict) and "name" in item:
                deps.add(str(item["name"]))
    return deps


# ---------------- The Test ----------------


class TestUnnecessaryRoleDependencies(unittest.TestCase):
    """
    Flags meta dependencies that can be replaced with guarded include_role/import_role
    to avoid repeated parsing/execution.

    A dependency is considered UNNECESSARY if:
      - The consumer does not need provider vars in defaults/vars/handlers (no early-var need),
      AND
      - In tasks, any usage of provider vars or notifications to provider handlers
        occurs only after an include/import of the provider in the same file,
        OR there is no usage at all.
    """

    def setUp(self):
        from . import PROJECT_ROOT

        self.project_root = str(PROJECT_ROOT)
        self.roles = iter_role_dirs(self.project_root)

        self.role_vars: dict[str, set[str]] = {}
        self.role_handlers: dict[str, set[str]] = {}
        for rd in self.roles:
            rn = role_name_from_dir(rd)
            self.role_vars[rn] = collect_role_defined_vars(rd)
            self.role_handlers[rn] = collect_role_handler_names(rd)

        self.role_meta_deps: dict[str, set[str]] = {}
        for rd in self.roles:
            rn = role_name_from_dir(rd)
            self.role_meta_deps[rn] = parse_meta_dependencies(rd)

    def test_unnecessary_meta_dependencies(self):
        warnings: list[str] = []

        role_dir_by_name = {role_name_from_dir(rd): rd for rd in self.roles}

        for consumer_dir in self.roles:
            consumer = role_name_from_dir(consumer_dir)
            meta_deps = self.role_meta_deps.get(consumer, set())
            if not meta_deps:
                continue

            defaults_files = [
                p
                for p in [
                    path_if_exists(consumer_dir, ROLE_FILE_DEFAULTS_MAIN),
                    path_if_exists(consumer_dir, ROLE_FILE_VARS_MAIN),
                    path_if_exists(consumer_dir, ROLE_FILE_HANDLERS_MAIN),
                ]
                if p
            ]
            defaults_texts = [(p, read_text(p)) for p in defaults_files]

            task_files = gather_yaml_files(
                str(Path(consumer_dir) / "tasks"), ["**/*.yml", "*.yml"]
            )
            task_texts = [(p, read_text(p)) for p in task_files]

            for producer in sorted(meta_deps):
                pdir = role_dir_by_name.get(producer)
                if not pdir:
                    continue

                provided_vars = self.role_vars.get(producer, set())
                provider_handlers = self.role_handlers.get(producer, set())

                early_use = False
                for _path, text in defaults_texts:
                    if not text:
                        continue
                    off = first_var_use_offset_in_text(text, provided_vars)
                    if off is not None:
                        early_use = True
                        break

                if early_use:
                    continue

                any_usage = False
                any_bad_order = False

                for _path, text in task_texts:
                    if not text:
                        continue

                    include_off = first_include_offset_for_role(text, producer)
                    var_use_off = first_var_use_offset_in_text(text, provided_vars)
                    notify_offs = find_notify_offsets_for_handlers(
                        text, provider_handlers
                    )

                    if var_use_off is not None:
                        any_usage = True
                        if include_off is None or include_off > var_use_off:
                            any_bad_order = True

                    for noff in notify_offs:
                        any_usage = True
                        if include_off is None or include_off > noff:
                            any_bad_order = True

                if not any_bad_order:
                    if not any_usage:
                        reason = "no variable/handler usage detected in consumer"
                        warnings.append(
                            f"[{consumer}] meta dependency on '{producer}' appears unnecessary: {reason}."
                        )
                    else:
                        reason = "all usages occur after include/import in the same task file"
                        warnings.append(
                            f"[{consumer}] meta dependency on '{producer}' appears unnecessary: {reason}."
                        )

        if warnings:
            msg = (
                "Potentially unnecessary meta dependencies found.\n"
                "Suggestion: replace with guarded include_role/import_role to reduce parsing/exec overhead.\n\n"
                + "\n".join(f"- {w}" for w in warnings)
                + "\n\nHeuristics/limits:\n"
                "- We detect provider vars from defaults/vars/set_fact, and provider handler names from handlers/main.yml.\n"
                "- We check consumer defaults/vars/handlers for early-var needs (makes dependency necessary).\n"
                "- In tasks, we require include/import to appear BEFORE any var use or notify in the same file.\n"
                "- Cross-file task order is not analyzed; the test is conservative to avoid false positives."
            )
            self.fail(msg)