Using custom lookup plugins inside Testinfra

[floot]

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!

[philpep]

This looks related to https://github.com/philpep/testinfra/issues/345 where I suggested to add a playbook argument to ansible.get_variables(). Not sure how it will be complex to handle lookup plugins here too. Unfortunately I don't have much time to work on this :/

[guillaumewatteeux]

Works with workarround

Use a specific ansible cfg for testinfra: ansible.testinfra.cfg with lookup_plugins params

...
[defaults]
lookup_plugins = roles/<role>/lookup_plugins
...

And run:

ANSIBLE_CONFIG=ansible.testinfra.cfg py.test testinfra/ ...