Ansible¶

Ansible Action Plugin micro framework¶

Requires ansible, to build custom action plugins

Action plugins are generally preferable to Module plugins:

they execute on the Ansible controller
which is faster
they can still call any module on the target host
they can stream output (logs) in real-time

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

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()

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(arg='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(arg='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¶

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:

from pathlib import Path
import os
import pytest
import sys
import yaml


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']


def test_playbook(playbook):
    tasks = yaml.safe_load("""
    - name: Create object
      register: object_create
      yourlabs.test.restful_api:
        name: test-cli2-ansible
        capacity: 10To
        price: 1

    - name: Update object
      yourlabs.test.restful_api:
        id: '{{ object_create.json.id }}'
        name: test-cli2-ansible
        capacity: 1To
        price: 2

    - name: Idempotence
      register: idempotence
      yourlabs.test.restful_api:
        id: '{{ object_create.json.id }}'
        name: test-cli2-ansible
        capacity: 1To
        price: 2

    - assert:
        that:
        - not idempotence.changed

    - name: Delete object
      yourlabs.test.restful_api:
        id: '{{ object_create.json.id }}'
        name: test-cli2-ansible
        capacity: 1To
        price: 2
        state: absent
    """)
    playbook.tasks += tasks
    result = playbook()

Variables reader¶

Ansible variables file reader with vault support

Why not use the Ansible Python API? We don’t have a lot to do here, and the CLI are less likely to be subject to changes.

class cli2.ansible.variables.Variables(root_path=None, pass_path=None)[source]¶

Ansible variables reader.

In general, it should be instanciated with root_path and pass_path to fully function correctly.

Example:

import cli2.ansible
variables = cli2.ansible.Variables(
    root_path=Path(__file__).parent,
    pass_path='~/.vault_password',
)
print(variables['playbooks/vars/example.yml'])

Every file read is cached in the variables object.

root_path¶: Unless you feed this with only absolute path, you’ll need a root_path so that relative paths can be resolved. This should be your collection root.

pass_path¶: Unless you don’t use ansible-vault, you’ll need to give the pass to the vault password here.

read(path)[source]¶

Read an ansible YAML variable file.

Parameters:: path – Absolute path or path relative to root_path

class cli2.ansible.variables.Vault[source]¶

classmethod from_yaml(loader, node)[source]¶: Convert a representation node to a Python object.

Playbook generator¶

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.

While the internal Python API of Ansible would also work, this uses Ansible public API which is less subject to change.

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.

API¶

Experimental: my base class for Ansible actions.

class cli2.ansible.action.ActionBase(*args, **kwargs)[source]¶

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')[source]¶

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')[source]¶

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()[source]¶

Return a client instance.

Raises:: NotImplementedError – By default

run(tmp=None, task_vars=None)[source]¶

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()[source]¶

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)[source]¶

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[source]¶

exception cli2.ansible.action.AnsibleOptionError(option)[source]¶

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

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)[source]¶

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)[source]¶

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)[source]¶

Add a module call

Parameters:

module – Name of the Ansible module
args – Ansible module args
kwargs – Task kwargs (register, etc)