Tutorial for cli2.ansible: Ansible Action Plugin framework¶

Experimental feature, requires ansible.

Features¶

Async¶

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

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
        self.result['failed'] = True

Option¶

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

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

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

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, 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.task_vars['ansible_verbosity'] > 1:
            # 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 > 1:
                self.after_set(obj.data)

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

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.

Example¶

Plugin¶

import cli2
import copy
from cli2 import ansible
from cli2.example_client import APIClient


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

    async def run_async(self):
        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 self.state == 'absent':
            if obj:
                response = await obj.delete()
                key, value = self.client.response_log_data(response)
                self.result[key] = value
                self.result['changed'] = True
            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()
            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',
            ),
        )],
    )
    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']

    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