Merging upstream version 24.6.1.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 134 insertions, 20 deletions

diff --git a/src/ansiblelint/rules/name.py b/src/ansiblelint/rules/name.py
index 41ce5cb..b814a41 100644
--- a/src/ansiblelint/rules/name.py
+++ b/src/ansiblelint/rules/name.py
@@ -1,19 +1,23 @@
 """Implementation of NameRule."""
+
 from __future__ import annotations
 
 import re
 import sys
-from copy import deepcopy
 from typing import TYPE_CHECKING, Any
 
+import wcmatch.pathlib
+import wcmatch.wcmatch
+
 from ansiblelint.constants import LINE_NUMBER_KEY
+from ansiblelint.file_utils import Lintable
 from ansiblelint.rules import AnsibleLintRule, TransformMixin
 
 if TYPE_CHECKING:
     from ruamel.yaml.comments import CommentedMap, CommentedSeq
 
+    from ansiblelint.config import Options
     from ansiblelint.errors import MatchError
-    from ansiblelint.file_utils import Lintable
     from ansiblelint.utils import Task
 
 
@@ -39,9 +43,11 @@ class NameRule(AnsibleLintRule, TransformMixin):
 
     def matchplay(self, file: Lintable, data: dict[str, Any]) -> list[MatchError]:
         """Return matches found for a specific play (entry in playbook)."""
-        results = []
+        results: list[MatchError] = []
         if file.kind != "playbook":
             return []
+        if file.failed():
+            return results
         if "name" not in data:
             return [
                 self.create_matcherror(
@@ -65,7 +71,9 @@ class NameRule(AnsibleLintRule, TransformMixin):
         task: Task,
         file: Lintable | None = None,
     ) -> list[MatchError]:
-        results = []
+        results: list[MatchError] = []
+        if file and file.failed():
+            return results
         name = task.get("name")
         if not name:
             results.append(
@@ -84,6 +92,7 @@ class NameRule(AnsibleLintRule, TransformMixin):
                     lineno=task[LINE_NUMBER_KEY],
                 ),
             )
+
         return results
 
     def _prefix_check(
@@ -120,10 +129,15 @@ class NameRule(AnsibleLintRule, TransformMixin):
         # stage one check prefix
         effective_name = name
         if self._collection and lintable:
-            prefix = self._collection.options.task_name_prefix.format(
-                stem=lintable.path.stem,
-            )
-            if lintable.kind == "tasks" and lintable.path.stem != "main":
+            full_stem = self._find_full_stem(lintable)
+            stems = [
+                self._collection.options.task_name_prefix.format(stem=stem)
+                for stem in wcmatch.pathlib.PurePath(
+                    full_stem,
+                ).parts
+            ]
+            prefix = "".join(stems)
+            if lintable.kind == "tasks" and full_stem != "main":
                 if not name.startswith(prefix):
                     # For the moment in order to raise errors this rule needs to be
                     # enabled manually. Still, we do allow use of prefixes even without
@@ -165,6 +179,38 @@ class NameRule(AnsibleLintRule, TransformMixin):
             )
         return results
 
+    def _find_full_stem(self, lintable: Lintable) -> str:
+        lintable_dir = wcmatch.pathlib.PurePath(lintable.dir)
+        stem = lintable.path.stem
+        kind = str(lintable.kind)
+
+        stems = [lintable_dir.name]
+        lintable_dir = lintable_dir.parent
+        pathex = lintable_dir / stem
+        glob = ""
+
+        if self.options:
+            for entry in self.options.kinds:
+                for key, value in entry.items():
+                    if kind == key:
+                        glob = value
+
+        while pathex.globmatch(
+            glob,
+            flags=(
+                wcmatch.pathlib.GLOBSTAR
+                | wcmatch.pathlib.BRACE
+                | wcmatch.pathlib.DOTGLOB
+            ),
+        ):
+            stems.insert(0, lintable_dir.name)
+            lintable_dir = lintable_dir.parent
+            pathex = lintable_dir / stem
+
+        if stems[0].startswith(kind):
+            del stems[0]
+        return str(wcmatch.pathlib.PurePath(*stems, stem))
+
     def transform(
         self,
         match: MatchError,
@@ -172,17 +218,44 @@ class NameRule(AnsibleLintRule, TransformMixin):
         data: CommentedMap | CommentedSeq | str,
     ) -> None:
         if match.tag == "name[casing]":
+
+            def update_task_name(task_name: str) -> str:
+                """Capitalize the first work of the task name."""
+                # Not using capitalize(), since that rewrites the rest of the name to lower case
+                if "|" in task_name:  # if using prefix
+                    [file_name, update_task_name] = task_name.split("|")
+                    return f"{file_name.strip()} | {update_task_name.strip()[:1].upper()}{update_task_name.strip()[1:]}"
+
+                return f"{task_name[:1].upper()}{task_name[1:]}"
+
             target_task = self.seek(match.yaml_path, data)
-            # Not using capitalize(), since that rewrites the rest of the name to lower case
-            target_task[
-                "name"
-            ] = f"{target_task['name'][:1].upper()}{target_task['name'][1:]}"
-            match.fixed = True
+            orig_task_name = target_task.get("name", None)
+            # pylint: disable=too-many-nested-blocks
+            if orig_task_name:
+                updated_task_name = update_task_name(orig_task_name)
+                for item in data:
+                    if isinstance(item, dict) and "tasks" in item:
+                        for task in item["tasks"]:
+                            # We want to rewrite task names in the notify keyword, but
+                            # if there isn't a notify section, there's nothing to do.
+                            if "notify" not in task:
+                                continue
+
+                            if (
+                                isinstance(task["notify"], str)
+                                and orig_task_name == task["notify"]
+                            ):
+                                task["notify"] = updated_task_name
+                            elif isinstance(task["notify"], list):
+                                for idx in range(len(task["notify"])):
+                                    if orig_task_name == task["notify"][idx]:
+                                        task["notify"][idx] = updated_task_name
+
+                target_task["name"] = updated_task_name
+                match.fixed = True
 
 
 if "pytest" in sys.modules:
-    from ansiblelint.config import options
-    from ansiblelint.file_utils import Lintable  # noqa: F811
     from ansiblelint.rules import RulesCollection
     from ansiblelint.runner import Runner
 
@@ -203,11 +276,23 @@ if "pytest" in sys.modules:
         errs = bad_runner.run()
         assert len(errs) == 5
 
-    def test_name_prefix_negative() -> None:
+    def test_name_prefix_positive(config_options: Options) -> None:
+        """Positive test for name[prefix]."""
+        config_options.enable_list = ["name[prefix]"]
+        collection = RulesCollection(options=config_options)
+        collection.register(NameRule())
+        success = Lintable(
+            "examples/playbooks/tasks/main.yml",
+            kind="tasks",
+        )
+        good_runner = Runner(success, rules=collection)
+        results = good_runner.run()
+        assert len(results) == 0
+
+    def test_name_prefix_negative(config_options: Options) -> None:
         """Negative test for name[missing]."""
-        custom_options = deepcopy(options)
-        custom_options.enable_list = ["name[prefix]"]
-        collection = RulesCollection(options=custom_options)
+        config_options.enable_list = ["name[prefix]"]
+        collection = RulesCollection(options=config_options)
         collection.register(NameRule())
         failure = Lintable(
             "examples/playbooks/tasks/rule-name-prefix-fail.yml",
@@ -221,6 +306,36 @@ if "pytest" in sys.modules:
         assert results[1].tag == "name[prefix]"
         assert results[2].tag == "name[prefix]"
 
+    def test_name_prefix_negative_2(config_options: Options) -> None:
+        """Negative test for name[prefix]."""
+        config_options.enable_list = ["name[prefix]"]
+        collection = RulesCollection(options=config_options)
+        collection.register(NameRule())
+        failure = Lintable(
+            "examples/playbooks/tasks/partial_prefix/foo.yml",
+            kind="tasks",
+        )
+        bad_runner = Runner(failure, rules=collection)
+        results = bad_runner.run()
+        assert len(results) == 2
+        assert results[0].tag == "name[prefix]"
+        assert results[1].tag == "name[prefix]"
+
+    def test_name_prefix_negative_3(config_options: Options) -> None:
+        """Negative test for name[prefix]."""
+        config_options.enable_list = ["name[prefix]"]
+        collection = RulesCollection(options=config_options)
+        collection.register(NameRule())
+        failure = Lintable(
+            "examples/playbooks/tasks/partial_prefix/main.yml",
+            kind="tasks",
+        )
+        bad_runner = Runner(failure, rules=collection)
+        results = bad_runner.run()
+        assert len(results) == 2
+        assert results[0].tag == "name[prefix]"
+        assert results[1].tag == "name[prefix]"
+
     def test_rule_name_lowercase() -> None:
         """Negative test for a task that starts with lowercase."""
         collection = RulesCollection()
@@ -255,6 +370,5 @@ if "pytest" in sys.modules:
     def test_when_no_lintable() -> None:
         """Test when lintable is None."""
         name_rule = NameRule()
-        # pylint: disable=protected-access
         result = name_rule._prefix_check("Foo", None, 1)  # noqa: SLF001
         assert len(result) == 0