Ansible Framework¶

Installation:

pip install cansible # or cli2[ansible]

Then:

import cansible

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:

Action plugin¶

Example Plugin¶

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

import cli2
import copy
import cansible
from chttpx.example import APIClient


class ActionModule(cansible.ActionBase):
    id = cansible.Option('id', default=None)
    name = cansible.Option('name')
    capacity = cansible.Option('capacity', default='1To')
    price = cansible.Option('price')
    state = cansible.Option('state', default='present')
    mask_keys = ['Price']

    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:
                self.result['deleted'] = obj.data
                await obj.delete()
                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')
            self.result['data'] = response.json()
            self.result['changed'] = True
            if self.verbosity:
                # causes a diff to be displayed
                self.after_set(obj.data)
        else:
            self.result['data'] = obj.data
            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(cansible.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(cansible.ActionBase):
    # an option with a default
    name = cansible.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 = cansible.Option('nodefault')

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

    # an option that takes value from a task arg if available, otherwise
    # from a fact, otherwise default
    region = cansible.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.

Fact update¶

Facts can be updated at runtime:

class ActionModule(cansible.ActionBase):
    stuff = cansible.Option(fact='stuff', default=None)

    async def run_async(self):
        self.stuff = 'foo'
        # the above will cause result to contain an ansible_facts dict of
        # stuff=foo, therefore causing ansible to update the stuff fact

This only works with options which have a fact, and works with mutable values.

Client¶

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

class ActionModule(cansible.ActionBase):
    name = cansible.Option('name', default='test-name')
    value = cansible.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(cansible.ActionBase):
    name = cansible.Option('name', default='test-name')
    value = cansible.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.

Secret masking¶

Until we get Data Tagging, we use the Mask class, which can learn values to mask on the fly.

The first thing the plugin does in mask_init(), is collecting values and keys to mask from various sources:

Class mask_keys attribute: list of keys to mask, hardcoded by yourself in your ActionModule.
Keys can also be set from the playbook in the mask_keys ansible fact.
Values can also be set from the playbook in the mask_values ansible fact.
If you have set client’s then it’s mask object will be used.

When this feature is used then you can use no_log: true and still This allows to use no_log: true and still have an output of the result.

Setting a module level mask:

class ActionModule(cansible.ActionBase):
    mask_keys = ['password']

Marking a variable value as masked:

- set_fact:
    mask_keys:
    - your_password
    mask_values:
    - '{{ some.password }}'

Will cause any occurence of the values of any password or whatever some.password contains to be replaced with ***MASKED***. by

print_yaml() also masks by default so that you can dump any dict safely in there.

We’re not shipping a collection, so it’s complicated to ship modules from a pip package, but you can make a shell module that will use masking:

class ActionModule(cansible.ActionBase):
    cmd = cansible.Option('cmd')

    async def run_async(self):
        self.result.update(self.subprocess_remote(self.cmd))

Where subprocess_remote() is a basic helper function we provide to run commands over the target host, with fully supported masking.

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


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['data']['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['data']['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['data']['id'] == 1
    assert module.result['changed']

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.

exception cansible.variables.AnsibleVariablesError[source]¶

exception cansible.variables.PathNotFoundError[source]¶

exception cansible.variables.UnresolvablePathError[source]¶

class cansible.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 cansible
variables = cansible.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 cansible.variables.Vault[source]¶

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

exception cansible.variables.VaultPasswordFileNotFoundError[source]¶

exception cansible.variables.VaultPasswordFileRequiredError[source]¶

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 when possible, because it doesn’t depend on the network to succeed. But if your action plugin is more complicated than that 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, and emulates exactly what a user would do with your plugin.

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¶

Base class for Ansible Actions.

class cansible.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.

masked_keys¶

Declare a list of keys to mask:

class AnsibleModule(ansible.ActionBase):
    masked_keys = ['secret', 'password']

mask¶: Mask object

action_run(name, **kwargs)[source]¶

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

print(data, mask=True)[source]¶: Print that masks by default.

print_yaml(data)[source]¶: Render data as masked yaml.

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

subprocess_remote(cmd, callback=None, print=True, **kwargs)[source]¶

Execute a shell command on the remote in a masked context

Parameters:

cmd – Command to run
kwargs – Other shell args, such as creates etc
print – Wether to print stdout/stderr, defaults to True
callback – Callback function called before output is printed and returned, use that to collect secrets and provision self.mask_values.

exception cansible.action.AnsibleError[source]¶

exception cansible.action.AnsibleOptionError(option)[source]¶

class cansible.action.Option(arg=None, fact=None, default='__UNSET__DEFAULT__', cast_in=None, cast_out=None)[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 cansible.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)