1 files changed, 94 insertions, 0 deletions
diff --git a/lib/ansible/plugins/action/validate_argument_spec.py b/lib/ansible/plugins/action/validate_argument_spec.py
new file mode 100644
index 0000000..dc7d6cb
--- /dev/null
+++ b/lib/ansible/plugins/action/validate_argument_spec.py
@@ -0,0 +1,94 @@
+# Copyright 2021 Red Hat
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from ansible.errors import AnsibleError
+from ansible.plugins.action import ActionBase
+from ansible.module_utils.six import string_types
+from ansible.module_utils.common.arg_spec import ArgumentSpecValidator
+from ansible.module_utils.errors import AnsibleValidationErrorMultiple
+from ansible.utils.vars import combine_vars
+
+
+class ActionModule(ActionBase):
+    ''' Validate an arg spec'''
+
+    TRANSFERS_FILES = False
+
+    def get_args_from_task_vars(self, argument_spec, task_vars):
+        '''
+        Get any arguments that may come from `task_vars`.
+
+        Expand templated variables so we can validate the actual values.
+
+        :param argument_spec: A dict of the argument spec.
+        :param task_vars: A dict of task variables.
+
+        :returns: A dict of values that can be validated against the arg spec.
+        '''
+        args = {}
+
+        for argument_name, argument_attrs in argument_spec.items():
+            if argument_name in task_vars:
+                args[argument_name] = task_vars[argument_name]
+        args = self._templar.template(args)
+        return args
+
+    def run(self, tmp=None, task_vars=None):
+        '''
+        Validate an argument specification against a provided set of data.
+
+        The `validate_argument_spec` module expects to receive the arguments:
+            - argument_spec: A dict whose keys are the valid argument names, and
+                  whose values are dicts of the argument attributes (type, etc).
+            - provided_arguments: A dict whose keys are the argument names, and
+                  whose values are the argument value.
+
+        :param tmp: Deprecated. Do not use.
+        :param task_vars: A dict of task variables.
+        :return: An action result dict, including a 'argument_errors' key with a
+            list of validation errors found.
+        '''
+        if task_vars is None:
+            task_vars = dict()
+
+        result = super(ActionModule, self).run(tmp, task_vars)
+        del tmp  # tmp no longer has any effect
+
+        # This action can be called from anywhere, so pass in some info about what it is
+        # validating args for so the error results make some sense
+        result['validate_args_context'] = self._task.args.get('validate_args_context', {})
+
+        if 'argument_spec' not in self._task.args:
+            raise AnsibleError('"argument_spec" arg is required in args: %s' % self._task.args)
+
+        # Get the task var called argument_spec. This will contain the arg spec
+        # data dict (for the proper entry point for a role).
+        argument_spec_data = self._task.args.get('argument_spec')
+
+        # the values that were passed in and will be checked against argument_spec
+        provided_arguments = self._task.args.get('provided_arguments', {})
+
+        if not isinstance(argument_spec_data, dict):
+            raise AnsibleError('Incorrect type for argument_spec, expected dict and got %s' % type(argument_spec_data))
+
+        if not isinstance(provided_arguments, dict):
+            raise AnsibleError('Incorrect type for provided_arguments, expected dict and got %s' % type(provided_arguments))
+
+        args_from_vars = self.get_args_from_task_vars(argument_spec_data, task_vars)
+        validator = ArgumentSpecValidator(argument_spec_data)
+        validation_result = validator.validate(combine_vars(args_from_vars, provided_arguments), validate_role_argument_spec=True)
+
+        if validation_result.error_messages:
+            result['failed'] = True
+            result['msg'] = 'Validation of arguments failed:\n%s' % '\n'.join(validation_result.error_messages)
+            result['argument_spec_data'] = argument_spec_data
+            result['argument_errors'] = validation_result.error_messages
+            return result
+
+        result['changed'] = False
+        result['msg'] = 'The arg spec validation passed'
+
+        return result