Avoid using get_app() from inside the rule

This change does add the app object to the rule collection in order
to make it available to the rules. Also it makes the field_checks
lazy loading, so it would only be initialized much later, after
application was fully initialized.

This fixed a bug observed while running tests in parallel, where
app was initialized multiple times causing conflicts with collection
installation.

src/ansiblelint/rules/__init__.py

@@ -22,6 +22,7 @@
     RuntimeErrorRule,
     WarningRule,
 )
+from ansiblelint.app import App, get_app
 from ansiblelint.config import PROFILES, Options, get_rule_config
 from ansiblelint.config import options as default_options
 from ansiblelint.constants import LINE_NUMBER_KEY, RULE_DOC_URL, SKIPPED_RULES_KEY
@@ -376,10 +377,13 @@ def __init__(
         profile_name: str | None = None,
         *,
         conditional: bool = True,
+        app: App | None = None,
     ) -> None:
         """Initialize a RulesCollection instance."""
         self.options = options
         self.profile = []
+        self.app = app or get_app()
+
         if profile_name:
             self.profile = PROFILES[profile_name]
         rulesdirs_str = [] if rulesdirs is None else [str(r) for r in rulesdirs]
@@ -408,6 +412,7 @@ def register(self, obj: AnsibleLintRule, *, conditional: bool = False) -> None:
         """Register a rule."""
         # We skip opt-in rules which were not manually enabled.
         # But we do include opt-in rules when listing all rules or tags
+        obj._collection = self  # pylint: disable=protected-access # noqa: SLF001
         if any(
             [
                 not conditional,
@@ -418,7 +423,6 @@ def register(self, obj: AnsibleLintRule, *, conditional: bool = False) -> None:
                 self.options.list_tags,
             ],
         ):
-            obj._collection = self  # pylint: disable=protected-access # noqa: SLF001
             self.rules.append(obj)
 
     def __iter__(self) -> Iterator[BaseRule]:

src/ansiblelint/rules/schema.py

@@ -5,7 +5,6 @@
 import sys
 from typing import TYPE_CHECKING, Any
 
-from ansiblelint.app import get_app
 from ansiblelint.errors import MatchError
 from ansiblelint.file_utils import Lintable
 from ansiblelint.rules import AnsibleLintRule
@@ -49,10 +48,6 @@
     },
 }
 
-FIELD_CHECKS = {
-    "become_method": get_app().runtime.plugins.become.keys(),  # pylint: disable=no-member
-}
-
 
 class ValidateSchemaRule(AnsibleLintRule):
     """Perform JSON Schema Validation for known lintable kinds."""
@@ -80,64 +75,79 @@ class ValidateSchemaRule(AnsibleLintRule):
         "schema[tasks]": "",
         "schema[vars]": "",
     }
-
-    become_method_msg = f"'become_method' must be one of the currently installed plugins: {', '.join(FIELD_CHECKS['become_method'])}"
+    _field_checks: dict[str, list[str]] = {}
+
+    @property
+    def field_checks(self) -> dict[str, list[str]]:
+        """Lazy property for returning field checks."""
+        if not self._collection:
+            msg = "Rule was not registered to a RuleCollection."
+            raise RuntimeError(msg)
+        if not self._field_checks:
+            self._field_checks = {
+                "become_method": sorted(
+                    self._collection.app.runtime.plugins.become.keys(),
+                ),
+            }
+        return self._field_checks
 
     def matchplay(self, file: Lintable, data: dict[str, Any]) -> list[MatchError]:
         """Return matches found for a specific playbook."""
         results: list[MatchError] = []
         if not data or file.kind not in ("tasks", "handlers", "playbook"):
             return results
         # check at play level
-        for key, value in FIELD_CHECKS.items():
+        results.extend(self._get_field_matches(file=file, data=data))
+        return results
+
+    def _get_field_matches(
+        self,
+        file: Lintable,
+        data: dict[str, Any],
+    ) -> list[MatchError]:
+        """Retrieve all matches related to fields for the given data block."""
+        results = []
+        for key, values in self.field_checks.items():
             if key in data:
-                plugin_value = data.get(key, None)
-                if not has_jinja(plugin_value) and plugin_value not in value:
+                plugin_value = data[key]
+                if not has_jinja(plugin_value) and plugin_value not in values:
+                    # breakpoint()
+                    msg = f"'{key}' must be one of the currently available values: {', '.join(values)}"
                     results.append(
                         MatchError(
-                            message=self.become_method_msg,
-                            lintable=file or Lintable(""),
+                            message=msg,
+                            lineno=data.get("__line__", 1),
+                            lintable=file,
                             rule=ValidateSchemaRule(),
                             details=ValidateSchemaRule.description,
                             tag=f"schema[{file.kind}]",
                         ),
                     )
-
         return results
 
     def matchtask(
         self,
         task: Task,
         file: Lintable | None = None,
     ) -> bool | str | MatchError | list[MatchError]:
-        result = []
-        for key, value in FIELD_CHECKS.items():
-            if key in task.raw_task:
-                plugin_value = task.raw_task.get(key, None)
-                if not has_jinja(plugin_value) and plugin_value not in value:
-                    result.append(
-                        MatchError(
-                            message=self.become_method_msg,
-                            lintable=file or Lintable(""),
-                            rule=ValidateSchemaRule(),
-                            details=ValidateSchemaRule.description,
-                            tag=f"schema[{file.kind}]",  # type: ignore[union-attr]
-                        ),
-                    )
+        results = []
+        if not file:
+            file = Lintable("", kind="tasks")
+        results.extend(self._get_field_matches(file=file, data=task.raw_task))
         for key in pre_checks["task"]:
             if key in task.raw_task:
                 msg = pre_checks["task"][key]["msg"]
                 tag = pre_checks["task"][key]["tag"]
-                result.append(
+                results.append(
                     MatchError(
                         message=msg,
-                        lintable=file or Lintable(""),
+                        lintable=file,
                         rule=ValidateSchemaRule(),
                         details=ValidateSchemaRule.description,
                         tag=f"schema[{tag}]",
                     ),
                 )
-        return result
+        return results
 
     def matchyaml(self, file: Lintable) -> list[MatchError]:
         """Return JSON validation errors found as a list of MatchError(s)."""
@@ -308,8 +318,8 @@ def matchyaml(self, file: Lintable) -> list[MatchError]:
                 "examples/playbooks/rule-schema-become-method-fail.yml",
                 "playbook",
                 [
-                    "'become_method' must be one of the currently installed plugins",
-                    "'become_method' must be one of the currently installed plugins",
+                    "'become_method' must be one of the currently available values",
+                    "'become_method' must be one of the currently available values",
                 ],
                 id="playbook2",
             ),