Server IP : 85.214.239.14 / Your IP : 3.15.34.105 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/3/root/proc/self/root/lib/python3/dist-packages/ansible/plugins/action/ |
Upload File : |
# Copyright 2012, Dag Wieers <dag@wieers.com> # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see <http://www.gnu.org/licenses/>. from __future__ import (absolute_import, division, print_function) __metaclass__ = type from ansible.errors import AnsibleError from ansible.playbook.conditional import Conditional from ansible.plugins.action import ActionBase from ansible.module_utils.six import string_types from ansible.module_utils.parsing.convert_bool import boolean class ActionModule(ActionBase): ''' Fail with custom message ''' TRANSFERS_FILES = False _VALID_ARGS = frozenset(('fail_msg', 'msg', 'quiet', 'success_msg', 'that')) def run(self, tmp=None, task_vars=None): if task_vars is None: task_vars = dict() result = super(ActionModule, self).run(tmp, task_vars) del tmp # tmp no longer has any effect if 'that' not in self._task.args: raise AnsibleError('conditional required in "that" string') fail_msg = None success_msg = None fail_msg = self._task.args.get('fail_msg', self._task.args.get('msg')) if fail_msg is None: fail_msg = 'Assertion failed' elif isinstance(fail_msg, list): if not all(isinstance(x, string_types) for x in fail_msg): raise AnsibleError('Type of one of the elements in fail_msg or msg list is not string type') elif not isinstance(fail_msg, (string_types, list)): raise AnsibleError('Incorrect type for fail_msg or msg, expected a string or list and got %s' % type(fail_msg)) success_msg = self._task.args.get('success_msg') if success_msg is None: success_msg = 'All assertions passed' elif isinstance(success_msg, list): if not all(isinstance(x, string_types) for x in success_msg): raise AnsibleError('Type of one of the elements in success_msg list is not string type') elif not isinstance(success_msg, (string_types, list)): raise AnsibleError('Incorrect type for success_msg, expected a string or list and got %s' % type(success_msg)) quiet = boolean(self._task.args.get('quiet', False), strict=False) # directly access 'that' via untemplated args from the task so we can intelligently trust embedded # templates and preserve the original inputs/locations for better messaging on assert failures and # errors. # FIXME: even in devel, things like `that: item` don't always work properly (truthy string value # is not really an embedded expression) # we could fix that by doing direct var lookups on the inputs # FIXME: some form of this code should probably be shared between debug, assert, and # Task.post_validate, since they # have a lot of overlapping needs try: thats = self._task.untemplated_args['that'] except KeyError: # in the case of "we got our entire args dict from a template", we can just consult the # post-templated dict (the damage has likely already been done for embedded templates anyway) thats = self._task.args['that'] # FIXME: this is a case where we only want to resolve indirections, NOT recurse containers # (and even then, the leaf-most expression being wrapped is at least suboptimal # (since its expression will be "eaten"). if isinstance(thats, str): thats = self._templar.template(thats) # make sure the 'that' items are a list if not isinstance(thats, list): thats = [thats] # Now we iterate over the that items, temporarily assigning them # to the task's when value so we can evaluate the conditional using # the built in evaluate function. The when has already been evaluated # by this point, and is not used again, so we don't care about mangling # that value now cond = Conditional(loader=self._loader) if not quiet: result['_ansible_verbose_always'] = True for that in thats: cond.when = [that] test_result = cond.evaluate_conditional(templar=self._templar, all_vars=task_vars) if not test_result: result['failed'] = True result['evaluated_to'] = test_result result['assertion'] = that result['msg'] = fail_msg return result result['changed'] = False result['msg'] = success_msg return result