def queue_exc(*arg, **kw): # pylint: disable=W0613
"""Queue undefined variable exception"""
_self = arg[0]
if not isinstance(_self, AnsibleUndefinedVariable):
# Run for AnsibleUndefinedVariable instance
return
_rslt_q = None
for stack_trace in inspect.stack():
# Check if method to be skipped
if stack_trace[3] in SKIP_METHODS:
continue
_frame = stack_trace[0]
_locals = inspect.getargvalues(_frame).locals
if 'self' not in _locals:
continue
# Check if current frame instance of worker
if isinstance(_locals['self'], WorkerProcess):
# Get queue to add exception
_rslt_q = getattr(_locals['self'], '_rslt_q')
if not _rslt_q:
raise ValueError("No Queue found.")
_rslt_q.put({"undefined_var": arg[3].message}, interceptor=True)
python类AnsibleUndefinedVariable()的实例源码
def is_template(self, data):
''' lets us know if data has a template'''
if isinstance(data, string_types):
try:
new = self.do_template(data, fail_on_undefined=True)
except (AnsibleUndefinedVariable, UndefinedError):
return True
except:
return False
return (new != data)
elif isinstance(data, (list, tuple)):
for v in data:
if self.is_template(v):
return True
elif isinstance(data, dict):
for k in data:
if self.is_template(k) or self.is_template(data[k]):
return True
return False
def aspects(self):
"""Get aspects"""
return {
AnsibleUndefinedVariable: {
r'__init__': dict(
around_after=queue_exc
)},
StrategyBase: {
r'\brun\b': dict(
before=extract_worker_exc(self.errors) # bind data store
)},
TaskQueueManager: {
r'\bsend_callback\b': dict(before=extract_setup_failures(self.errors))
}
}
def _lookup(self, name, *args, **kwargs):
instance = self._lookup_loader.get(name.lower(), loader=self._loader, templar=self)
if instance is not None:
wantlist = kwargs.pop('wantlist', False)
from ansible.utils.listify import listify_lookup_plugin_terms
loop_terms = listify_lookup_plugin_terms(terms=args, templar=self, loader=self._loader, fail_on_undefined=True, convert_bare=False)
# safely catch run failures per #5059
try:
ran = instance.run(loop_terms, variables=self._available_variables, **kwargs)
except (AnsibleUndefinedVariable, UndefinedError) as e:
raise AnsibleUndefinedVariable(e)
except Exception as e:
if self._fail_on_lookup_errors:
raise
ran = None
if ran:
from ansible.vars.unsafe_proxy import UnsafeProxy, wrap_var
if wantlist:
ran = wrap_var(ran)
else:
try:
ran = UnsafeProxy(",".join(ran))
except TypeError:
if isinstance(ran, list) and len(ran) == 1:
ran = wrap_var(ran[0])
else:
ran = wrap_var(ran)
return ran
else:
raise AnsibleError("lookup plugin (%s) not found" % name)
def _lookup_variables(self, terms, variables):
results = []
for x in terms:
try:
intermediate = listify_lookup_plugin_terms(x, templar=self._templar, loader=self._loader, fail_on_undefined=True)
except UndefinedError as e:
raise AnsibleUndefinedVariable("One of the nested variables was undefined. The error was: %s" % e)
results.append(intermediate)
return results
def run(self, terms, variables, **kwargs):
if len(terms) <= 1:
raise AnsibleUndefinedVariable("Mergedict takes 2 options")
if not (isinstance(terms[0], dict) or isinstance(terms[0], list)):
raise AnsibleUndefinedVariable("First option must be a dict or list")
if not isinstance(terms[1], dict):
raise AnsibleUndefinedVariable("Second option must be a dict")
result=[]
temp={}
to_data = terms[0]
from_dict = terms[1]
for item in to_data:
if isinstance(to_data, list):
temp = self.dict_merge(from_dict.copy(), item)
if isinstance(to_data, dict):
if len(terms) == 3:
if isinstance(terms[2], list):
if item not in terms[2]:
continue
else:
raise AnsibleUndefinedVariable("Third option must be a list")
temp = self.dict_merge(from_dict.copy(), to_data[item])
temp['key'] = item
result.append(temp)
return result
def _post_validate_environment(self, attr, value, templar):
'''
Override post validation of vars on the play, as we don't want to
template these too early.
'''
env = {}
if value is not None:
def _parse_env_kv(k, v):
try:
env[k] = templar.template(v, convert_bare=False)
except AnsibleUndefinedVariable as e:
if self.action in ('setup', 'gather_facts') and 'ansible_env' in to_native(e):
# ignore as fact gathering sets ansible_env
pass
if isinstance(value, list):
for env_item in value:
if isinstance(env_item, dict):
for k in env_item:
_parse_env_kv(k, env_item[k])
else:
isdict = templar.template(env_item, convert_bare=False)
if isinstance(isdict, dict):
env.update(isdict)
else:
display.warning("could not parse environment value, skipping: %s" % value)
elif isinstance(value, dict):
# should not really happen
env = dict()
for env_item in value:
_parse_env_kv(env_item, value[env_item])
else:
# at this point it should be a simple string, also should not happen
env = templar.template(value, convert_bare=False)
return env
def _lookup(self, name, *args, **kwargs):
instance = self._lookup_loader.get(name.lower(), loader=self._loader, templar=self)
if instance is not None:
wantlist = kwargs.pop('wantlist', False)
allow_unsafe = kwargs.pop('allow_unsafe', C.DEFAULT_ALLOW_UNSAFE_LOOKUPS)
from ansible.utils.listify import listify_lookup_plugin_terms
loop_terms = listify_lookup_plugin_terms(terms=args, templar=self, loader=self._loader, fail_on_undefined=True, convert_bare=False)
# safely catch run failures per #5059
try:
ran = instance.run(loop_terms, variables=self._available_variables, **kwargs)
except (AnsibleUndefinedVariable, UndefinedError) as e:
raise AnsibleUndefinedVariable(e)
except Exception as e:
if self._fail_on_lookup_errors:
raise AnsibleError("An unhandled exception occurred while running the lookup plugin '%s'. Error was a %s, "
"original message: %s" % (name, type(e), e))
ran = None
if ran and not allow_unsafe:
if wantlist:
ran = wrap_var(ran)
else:
try:
ran = UnsafeProxy(",".join(ran))
except TypeError:
if isinstance(ran, list) and len(ran) == 1:
ran = wrap_var(ran[0])
else:
ran = wrap_var(ran)
if self.cur_context:
self.cur_context.unsafe = True
return ran
else:
raise AnsibleError("lookup plugin (%s) not found" % name)
def _lookup_variables(self, terms, variables):
results = []
for x in terms:
try:
intermediate = listify_lookup_plugin_terms(x, templar=self._templar, loader=self._loader, fail_on_undefined=True)
except UndefinedError as e:
raise AnsibleUndefinedVariable("One of the nested variables was undefined. The error was: %s" % e)
results.append(intermediate)
return results
def _check_conditional(self, conditional, templar, all_vars):
'''
This method does the low-level evaluation of each conditional
set on this object, using jinja2 to wrap the conditionals for
evaluation.
'''
original = conditional
if conditional is None or conditional == '':
return True
if conditional in all_vars and '-' not in text_type(all_vars[conditional]):
conditional = all_vars[conditional]
# make sure the templar is using the variables specified with this method
templar.set_available_variables(variables=all_vars)
try:
conditional = templar.template(conditional)
if not isinstance(conditional, text_type) or conditional == "":
return conditional
# a Jinja2 evaluation that results in something Python can eval!
presented = "{%% if %s %%} True {%% else %%} False {%% endif %%}" % conditional
conditional = templar.template(presented)
val = conditional.strip()
if val == "True":
return True
elif val == "False":
return False
else:
raise AnsibleError("unable to evaluate conditional: %s" % original)
except (AnsibleUndefinedVariable, UndefinedError) as e:
# the templating failed, meaning most likely a
# variable was undefined. If we happened to be
# looking for an undefined variable, return True,
# otherwise fail
if "is undefined" in original:
return True
elif "is defined" in original:
return False
else:
raise AnsibleError("error while evaluating conditional (%s): %s" % (original, e))
def run(self, terms, variables, **kwargs):
anydict = False
skip = False
for term in terms:
if isinstance(term, dict):
anydict = True
total_search = []
if anydict:
for term in terms:
if isinstance(term, dict):
files = term.get('files', [])
paths = term.get('paths', [])
skip = boolean(term.get('skip', False))
filelist = files
if isinstance(files, string_types):
files = files.replace(',', ' ')
files = files.replace(';', ' ')
filelist = files.split(' ')
pathlist = paths
if paths:
if isinstance(paths, string_types):
paths = paths.replace(',', ' ')
paths = paths.replace(':', ' ')
paths = paths.replace(';', ' ')
pathlist = paths.split(' ')
if not pathlist:
total_search = filelist
else:
for path in pathlist:
for fn in filelist:
f = os.path.join(path, fn)
total_search.append(f)
else:
total_search.append(term)
else:
total_search = self._flatten(terms)
for fn in total_search:
try:
fn = self._templar.template(fn)
except (AnsibleUndefinedVariable, UndefinedError):
continue
# get subdir if set by task executor, default to files otherwise
subdir = getattr(self, '_subdir', 'files')
path = None
path = self.find_file_in_search_path(variables, subdir, fn, ignore_missing=True)
if path is not None:
return [path]
else:
if skip:
return []
else:
raise AnsibleLookupError("No file was found when using with_first_found. Use the 'skip: true' option to allow this task to be skipped if no files are found")
def run(self, tmp=None, task_vars=None):
if task_vars is None:
task_vars = dict()
for arg in self._task.args:
if arg not in self.VALID_ARGS:
return {"failed": True, "msg": "'%s' is not a valid option in debug" % arg}
if 'msg' in self._task.args and 'var' in self._task.args:
return {"failed": True, "msg": "'msg' and 'var' are incompatible options"}
result = super(ActionModule, self).run(tmp, task_vars)
verbosity = 0
# get task verbosity
if 'verbosity' in self._task.args:
verbosity = int(self._task.args['verbosity'])
if verbosity <= self._display.verbosity:
if 'msg' in self._task.args:
result['msg'] = self._task.args['msg']
elif 'var' in self._task.args:
try:
results = self._templar.template(self._task.args['var'], convert_bare=True, fail_on_undefined=True, bare_deprecated=False)
if results == self._task.args['var']:
# if results is not str/unicode type, raise an exception
if not isinstance(results, string_types):
raise AnsibleUndefinedVariable
# If var name is same as result, try to template it
results = self._templar.template("{{" + results + "}}", convert_bare=True, fail_on_undefined=True)
except AnsibleUndefinedVariable:
results = "VARIABLE IS NOT DEFINED!"
if isinstance(self._task.args['var'], (list, dict)):
# If var is a list or dict, use the type as key to display
result[to_text(type(self._task.args['var']))] = results
else:
result[self._task.args['var']] = results
else:
result['msg'] = 'Hello world!'
# force flag to make debug output module always verbose
result['_ansible_verbose_always'] = True
else:
result['skipped_reason'] = "Verbosity threshold not met."
result['skipped'] = True
return result
def run(self, terms, variables, **kwargs):
anydict = False
skip = False
for term in terms:
if isinstance(term, dict):
anydict = True
total_search = []
if anydict:
for term in terms:
if isinstance(term, dict):
files = term.get('files', [])
paths = term.get('paths', [])
skip = boolean(term.get('skip', False), strict=False)
filelist = files
if isinstance(files, string_types):
files = files.replace(',', ' ')
files = files.replace(';', ' ')
filelist = files.split(' ')
pathlist = paths
if paths:
if isinstance(paths, string_types):
paths = paths.replace(',', ' ')
paths = paths.replace(':', ' ')
paths = paths.replace(';', ' ')
pathlist = paths.split(' ')
if not pathlist:
total_search = filelist
else:
for path in pathlist:
for fn in filelist:
f = os.path.join(path, fn)
total_search.append(f)
else:
total_search.append(term)
else:
total_search = self._flatten(terms)
for fn in total_search:
try:
fn = self._templar.template(fn)
except (AnsibleUndefinedVariable, UndefinedError):
continue
# get subdir if set by task executor, default to files otherwise
subdir = getattr(self, '_subdir', 'files')
path = None
path = self.find_file_in_search_path(variables, subdir, fn, ignore_missing=True)
if path is not None:
return [path]
else:
if skip:
return []
else:
raise AnsibleLookupError("No file was found when using with_first_found. Use the 'skip: true' option to allow this task to be skipped if no "
"files are found")
