Handle FQCN when using import_playbook (#4369)

collections/ansible_collections/local/testcollection/playbooks/foo.yml

@@ -0,0 +1,9 @@
+---
+- name: Fixture
+  hosts: localhost
+  connection: local
+  gather_facts: false
+  tasks:
+    - name: Another task
+      ansible.builtin.debug:
+        msg: debug message

examples/playbooks/import_playbook_fqcn.yml

@@ -0,0 +1,3 @@
+---
+- name: Import a playbook
+  ansible.builtin.import_playbook: local.testcollection.foo

src/ansiblelint/runner.py

@@ -553,8 +553,8 @@ def play_children(
             "block": handlers.taskshandlers_children,
             "include": handlers.include_children,
             "ansible.builtin.include": handlers.include_children,
-            "import_playbook": handlers.include_children,
-            "ansible.builtin.import_playbook": handlers.include_children,
+            "import_playbook": handlers.import_playbook_children,
+            "ansible.builtin.import_playbook": handlers.import_playbook_children,
             "roles": handlers.roles_children,
             "dependencies": handlers.roles_children,
             "handlers": handlers.taskshandlers_children,

src/ansiblelint/utils.py

@@ -299,12 +299,6 @@ def include_children(
     ) -> list[Lintable]:
         """Include children."""
         basedir = str(lintable.path.parent)
-        # import_playbook only accepts a string as argument (no dict syntax)
-        if k in (
-            "import_playbook",
-            "ansible.builtin.import_playbook",
-        ) and not isinstance(v, str):
-            return []
 
         # handle special case include_tasks: name=filename.yml
         if k in INCLUSION_ACTION_NAMES and isinstance(v, dict) and "file" in v:
@@ -314,17 +308,6 @@ def include_children(
         if not v or "{{" in v:
             return []
 
-        if k in ("import_playbook", "ansible.builtin.import_playbook"):
-            included = Path(basedir) / v
-            if not included.exists():
-                msg = f"Failed to find {v} playbook."
-            elif not self.app.runtime.has_playbook(v, basedir=Path(basedir)):
-                msg = f"Failed to load {v} playbook due to failing syntax check."
-            else:
-                return [Lintable(included, kind=parent_type)]
-            logging.error(msg)
-            return []
-
         # handle include: filename.yml tags=blah
         (args, kwargs) = tokenize(v)
 
@@ -457,6 +440,61 @@ def roles_children(
                 results.extend(_look_for_role_files(basedir, role))
         return results
 
+    def import_playbook_children(
+        self,
+        lintable: Lintable,
+        k: str,  # pylint: disable=unused-argument
+        v: Any,
+        parent_type: FileType,
+    ) -> list[Lintable]:
+        """Include import_playbook children."""
+
+        def append_playbook_path(loc: str, playbook_name: str) -> None:
+            possible_paths.append(
+                Path(
+                    path_dwim(
+                        os.path.expanduser(loc),
+                        os.path.join(
+                            "ansible_collections",
+                            namespace_name,
+                            collection_name,
+                            "playbooks",
+                            playbook_name,
+                        ),
+                    ),
+                ),
+            )
+
+        # import_playbook only accepts a string as argument (no dict syntax)
+        if not isinstance(v, str):
+            return []
+
+        possible_paths = []
+        namespace_name, collection_name, playbook_name = parse_fqcn(v)
+        if namespace_name and collection_name:
+            for loc in get_app(cached=True).runtime.config.collections_paths:
+                append_playbook_path(loc, f"{playbook_name}.yml")
+                append_playbook_path(loc, f"{playbook_name}.yaml")
+        else:
+            possible_paths.append(lintable.path.parent / v)
+
+        for possible_path in possible_paths:
+            if not possible_path.exists():
+                msg = f"Failed to find {v} playbook."
+            elif not self.app.runtime.has_playbook(
+                str(possible_path),
+            ):
+                msg = f"Failed to load {v} playbook due to failing syntax check."
+                break
+            elif namespace_name and collection_name:
+                # don't lint foreign playbook
+                return []
+            else:
+                return [Lintable(possible_path, kind=parent_type)]
+
+        logging.error(msg)
+        return []
+
 
 def _get_task_handler_children_for_tasks_or_playbooks(
     task_handler: dict[str, Any],
@@ -505,9 +543,7 @@ def _get_task_handler_children_for_tasks_or_playbooks(
 
 def _rolepath(basedir: str, role: str) -> str | None:
     role_path = None
-    namespace_name, collection_name, role_name = (
-        role.split(".") if is_fqcn(role) else ("", "", role)
-    )
+    namespace_name, collection_name, role_name = parse_fqcn(role)
 
     possible_paths = [
         # if included from a playbook
@@ -1135,3 +1171,8 @@ def load_plugin(name: str) -> PluginLoadContext:
             check_aliases=True,
         )
     return loaded_module
+
+
+def parse_fqcn(name: str) -> tuple[str, ...]:
+    """Parse name parameter into FQCN segments."""
+    return tuple(name.split(".")) if is_fqcn(name) else ("", "", name)

test/test_utils.py

@@ -503,3 +503,19 @@ def test_include_children_load_playbook_failed_syntax_check() -> None:
         "Failed to load syntax-error.yml playbook due to failing syntax check."
         in result.stderr
     )
+
+
+def test_import_playbook_children() -> None:
+    """Verify import_playbook_children()."""
+    result = run_ansible_lint(
+        Path("playbooks/import_playbook_fqcn.yml"),
+        cwd=Path(__file__).resolve().parent.parent / "examples",
+        env={
+            "ANSIBLE_COLLECTIONS_PATH": "../collections",
+        },
+    )
+    assert "Failed to find local.testcollection.foo playbook." not in result.stderr
+    assert (
+        "Failed to load local.testcollection.foo playbook due to failing syntax check."
+        not in result.stderr
+    )

tox.ini

@@ -75,7 +75,7 @@ setenv =
   PRE_COMMIT_COLOR = always
   # Number of expected test passes, safety measure for accidental skip of
   # tests. Update value if you add/remove tests. (tox-extra)
-  PYTEST_REQPASS = 894
+  PYTEST_REQPASS = 895
   FORCE_COLOR = 1
   pre: PIP_PRE = 1
 allowlist_externals =