refactor: extract common action preparation

changelogs/fragments/base_action_preparation.yml

@@ -0,0 +1,5 @@
+---
+minor_changes:
+  - "Extract common action preparation and resource construction helpers into BaseResourceActionPlugin."
+bugfixes:
+  - "Preserve the original error when action preparation fails before a result is returned."

plugins/action/base_action.py

@@ -359,6 +359,67 @@ def _pre_execute_hook(self, ansible_data: dict, write_only_data: dict, validated
             operation: The resolved operation string.
         """
 
+    def _build_resource(self, resource_data: dict) -> Any:
+        """Construct the resource model from filtered task parameters.
+
+        Launch-style action plugins can override this when their argument
+        specification contains control parameters that are not model fields.
+        The default preserves the existing CRUD action-plugin behavior.
+        """
+        return self.MODEL_CLASS(**resource_data)
+
+    def _prepare_action(self, tmp: object = None, task_vars: Optional[dict] = None) -> dict:
+        """Prepare the common inputs needed by an action plugin.
+
+        This helper deliberately stops before operation detection and manager
+        execution.  It is therefore usable by non-CRUD action plugins without
+        entering the idempotency state machine implemented by :meth:`run`.
+
+        Args:
+            tmp: Temporary directory passed through to Ansible's base action.
+            task_vars: Ansible task variables used to initialize the manager.
+
+        Returns:
+            dict: A dictionary containing ``result``, ``argspec``,
+                ``validated_params``, ``resource_data``, ``write_only_data``,
+                and ``manager``.
+        """
+        if task_vars is None:
+            task_vars = {}
+        self._task_vars = task_vars
+        result = super(BaseResourceActionPlugin, self).run(tmp, task_vars)
+        del tmp
+
+        doc = self._get_documentation()
+        argspec = self._build_argspec_from_docs(doc) if doc else None
+        if not argspec:
+            raise AnsibleError("Could not load DOCUMENTATION for %s module" % self.MODULE_NAME)
+
+        validated_input = self._validate_data(self._task.args.copy(), argspec, "input")
+        manager, facts_to_set = self._get_or_spawn_manager(task_vars)
+        self._client = manager
+        if facts_to_set:
+            result["ansible_facts"] = facts_to_set
+            result["_ansible_facts_cacheable"] = True
+
+        validated_params = validated_input.validated_parameters
+        resource_data = {k: v for k, v in validated_params.items() if v is not None and k not in self._AUTH_PARAMS}
+
+        for field, (msg, version) in self._DEPRECATED_FIELDS.items():
+            if resource_data.pop(field, None) is not None:
+                result.setdefault("deprecations", []).append({"msg": msg, "version": version, "collection_name": "ansible.platform"})
+
+        write_only_data = {f: resource_data.pop(f) for f in self._WRITE_ONLY_FIELDS if f in resource_data}
+
+        return {
+            "result": result,
+            "argspec": argspec,
+            "validated_params": validated_params,
+            "resource_data": resource_data,
+            "write_only_data": write_only_data,
+            "manager": manager,
+        }
+
     def _get_or_spawn_manager(self, task_vars: dict) -> Tuple[Union["DirectHTTPClient", "ManagerRPCClient"], Optional[Dict[str, Any]]]:
         """
         Dispatcher: Get connection client from the connection plugin.
@@ -1036,44 +1097,24 @@ def run(self, tmp: object = None, task_vars: Optional[dict] = None) -> dict:
         Returns:
             dict: Ansible result dictionary
         """
-        if task_vars is None:
-            task_vars = {}
-        self._task_vars = task_vars
-        result = super(BaseResourceActionPlugin, self).run(tmp, task_vars)
-        del tmp
-
         if self.MODEL_CLASS is None:
             raise AnsibleError("%s must set MODEL_CLASS or override run()" % type(self).__name__)
 
+        # Preparation can fail before _prepare_action() returns a result.
+        # Keep a valid Ansible result available so the exception handler does
+        # not mask the original validation, documentation, or connection error.
+        result = {}
         try:
-            # ---- argspec & input validation --------------------------------
-            doc = self._get_documentation()
-            argspec = self._build_argspec_from_docs(doc) if doc else None
-            if not argspec:
-                raise AnsibleError("Could not load DOCUMENTATION for %s module" % self.MODULE_NAME)
-            validated_input = self._validate_data(self._task.args.copy(), argspec, "input")
-
-            # ---- manager connection ----------------------------------------
-            manager, facts_to_set = self._get_or_spawn_manager(task_vars)
-            self._client = manager
-            if facts_to_set:
-                result["ansible_facts"] = facts_to_set
-                result["_ansible_facts_cacheable"] = True
+            prepared = self._prepare_action(tmp, task_vars)
+            result = prepared["result"]
+            argspec = prepared["argspec"]
+            validated_params = prepared["validated_params"]
+            resource_data = prepared["resource_data"]
+            _write_only_data = prepared["write_only_data"]
+            manager = prepared["manager"]
 
             # ---- build resource object -------------------------------------
-            validated_params = validated_input.validated_parameters
-            resource_data = {k: v for k, v in validated_params.items() if v is not None and k not in self._AUTH_PARAMS}
-
-            # Warn about and strip deprecated argspec fields.
-            for field, (msg, version) in self._DEPRECATED_FIELDS.items():
-                if resource_data.pop(field, None) is not None:
-                    result.setdefault("deprecations", []).append({"msg": msg, "version": version, "collection_name": "ansible.platform"})
-
-            # Pop write-only fields (not present in MODEL_CLASS) before instantiation;
-            # they are passed to _pre_execute_hook for use just before manager.execute().
-            _write_only_data = {f: resource_data.pop(f) for f in self._WRITE_ONLY_FIELDS if f in resource_data}
-
-            resource = self.MODEL_CLASS(**resource_data)
+            resource = self._build_resource(resource_data)
 
             # Allow subclasses to resolve lookup-by-id or other mutations.
             self._resolve_lookup(resource, resource_data, validated_params)
@@ -1196,7 +1237,7 @@ def run(self, tmp: object = None, task_vars: Optional[dict] = None) -> dict:
                             }
                         )
                         return result
-                    resource = self.MODEL_CLASS(**{k: v for k, v in merged.items() if hasattr(self.MODEL_CLASS, k)})
+                    resource = self._build_resource({k: v for k, v in merged.items() if hasattr(self.MODEL_CLASS, k)})
                     operation = "update"
                 else:
                     operation = "create"

tests/unit/plugins/action/test_base_action.py

@@ -0,0 +1,118 @@
+# (c) 2026 Red Hat Inc.
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+"""Unit tests for reusable BaseResourceActionPlugin preparation helpers."""
+
+from __future__ import absolute_import, division, print_function
+
+import unittest
+from dataclasses import dataclass
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from ansible.plugins.action import ActionBase
+from ansible_collections.ansible.platform.plugins.action.base_action import BaseResourceActionPlugin
+
+
+@dataclass
+class Resource:
+    name: str
+    value: int = 0
+
+
+class ExampleAction(BaseResourceActionPlugin):
+    MODULE_NAME = "example"
+    MODEL_CLASS = Resource
+    _WRITE_ONLY_FIELDS = frozenset({"write_only"})
+    _DEPRECATED_FIELDS = {"old_value": ("old_value is deprecated", "2.0.0")}
+
+
+class OverrideResourceAction(ExampleAction):
+    def _build_resource(self, resource_data):
+        return Resource(name=resource_data["name"], value=resource_data.get("value", 0) + 1)
+
+
+class TestActionPreparation(unittest.TestCase):
+    def _make_action(self):
+        action = ExampleAction.__new__(ExampleAction)
+        action._task = MagicMock()
+        action._task.args = {
+            "name": "resource",
+            "value": 3,
+            "write_only": "secret",
+            "old_value": "legacy",
+        }
+        action._display = MagicMock()
+        action._display.verbosity = 0
+        return action
+
+    def test_prepare_action_reuses_documentation_validation_and_manager(self):
+        action = self._make_action()
+        manager = MagicMock()
+        validation = SimpleNamespace(
+            validated_parameters=dict(action._task.args),
+        )
+
+        with patch.object(ActionBase, "run", return_value={"initial": True}) as base_run:
+            with patch.object(action, "_get_documentation", return_value="module: example") as get_doc:
+                with patch.object(action, "_build_argspec_from_docs", return_value={"argument_spec": {}}) as build_argspec:
+                    with patch.object(action, "_validate_data", return_value=validation) as validate:
+                        with patch.object(action, "_get_or_spawn_manager", return_value=(manager, {"platform": "fact"})) as get_manager:
+                            prepared = action._prepare_action(task_vars={"inventory_hostname": "localhost"})
+
+        base_run.assert_called_once()
+        get_doc.assert_called_once_with()
+        build_argspec.assert_called_once_with("module: example")
+        validate.assert_called_once_with(action._task.args.copy(), {"argument_spec": {}}, "input")
+        get_manager.assert_called_once_with({"inventory_hostname": "localhost"})
+        self.assertIs(prepared["manager"], manager)
+        self.assertEqual(prepared["resource_data"], {"name": "resource", "value": 3})
+        self.assertEqual(prepared["write_only_data"], {"write_only": "secret"})
+        self.assertEqual(prepared["result"]["ansible_facts"], {"platform": "fact"})
+        self.assertEqual(prepared["result"]["deprecations"][0]["version"], "2.0.0")
+
+    def test_run_dispatches_to_build_resource_override(self):
+        action = OverrideResourceAction.__new__(OverrideResourceAction)
+        action._task = MagicMock()
+        action._task.check_mode = False
+        action._display = MagicMock()
+        action._display.verbosity = 0
+        manager = MagicMock()
+        manager.execute.return_value = {"id": 1, "name": "resource", "value": 5, "changed": True}
+        prepared = {
+            "result": {},
+            "argspec": {"argument_spec": {"name": {}, "value": {}}},
+            "validated_params": {"name": "resource", "value": 4, "state": "present"},
+            "resource_data": {"name": "resource", "value": 4},
+            "write_only_data": {},
+            "manager": manager,
+        }
+
+        with patch.object(action, "_prepare_action", return_value=prepared):
+            with patch.object(action, "_resolve_lookup"):
+                with patch.object(action, "_build_resource", wraps=action._build_resource) as build_resource:
+                    result = action.run(task_vars={})
+
+        build_resource.assert_called_once_with({"name": "resource", "value": 4})
+        self.assertEqual(result["id"], 1)
+
+    def test_prepare_action_rejects_missing_documentation(self):
+        action = self._make_action()
+
+        with patch.object(ActionBase, "run", return_value={}):
+            with patch.object(action, "_get_documentation", return_value=""):
+                with self.assertRaisesRegex(Exception, "Could not load DOCUMENTATION"):
+                    action._prepare_action()
+
+    def test_run_preserves_preparation_error(self):
+        action = self._make_action()
+
+        with patch.object(action, "_prepare_action", side_effect=ValueError("invalid documentation")):
+            result = action.run(task_vars={})
+
+        self.assertTrue(result["failed"])
+        self.assertEqual(result["msg"], "invalid documentation")
+
+
+if __name__ == "__main__":
+    unittest.main()