Using custom lookup plugins inside Testinfra #556
Description
Hi,
Inside our Testinfra test suite, we would like to make checks against values that make calls to custom Ansible lookup plugins.
The fact is, I could not find a way of enabling a custom lookup plugin in Testinfra like we already do in our playbooks.
What we usually do in our playbooks looks like this:
# This enables the lookup plugins contained in a dedicated Ansible role
- hosts: localhost
connection: local
gather_facts: False
roles:
- role: role_containing_custom_lookup_plugin
tags:
- always
# This is the role that performs the actual work
- hosts: all
gather_facts: True
become: True
roles:
- role: actual_role...so that when a variable with the form foo: {{ lookup('custom_lookup_plugin', param1, param2 }} is used in actual_role, it is indeed evaluated by calling the lookup plugin.
Then when I use inventory's host/group vars containing such vars in Testinfra, I write a Pytest fixture like this one:
from ansible.template import Templar
from ansible.parsing.dataloader import DataLoader
@pytest.fixture(scope='module')
def ansible_vars(host):
# Load the role containing the custom lookup plugin
testinfra.get_host('ansible://localhost').ansible('include_role', 'name=role_containing_custom_lookup_plugin')
# Load host vars
host_vars = host.ansible.get_variables()['ansible_facts']
# Create a templar with the host vars as a context
templar = Templar(loader=DataLoader(), variables=host_vars)
# Evaluate the host vars
return templar.template(host_vars, fail_on_undefined=False)But when I use this fixture in a test method, it just fails with finding my lookup plugin:
def test_foobar(host, ansible_vars):
> expanded_vars = ansible_vars
tests/test_custom.py:64:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/home/user/venvs/ansible/lib/python3.7/site-packages/ansible/template/__init__.py:627: in template
disable_lookups=disable_lookups,
/home/user/venvs/ansible/lib/python3.7/site-packages/ansible/template/__init__.py:582: in template
disable_lookups=disable_lookups,
/home/user/venvs/ansible/lib/python3.7/site-packages/ansible/template/__init__.py:841: in do_template
res = j2_concat(rf)
<template>:13: in root
???
/home/user/venvs/ansible/lib/python3.7/site-packages/jinja2/runtime.py:290: in call
return __obj(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <ansible.template.Templar object at 0x7f3c347a9ed0>
name = 'custom_lookup_plugin', args = ('var_name',)
kwargs = {'client_id': AnsibleUndefined, 'secret': AnsibleUndefined, 'tenant_id': AnsibleUndefined, 'vault_url': AnsibleUndefined}
instance = None
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)
errors = kwargs.pop('errors', 'strict')
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:
msg = u"An unhandled exception occurred while running the lookup plugin '%s'. Error was a %s, original message: %s" % \
(name, type(e), to_text(e))
if errors == 'warn':
display.warning(msg)
elif errors == 'ignore':
display.display(msg, log_only=True)
else:
raise AnsibleError(to_native(msg))
ran = [] if wantlist else None
if ran and not allow_unsafe:
if wantlist:
ran = wrap_var(ran)
else:
try:
ran = wrap_var(",".join(ran))
except TypeError:
# Lookup Plugins should always return lists. Throw an error if that's not
# the case:
if not isinstance(ran, Sequence):
raise AnsibleError("The lookup plugin '%s' did not return a list."
% name)
# The TypeError we can recover from is when the value *inside* of the list
# is not a string
if 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)
E ansible.errors.AnsibleError: lookup plugin (custom_lookup_plugin) not found
So my question is: how can I use a custom Ansible lookup plugin properly in Testinfra, when it is provided/loaded/activated by a (third-party) role?
Thanks!