Ansible Action Plugin

Experimental feature, requires ansible, to build custom action plugins

Useful on its own, it also integrates very well with the rest of our stuff:

• Display

• Logging

• HTTP client

• Lock

Features

Async

With ActionBase, we don’t define run, we define run_async()

import cli2
from cli2 import ansible

class ActionModule(ansible.ActionBase):
    async def run_async(self):
        self.tmp        # is the usual tmp arg
        self.task_vars  # is the usual task_vars arg
        cli2.log.debug('Getting something')
        self.result['failed'] = True
        cli2.log.info('Got something', json=something)

And then you get absolutely beautiful logging:

• json logger key is configured to render as colored yaml

• with -v: log level will be set to INFO, which you should you use to indicate that something has been done

• with -vv: log level will be set to DEBUG, which you should you use to indicate that something is going to be attempted

Without -v:

With -v:

With -vv:

Option

With Option, we can declare task options instead of fiddling with task_vars:

class ActionModule(ansible.ActionBase):
    # an option with a default
    name = ansible.Option('name', default='test')

    # an option without default: if not specified, the module will fail and
    # request that the user configures a value for this option
    nodefault = ansible.Option('nodefault')

    # option that takes value from a fact
    global_option = ansible.Option(fact='your_fact')

    # an option that takes value from a task arg if available, otherwise
    # from a fact, otherwise default
    region = ansible.Option('region', fact='region', default='EU')

    async def run_async(self):
        self.result['name'] = self.name
        self.result['region'] = self.region
        # ...

An option can specify an argument name, and/or a global fact name, and a default.

Client

You need to return a Client instance in the client_factory() method to have a self.client attribute:

class ActionModule(ansible.ActionBase):
    name = ansible.Option('name', default='test-name')
    value = ansible.Option('value', default='test-value')

    async def client_factory(self):
        return YourClient()

    async def run_async(self):
        obj = await self.client.YourObject.find(name=self.name).first()
        # ....

And then you get absolutely beautiful logging:

• with -v: log level will be set to INFO, which means you will see responses.

• with -vv: log level will be set to DEBUG, which means you will see requests too.

There is no way to set DEBUG from ansible, as we never want masked secrets to output in an Ansible Tower job. But you can still export DEBUG=1 prior to executing Ansible manually, which will dump all pagination requests/responses and secrets.

Without -v:

With -v:

With -vv:

Diff

We’re going to be changing stuff, and Ansible doesn’t interpret before/after result keys which means it won’t dump a diff even with --diff.

Instead of calling diff_data() manually, you can call before_set() and after_set(), then a diff will be displayed automatically.

class ActionModule(ansible.ActionBase):
    name = ansible.Option('name', default='test-name')
    value = ansible.Option('value', default='test-value')

    async def client_factory(self):
        return YourClient()

    async def run_async(self):
        obj = await self.client.Object.find(name=self.name).first()

        if self.verbosity:
            # don't display diff if not -v
            self.before_set(obj.data)

        obj.value = self.value

        if obj.changed_fields:
            response = await obj.save()
            self.result['changed'] = True

            if self.verbosity:
                self.after_set(obj.data)

            # we can also get masked data
            key, value = self.client.response_log_data(response)
            self.result[key] = value

Also, I note that I always forget to pass --diff anyway, so do my users, I’m assuming the user is trying to understand what’s going on as soon as they pass a single -v, so, this example will only check if any verbosity is activated at all to display the diff.

If you really want the diff to display only with --diff, then wrap your before_set/after_set in if self.task_vars['ansible_diff_mode'] instead of if self.verbosity.

Testing

Mock

You can run the module in mocked mode in tests with the run_test_async() method:

@pytest.mark.asyncio
async def test_module():
    module = await your.ActionModule.run_test_async(
        args=dict(
            name='test',
            capacity='5',
            price='3',
        )
    )
    assert module.result['changed']

For HTTP response mocking, you should use httpx_mock from pytest-httpx, as seen in the example below.

Subprocess

You can also create playbooks on the fly and run them in a subprocess that calls ansible-playbook in localhost, thanks to the Playbook fixture:

def test_playbook_exec(playbook):
    playbook.task_add('debug', args=dict(msg='hello'))
    result = playbook()
    assert result['changed'] == 0
    assert result['ok'] == 2

The previous, mocking solution, is always preferable. But if you also want functional tests, then this works great.

Documenting

That’s a bit trickier, you have to put a module plugin with a name matching your action plugin and set the documentation in YAML strings in there.

Once your documentation outputs properly with ansible-doc command, you can have it in your Sphinx documentation with various plugins that you’ll find easily on internet, except probably for mine: ansible-sphinx.

Example

Plugin

This uses the example cli2 Client that we developed in the cli2.client example:

import cli2
import copy
from cli2 import ansible
from cli2.examples.client import APIClient


class ActionModule(ansible.ActionBase):
    id = ansible.Option('id', default=None)
    name = ansible.Option('name')
    capacity = ansible.Option('capacity', default='1To')
    price = ansible.Option('price')
    state = ansible.Option('state', default='present')

    async def run_async(self):
        self.log = cli2.log.bind(id=self.id, name=self.name)
        obj = None

        if self.id:
            obj = await self.client.Object.get(id=self.id)
        elif self.name:
            obj = await self.client.Object.find(name=self.name).first()

        if obj:
            self.log.info(f'Found object')

        if self.state == 'absent':
            if obj:
                response = await obj.delete()
                # this returns masked json or content in value
                # key will be "json" or "content"
                key, value = self.client.response_log_data(response)
                self.result[key] = value
                self.result['changed'] = True
                self.log.info(f'Deleted object')
            return

        if obj is not None:
            if self.verbosity:
                # will cause a diff to display
                self.before_set(obj.data)
        else:
            obj = self.client.Object()

        obj.name = self.name
        obj.capacity = self.capacity
        obj.price = self.price

        if obj.changed_fields:
            response = await obj.save()
            self.log.info(f'Object changes saved')
            key, value = self.client.response_log_data(response)
            self.result[key] = value
            self.result['changed'] = True
            if self.verbosity:
                # causes a diff to be displayed
                self.after_set(obj.data)
        else:
            self.result['json'] = obj.data_masked
            self.result['changed'] = False

    async def client_factory(self):
        return APIClient()

Tests

from pathlib import Path
import os
import pytest
import sys


plugin_path = Path(__file__).parent.parent / 'tests/yourlabs/test/plugins'
sys.path.insert(0, str(plugin_path))
from action import restful_api  # noqa


# enforce localhost in our client
os.environ['URL'] = 'http://localhost:8000'


@pytest.mark.asyncio
async def test_create(httpx_mock):
    os.environ['URL'] = 'http://localhost:8000'
    httpx_mock.add_response(
        url='http://localhost:8000/objects/?name=test',
        method='GET',
        json=[]
    )
    httpx_mock.add_response(
        url='http://localhost:8000/objects/',
        method='POST',
        json=dict(
            name='test',
            id=1,
            data=dict(
                Capacity='5',
                Price='3',
            ),
        ),
    )
    module = await restful_api.ActionModule.run_test_async(
        args=dict(
            name='test',
            price='3',
            capacity='5',
        )
    )
    assert module.result['json']['id'] == 1


@pytest.mark.asyncio
async def test_update(httpx_mock):
    httpx_mock.add_response(
        is_reusable=True,
        url='http://localhost:8000/objects/?name=test',
        method='GET',
        json=[dict(
            name='test',
            id=1,
            data=dict(
                Capacity='5',
                Price='3',
            ),
        )],
    )

    # testing for idempotence
    module = await restful_api.ActionModule.run_test_async(
        args=dict(
            name='test',
            capacity='5',
            price='3',
        )
    )
    assert not module.result['changed']
    assert module.result['json']['id'] == 1

    # testing for update
    httpx_mock.add_response(
        url='http://localhost:8000/objects/1/',
        method='PUT',
        json=dict(
            name='test',
            id=1,
            data=dict(
                Capacity='6',
                Price='4',
            ),
        ),
    )

    module = await restful_api.ActionModule.run_test_async(
        args=dict(
            name='test',
            capacity='4',
            price='6',
        )
    )
    assert module.result['json']['id'] == 1
    assert module.result['changed']

API

Experimental: my base class for Ansible actions.

class cli2.ansible.action.ActionBase(*args, **kwargs)

Base action class

result

Result dict that will be returned to Ansible

task_vars

The task_vars that the module was called with

client

The client object generated by client_factory() if you implement it.

after_set(data, label='after')

Set the data we’re going to display the diff for at the end.

Parameters:

• data – Dictionnary of data

• label – Label to show in diff

before_set(data, label='before')

Set the data we’re going to display the diff for at the end.

Parameters:

• data – Dictionnary of data

• label – Label to show in diff

async client_factory()

Return a client instance.

Raises:

NotImplementedError – By default

run(tmp=None, task_vars=None)

Action Plugins should implement this method to perform their tasks. Everything else in this base class is a helper method for the action plugin to do that.

Parameters:

• tmp – Deprecated parameter. This is no longer used. An action plugin that calls another one and wants to use the same remote tmp for both should set self._connection._shell.tmpdir rather than this parameter.

• task_vars – The variables (host vars, group vars, config vars, etc) associated with this task.

Returns:

dictionary of results from the module

Implementers of action modules may find the following variables especially useful:

• Module parameters. These are stored in self._task.args

async run_async()

The method you are supposed to implement.

It should:

• provision the result dict

• find task_vars in task_vars

async classmethod run_test_async(args=None, facts=None, client=None, fail=False)

Test run the module in a mocked context.

Parameters:

• args – Dict of task arguments

• facts – Dict of play facts

• client – Client instance, overrides the factory

• fail – Allow this test to fail without exception

exception cli2.ansible.action.AnsibleError

exception cli2.ansible.action.AnsibleOptionError(option)

class cli2.ansible.action.Option(arg=None, fact=None, default='__UNSET__DEFAULT__')

Ansible Option descriptor.

arg

Name of the task argument to get this option value from

fact

Name of the fact, if any, to get a value for this option if no task arg is provided

default

Default value, if any, in case neither of arg and fact were defined.

class cli2.ansible.playbook.Playbook(root, name)

On-the-fly playbook generator

root

This would be a tmp_path returned by pytest

name

Name of the playbook, test name by default

vars

Playbook vars

roles

Playbook roles, use role_add() to add a role

tasks

Playbook tasks, use task_add() to add a task

play

Main playbook play

plays

Playbook plays, contains the main one by default

yaml

Property that returns the generated yaml

role_add(name, *tasks, **variables)

Create a new role with given tasks, include it with given variables

Parameters:

• name – role name

• tasks – List of task dicts

• variables – Variables that will be passed to include_role

task_add(module, args=None, **kwargs)

Add a module call

Parameters:

• module – Name of the Ansible module

• args – Ansible module args

• kwargs – Task kwargs (register, etc)