a-c-m
/
ansible.utils
mirror of https://github.com/ansible-collections/ansible.utils.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

update_fact plugin + units (#10) 

* update_fact plugin + units

* Update README, add doc

* Add some update_fact integration tests

* Add missing test for not (index_of)

* Update fact doc updates

* Update rst from doc

* Examples as yaml

* Some developer notes (#11)

Co-authored-by: cidrblock <brad@thethorntons.net>

* Updated self._task.args from updated_data provided back from aav

* Argspec default fix (#12)

* Return data updated with default values from aav.validate()

* Update aav docs

Co-authored-by: cidrblock <brad@thethorntons.net>

* update_fact plugin + units

* Update README, add doc

* Add some update_fact integration tests

* Add missing test for not (index_of)

* Update fact doc updates

* Update rst from doc

* Examples as yaml

* Updated self._task.args from updated_data provided back from aav

* recheck

* Plugin cleanup (#14)

* WIP

* Add argspec validation to plugins, restructure tests

* Update docs

* Pre PY3.8 compat changes

* Run black to fix quotes

* Seems the order of missing keys varies between 2.9 and 2.10

* More error ordering issues fixed during argspec validation

* More black, wrong version

Co-authored-by: cidrblock <brad@thethorntons.net>

* update_fact plugin + units

* Update README, add doc

* Add some update_fact integration tests

* Update fact doc updates

* Update rst from doc

* Examples as yaml

* Updated self._task.args from updated_data provided back from aav

* recheck

Co-authored-by: cidrblock <brad@thethorntons.net>
pull/16/head
Bradley A. Thornton 2020-10-20 12:32:35 -07:00 committed by GitHub
parent bcddde229d
commit 188463f7be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 1376 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@ -32,6 +32,11 @@ Name | Description
[ansible.utils.index_of](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.index_of_lookup.rst)|Find the indicies of items in a list matching some criteria
[ansible.utils.to_paths](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.to_paths_lookup.rst)|Flatten a complex object into a dictionary of paths and values
### Modules
Name | Description
--- | ---
[ansible.utils.update_fact](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.update_fact_module.rst)|Update currently set facts
<!--end collection content-->
## Installing this collection

405
docs/ansible.utils.update_fact_module.rst Normal file
View File

@ -0,0 +1,405 @@
.. _ansible.utils.update_fact_module:
*************************
ansible.utils.update_fact
*************************
**Update currently set facts**
Version added: 1.0.0
.. contents::
:local:
:depth: 1
Synopsis
--------
- This module allows updating existing variables.
- Variables are updated on a host-by-host basis.
- Variable are not modified in place, instead they are returned by the module
Parameters
----------
.. raw:: html
<table border=0 cellpadding=0 class="documentation-table">
<tr>
<th colspan="2">Parameter</th>
<th>Choices/<font color="blue">Defaults</font></th>
<th width="100%">Comments</th>
</tr>
<tr>
<td colspan="2">
<div class="ansibleOptionAnchor" id="parameter-"></div>
<b>updates</b>
<a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
<div style="font-size: small">
<span style="color: purple">list</span>
/ <span style="color: purple">elements=dictionary</span>
/ <span style="color: red">required</span>
</div>
</td>
<td>
</td>
<td>
<div>A list of dictionaries, each a desired update to make</div>
</td>
</tr>
<tr>
<td class="elbow-placeholder"></td>
<td colspan="1">
<div class="ansibleOptionAnchor" id="parameter-"></div>
<b>path</b>
<a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
<div style="font-size: small">
<span style="color: purple">string</span>
/ <span style="color: red">required</span>
</div>
</td>
<td>
</td>
<td>
<div>The path in a currently set variable to update</div>
<div>The path can be in dot or bracket notation</div>
<div>It should be a valid jinja reference</div>
</td>
</tr>
<tr>
<td class="elbow-placeholder"></td>
<td colspan="1">
<div class="ansibleOptionAnchor" id="parameter-"></div>
<b>value</b>
<a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
<div style="font-size: small">
<span style="color: purple">raw</span>
/ <span style="color: red">required</span>
</div>
</td>
<td>
</td>
<td>
<div>The value to be set at the path</div>
<div>Can be a simple or complex data structure</div>
</td>
</tr>
</table>
<br/>
Examples
--------
.. code-block:: yaml
# Update an exisitng fact, dot or bracket notation
- name: Set a fact
set_fact:
a:
b:
c:
- 1
- 2
- name: Update the fact
ansible.utils.update_fact:
updates:
- path: a.b.c.0
value: 10
- path: "a['b']['c'][1]"
value: 20
register: updated
- debug:
var: updated.a
# updated:
# a:
# b:
# c:
# - 10
# - 20
# changed: true
# Lists can be appended, new keys added to dictionaries
- name: Set a fact
set_fact:
a:
b:
b1:
- 1
- 2
- name: Update, add to list, add new key
ansible.utils.update_fact:
updates:
- path: a.b.b1.2
value: 3
- path: a.b.b2
value:
- 10
- 20
- 30
register: updated
- debug:
var: updated.a
# updated:
# a:
# b:
# b1:
# - 1
# - 2
# - 3
# b2:
# - 10
# - 20
# - 30
# changed: true
#####################################################################
# Update every item in a list of dictionaries
# build the update list ahead of time using a loop
# and then apply the changes to the fact
#####################################################################
- name: Set fact
set_fact:
addresses:
- raw: 10.1.1.0/255.255.255.0
name: servers
- raw: 192.168.1.0/255.255.255.0
name: printers
- raw: 8.8.8.8
name: dns
- name: Build a list of updates
set_fact:
update_list: "{{ update_list + update }}"
loop: "{{ addresses }}"
loop_control:
index_var: idx
vars:
update_list: []
update:
- path: addresses[{{ idx }}].network
value: "{{ item['raw'] | ansible.netcommon.ipaddr('network') }}"
- path: addresses[{{ idx }}].prefix
value: "{{ item['raw'] | ansible.netcommon.ipaddr('prefix') }}"
- debug:
var: update_list
# TASK [debug] *******************
# ok: [localhost] =>
# update_list:
# - path: addresses[0].network
# value: 10.1.1.0
# - path: addresses[0].prefix
# value: '24'
# - path: addresses[1].network
# value: 192.168.1.0
# - path: addresses[1].prefix
# value: '24'
# - path: addresses[2].network
# value: 8.8.8.8
# - path: addresses[2].prefix
# value: '32'
- name: Make the updates
ansible.utils.update_fact:
updates: "{{ update_list }}"
register: updated
- debug:
var: updated
# TASK [debug] ***********************
# ok: [localhost] =>
# updated:
# addresses:
# - name: servers
# network: 10.1.1.0
# prefix: '24'
# raw: 10.1.1.0/255.255.255.0
# - name: printers
# network: 192.168.1.0
# prefix: '24'
# raw: 192.168.1.0/255.255.255.0
# - name: dns
# network: 8.8.8.8
# prefix: '32'
# raw: 8.8.8.8
# changed: true
# failed: false
#####################################################################
# Retrieve, update, and apply interface description change
# use index_of to locate Etherent1/1
#####################################################################
- name: Get the current interface config
cisco.nxos.nxos_interfaces:
state: gathered
register: interfaces
- name: Update the description of Ethernet1/1
ansible.utils.update_fact:
updates:
- path: "interfaces.gathered[{{ index }}].description"
value: "Configured by ansible"
vars:
index: "{{ interfaces.gathered|ansible.utils.index_of('eq', 'Ethernet1/1', 'name') }}"
register: updated
- name: Update the configuration
cisco.nxos.nxos_interfaces:
config: "{{ updated.interfaces.gathered }}"
state: overridden
register: result
- name: Show the commands issued
debug:
msg: "{{ result['commands'] }}"
# TASK [Show the commands issued] *************************************
# ok: [nxos101] => {
# "msg": [
# "interface Ethernet1/1",
# "description Configured by ansible"
# ]
# }
#####################################################################
# Retrieve, update, and apply an ipv4 ACL change
# finding the index of AFI ipv4 acls
# finding the index of the ACL named 'test1'
# finding the index of sequence 10
#####################################################################
- name: Retrieve the current acls
arista.eos.eos_acls:
state: gathered
register: current
- name: Update the source of sequenmce 10 in the IPv4 ACL named test1
ansible.utils.update_fact:
updates:
- path: current.gathered[{{ afi }}].acls[{{ acl }}].aces[{{ ace }}].source
value:
subnet_address: "192.168.2.0/24"
vars:
afi: "{{ current.gathered|ansible.utils.index_of('eq', 'ipv4', 'afi') }}"
acl: "{{ current.gathered[afi|int].acls|ansible.utils.index_of('eq', 'test1', 'name') }}"
ace: "{{ current.gathered[afi|int].acls[acl|int].aces|ansible.utils.index_of('eq', 10, 'sequence') }}"
register: updated
- name: Apply the changes
arista.eos.eos_acls:
config: "{{ updated.current.gathered }}"
state: overridden
register: changes
- name: Show the commands issued
debug:
msg: "{{ changes['commands'] }}"
# TASK [Show the commands issued] *************************************
# ok: [eos101] => {
# "msg": [
# "ip access-list test1",
# "no 10",
# "10 permit ip 192.168.2.0/24 host 10.1.1.2"
# ]
# }
#####################################################################
# Disable ip redirects on any layer3 interface
# find the layer 3 interfaces
# use each name to find their index in l3 interface
# build an 'update' list and apply the updates
#####################################################################
- name: Get the current interface and L3 interface configuration
cisco.nxos.nxos_facts:
gather_subset: min
gather_network_resources:
- interfaces
- l3_interfaces
- name: Build the list of updates to make
set_fact:
updates: "{{ updates + [entry] }}"
vars:
updates: []
entry:
path: "ansible_network_resources.l3_interfaces[{{ item }}].redirects"
value: False
w_mode: "{{ ansible_network_resources.interfaces|selectattr('mode', 'defined') }}"
m_l3: "{{ w_mode|selectattr('mode', 'eq', 'layer3') }}"
names: "{{ m_l3|map(attribute='name')|list }}"
l3_indicies: "{{ ansible_network_resources.l3_interfaces|ansible.utils.index_of('in', names, 'name', wantlist=True) }}"
loop: "{{ l3_indicies }}"
# TASK [Build the list of updates to make] ****************************
# ok: [nxos101] => (item=99) => changed=false
# ansible_facts:
# updates:
# - path: ansible_network_resources.l3_interfaces[99].redirects
# value: false
# ansible_loop_var: item
# item: 99
- name: Update the l3 interfaces
ansible.utils.update_fact:
updates: "{{ updates }}"
register: updated
# TASK [Update the l3 interfaces] *************************************
# changed: [nxos101] => changed=true
# ansible_network_resources:
# l3_interfaces:
# <...>
# - ipv4:
# - address: 10.1.1.1/24
# name: Ethernet1/100
# redirects: false
- name: Apply the configuration changes
cisco.nxos.l3_interfaces:
config: "{{ updated.ansible_network_resources.l3_interfaces }}"
state: overridden
register: changes
# TASK [Apply the configuration changes] ******************************
# changed: [nxos101] => changed=true
# commands:
# - interface Ethernet1/100
# - no ip redirects
Status
------
Authors
~~~~~~~
- Bradley Thornton (@cidrblock)

189
plugins/action/update_fact.py Normal file
View File

@ -0,0 +1,189 @@
# -*- coding: utf-8 -*-
# Copyright 2020 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import ast
import json
import re
from ansible.plugins.action import ActionBase
from ansible.errors import AnsibleActionFail
from ansible.module_utils.common._collections_compat import (
MutableMapping,
MutableSequence,
)
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes, to_native
from jinja2 import Template, TemplateSyntaxError
from ansible_collections.ansible.utils.plugins.modules.update_fact import (
DOCUMENTATION,
)
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
AnsibleArgSpecValidator,
)
from ansible.errors import AnsibleActionFail
class ActionModule(ActionBase):
"""action module"""
def __init__(self, *args, **kwargs):
"""Start here"""
super(ActionModule, self).__init__(*args, **kwargs)
self._supports_async = True
self._updates = None
self._result = None
def _check_argspec(self):
aav = AnsibleArgSpecValidator(
data=self._task.args,
schema=DOCUMENTATION,
schema_format="doc",
name=self._task.action,
)
valid, errors, self._task.args = aav.validate()
if not valid:
raise AnsibleActionFail(errors)
def _ensure_valid_jinja(self):
"""Ensure each path is jinja valid"""
errors = []
for entry in self._task.args["updates"]:
try:
Template("{{" + entry["path"] + "}}")
except TemplateSyntaxError as exc:
error = (
"While processing '{path}' found malformed path."
" Ensure syntax follows valid jinja format. The error was:"
" {error}"
).format(path=entry["path"], error=to_native(exc))
errors.append(error)
if errors:
raise AnsibleActionFail(" ".join(errors))
@staticmethod
def _field_split(path):
"""Split the path into it's parts
:param path: The user provided path
:type path: str
:return: the individual parts of the path
:rtype: list
"""
que = list(path)
val = que.pop(0)
fields = []
try:
while True:
field = ""
# found a '.', move to the next character
if val == ".":
val = que.pop(0)
# found a '[', pop until ']' and then get the next
if val == "[":
val = que.pop(0)
while val != "]":
field += val
val = que.pop(0)
val = que.pop(0)
else:
while val not in [".", "["]:
field += val
val = que.pop(0)
try:
# make numbers numbers
fields.append(ast.literal_eval(field))
except Exception:
# or strip the quotes
fields.append(re.sub("['\"]", "", field))
except IndexError:
# pop'ed past the end of the que
# so add the final field
try:
fields.append(ast.literal_eval(field))
except Exception:
fields.append(re.sub("['\"]", "", field))
return fields
def set_value(self, obj, path, val):
"""Set a value
:param obj: The object to modify
:type obj: mutable object
:param path: The path to where the update should be made
:type path: list
:param val: The new value to place at path
:type val: string, dict, list, bool, etc
"""
first, rest = path[0], path[1:]
if rest:
try:
new_obj = obj[first]
except (KeyError, TypeError):
msg = (
"Error: the key '{first}' was not found "
"in {obj}.".format(obj=obj, first=first)
)
raise AnsibleActionFail(msg)
self.set_value(new_obj, rest, val)
else:
if isinstance(obj, MutableMapping):
if obj.get(first) != val:
self._result["changed"] = True
obj[first] = val
elif isinstance(obj, MutableSequence):
if not isinstance(first, int):
msg = (
"Error: {obj} is a list, "
"but index provided was not an integer: '{first}'"
).format(obj=obj, first=first)
raise AnsibleActionFail(msg)
if first > len(obj):
msg = "Error: {obj} not long enough for item #{first} to be set.".format(
obj=obj, first=first
)
raise AnsibleActionFail(msg)
if first == len(obj):
obj.append(val)
self._result["changed"] = True
else:
if obj[first] != val:
obj[first] = val
self._result["changed"] = True
else:
msg = "update_fact can only modify mutable objects."
raise AnsibleActionFail(msg)
def run(self, tmp=None, task_vars=None):
"""action entry point"""
self._task.diff = False
self._result = super(ActionModule, self).run(tmp, task_vars)
self._result["changed"] = False
self._check_argspec()
results = set()
self._ensure_valid_jinja()
for entry in self._task.args["updates"]:
parts = self._field_split(entry["path"])
obj, path = parts[0], parts[1:]
results.add(obj)
if obj not in task_vars["vars"]:
msg = "'{obj}' was not found in the current facts.".format(
obj=obj
)
raise AnsibleActionFail(msg)
retrieved = task_vars["vars"].get(obj)
if path:
self.set_value(retrieved, path, entry["value"])
else:
if task_vars["vars"][obj] != entry["value"]:
task_vars["vars"][obj] = entry["value"]
self._result["changed"] = True
for key in results:
value = task_vars["vars"].get(key)
self._result[key] = value
return self._result

343
plugins/modules/update_fact.py Normal file
View File

@ -0,0 +1,343 @@
# -*- coding: utf-8 -*-
# Copyright 2020 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r"""
---
module: update_fact
short_description: Update currently set facts
version_added: "1.0.0"
description:
- This module allows updating existing variables.
- Variables are updated on a host-by-host basis.
- Variable are not modified in place, instead they are returned by the module
options:
updates:
description:
- A list of dictionaries, each a desired update to make
type: list
elements: dict
required: True
suboptions:
path:
description:
- The path in a currently set variable to update
- The path can be in dot or bracket notation
- It should be a valid jinja reference
type: str
required: True
value:
description:
- The value to be set at the path
- Can be a simple or complex data structure
type: raw
required: True
notes:
author:
- Bradley Thornton (@cidrblock)
"""
EXAMPLES = r"""
# Update an exisitng fact, dot or bracket notation
- name: Set a fact
set_fact:
a:
b:
c:
- 1
- 2
- name: Update the fact
ansible.utils.update_fact:
updates:
- path: a.b.c.0
value: 10
- path: "a['b']['c'][1]"
value: 20
register: updated
- debug:
var: updated.a
# updated:
# a:
# b:
# c:
# - 10
# - 20
# changed: true
# Lists can be appended, new keys added to dictionaries
- name: Set a fact
set_fact:
a:
b:
b1:
- 1
- 2
- name: Update, add to list, add new key
ansible.utils.update_fact:
updates:
- path: a.b.b1.2
value: 3
- path: a.b.b2
value:
- 10
- 20
- 30
register: updated
- debug:
var: updated.a
# updated:
# a:
# b:
# b1:
# - 1
# - 2
# - 3
# b2:
# - 10
# - 20
# - 30
# changed: true
#####################################################################
# Update every item in a list of dictionaries
# build the update list ahead of time using a loop
# and then apply the changes to the fact
#####################################################################
- name: Set fact
set_fact:
addresses:
- raw: 10.1.1.0/255.255.255.0
name: servers
- raw: 192.168.1.0/255.255.255.0
name: printers
- raw: 8.8.8.8
name: dns
- name: Build a list of updates
set_fact:
update_list: "{{ update_list + update }}"
loop: "{{ addresses }}"
loop_control:
index_var: idx
vars:
update_list: []
update:
- path: addresses[{{ idx }}].network
value: "{{ item['raw'] | ansible.netcommon.ipaddr('network') }}"
- path: addresses[{{ idx }}].prefix
value: "{{ item['raw'] | ansible.netcommon.ipaddr('prefix') }}"
- debug:
var: update_list
# TASK [debug] *******************
# ok: [localhost] =>
# update_list:
# - path: addresses[0].network
# value: 10.1.1.0
# - path: addresses[0].prefix
# value: '24'
# - path: addresses[1].network
# value: 192.168.1.0
# - path: addresses[1].prefix
# value: '24'
# - path: addresses[2].network
# value: 8.8.8.8
# - path: addresses[2].prefix
# value: '32'
- name: Make the updates
ansible.utils.update_fact:
updates: "{{ update_list }}"
register: updated
- debug:
var: updated
# TASK [debug] ***********************
# ok: [localhost] =>
# updated:
# addresses:
# - name: servers
# network: 10.1.1.0
# prefix: '24'
# raw: 10.1.1.0/255.255.255.0
# - name: printers
# network: 192.168.1.0
# prefix: '24'
# raw: 192.168.1.0/255.255.255.0
# - name: dns
# network: 8.8.8.8
# prefix: '32'
# raw: 8.8.8.8
# changed: true
# failed: false
#####################################################################
# Retrieve, update, and apply interface description change
# use index_of to locate Etherent1/1
#####################################################################
- name: Get the current interface config
cisco.nxos.nxos_interfaces:
state: gathered
register: interfaces
- name: Update the description of Ethernet1/1
ansible.utils.update_fact:
updates:
- path: "interfaces.gathered[{{ index }}].description"
value: "Configured by ansible"
vars:
index: "{{ interfaces.gathered|ansible.utils.index_of('eq', 'Ethernet1/1', 'name') }}"
register: updated
- name: Update the configuration
cisco.nxos.nxos_interfaces:
config: "{{ updated.interfaces.gathered }}"
state: overridden
register: result
- name: Show the commands issued
debug:
msg: "{{ result['commands'] }}"
# TASK [Show the commands issued] *************************************
# ok: [nxos101] => {
# "msg": [
# "interface Ethernet1/1",
# "description Configured by ansible"
# ]
# }
#####################################################################
# Retrieve, update, and apply an ipv4 ACL change
# finding the index of AFI ipv4 acls
# finding the index of the ACL named 'test1'
# finding the index of sequence 10
#####################################################################
- name: Retrieve the current acls
arista.eos.eos_acls:
state: gathered
register: current
- name: Update the source of sequenmce 10 in the IPv4 ACL named test1
ansible.utils.update_fact:
updates:
- path: current.gathered[{{ afi }}].acls[{{ acl }}].aces[{{ ace }}].source
value:
subnet_address: "192.168.2.0/24"
vars:
afi: "{{ current.gathered|ansible.utils.index_of('eq', 'ipv4', 'afi') }}"
acl: "{{ current.gathered[afi|int].acls|ansible.utils.index_of('eq', 'test1', 'name') }}"
ace: "{{ current.gathered[afi|int].acls[acl|int].aces|ansible.utils.index_of('eq', 10, 'sequence') }}"
register: updated
- name: Apply the changes
arista.eos.eos_acls:
config: "{{ updated.current.gathered }}"
state: overridden
register: changes
- name: Show the commands issued
debug:
msg: "{{ changes['commands'] }}"
# TASK [Show the commands issued] *************************************
# ok: [eos101] => {
# "msg": [
# "ip access-list test1",
# "no 10",
# "10 permit ip 192.168.2.0/24 host 10.1.1.2"
# ]
# }
#####################################################################
# Disable ip redirects on any layer3 interface
# find the layer 3 interfaces
# use each name to find their index in l3 interface
# build an 'update' list and apply the updates
#####################################################################
- name: Get the current interface and L3 interface configuration
cisco.nxos.nxos_facts:
gather_subset: min
gather_network_resources:
- interfaces
- l3_interfaces
- name: Build the list of updates to make
set_fact:
updates: "{{ updates + [entry] }}"
vars:
updates: []
entry:
path: "ansible_network_resources.l3_interfaces[{{ item }}].redirects"
value: False
w_mode: "{{ ansible_network_resources.interfaces|selectattr('mode', 'defined') }}"
m_l3: "{{ w_mode|selectattr('mode', 'eq', 'layer3') }}"
names: "{{ m_l3|map(attribute='name')|list }}"
l3_indicies: "{{ ansible_network_resources.l3_interfaces|ansible.utils.index_of('in', names, 'name', wantlist=True) }}"
loop: "{{ l3_indicies }}"
# TASK [Build the list of updates to make] ****************************
# ok: [nxos101] => (item=99) => changed=false
# ansible_facts:
# updates:
# - path: ansible_network_resources.l3_interfaces[99].redirects
# value: false
# ansible_loop_var: item
# item: 99
- name: Update the l3 interfaces
ansible.utils.update_fact:
updates: "{{ updates }}"
register: updated
# TASK [Update the l3 interfaces] *************************************
# changed: [nxos101] => changed=true
# ansible_network_resources:
# l3_interfaces:
# <...>
# - ipv4:
# - address: 10.1.1.1/24
# name: Ethernet1/100
# redirects: false
- name: Apply the configuration changes
cisco.nxos.l3_interfaces:
config: "{{ updated.ansible_network_resources.l3_interfaces }}"
state: overridden
register: changes
# TASK [Apply the configuration changes] ******************************
# changed: [nxos101] => changed=true
# commands:
# - interface Ethernet1/100
# - no ip redirects
"""

68
tests/integration/targets/update_fact/tasks/main.yaml Normal file
View File

@ -0,0 +1,68 @@
- name: Set a fact
set_fact:
a:
b:
c:
- 1
- 2
- name: Update the fact
ansible.utils.update_fact:
updates:
- path: a.b.c.0
value: 10
- path: "a['b']['c'][1]"
value: 20
register: updated
- assert:
that: "{{ updated.a == expected.a }}"
vars:
expected:
a:
b:
c:
- 10
- 20
- name: Update the fact
ansible.utils.update_fact:
updates:
- path: a
value:
x:
y:
z:
- 100
- True
register: updated
- assert:
that: "{{ updated.a == expected.a }}"
vars:
expected:
a:
x:
y:
z:
- 100
- True
- name: Update the fact
ansible.utils.update_fact:
updates:
- path: "a.b.c[{{ index }}]"
value: 20
vars:
index: "{{ a.b.c|ansible.utils.index_of('eq', 2) }}"
register: updated
- assert:
that: "{{ updated.a == expected.a }}"
vars:
expected:
a:
b:
c:
- 1
- 20

366
tests/unit/plugins/action/test_update_fact.py Normal file
View File

@ -0,0 +1,366 @@
# (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import copy
import unittest
from jinja2 import Template, TemplateSyntaxError
from mock import MagicMock
from ansible.playbook.task import Task
from ansible.template import Templar
from ansible_collections.ansible.utils.plugins.action.update_fact import (
ActionModule,
)
VALID_DATA = {
"a": {
"b": {"4.4": [{"1": {5: {"foo": 123}}}], 5.5: "float5.5"},
"127.0.0.1": "localhost",
}
}
VALID_TESTS = [
{
"path": 'a.b["4.4"][0]["1"].5[\'foo\']',
"split": ["a", "b", "4.4", 0, "1", 5, "foo"],
"template_result": "123",
},
{
"path": 'a.b["4.4"][0]["1"].5[\'foo\']',
"split": ["a", "b", "4.4", 0, "1", 5, "foo"],
"template_result": "123",
},
{
"path": "a.b[5.5]",
"split": ["a", "b", 5.5],
"template_result": "float5.5",
},
{
"path": "a['127.0.0.1']",
"split": ["a", "127.0.0.1"],
"template_result": "localhost",
},
{
"path": "a.b['4.4'].0['1'].5['foo']",
"split": ["a", "b", "4.4", 0, "1", 5, "foo"],
"template_result": "123",
},
]
INVALID_JINJA = [
{
"path": "a.'1'",
"note": "quoted values are required to be in brackets",
"error": "expected name or number",
},
{
"path": "a.[1]",
"note": "brackets can't follow dots",
"error": "expected name or number",
},
{
"path": 'a.b["4.4"][0]["1"]."5"[\'foo\']',
"note": "quoted values are required to be in brackets",
"error": "expected name or number",
},
]
class TestUpdate_Fact(unittest.TestCase):
def setUp(self):
task = MagicMock(Task)
play_context = MagicMock()
play_context.check_mode = False
connection = MagicMock()
fake_loader = {}
templar = Templar(loader=fake_loader)
self._plugin = ActionModule(
task=task,
connection=connection,
play_context=play_context,
loader=fake_loader,
templar=templar,
shared_loader_obj=None,
)
self._plugin._task.action = "update_fact"
def test_argspec_no_updates(self):
"""Check passing invalid argspec"""
self._plugin._task.args = {"a": 10}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=None)
self.assertIn(
"missing required arguments: updates",
str(error.exception),
)
def test_argspec_none(self):
"""Check passing a dict"""
self._plugin._task.args = {}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=None)
self.assertIn(
"missing required arguments: updates", str(error.exception)
)
def test_valid_jinja(self):
for test in VALID_TESTS:
tmplt = Template("{{" + test["path"] + "}}")
result = tmplt.render(VALID_DATA)
self.assertEqual(result, test["template_result"])
def test_invalid_jinja(self):
for test in INVALID_JINJA:
with self.assertRaises(TemplateSyntaxError) as error:
Template("{{" + test["path"] + "}}")
self.assertIn(test["error"], str(error.exception))
def test_fields(self):
"""Check the parsing of a path into it's parts"""
for stest in VALID_TESTS:
result = self._plugin._field_split(stest["path"])
self.assertEqual(result, stest["split"])
def test_missing_var(self):
"""Check for a missing fact"""
self._plugin._task.args = {"updates": [{"path": "a.b.c", "value": 5}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={"vars": {}})
self.assertIn(
"'a' was not found in the current facts.", str(error.exception)
)
def test_run_simple(self):
"""Confirm a valid argspec passes"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"] = 5
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b", "value": 5}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_multiple(self):
"""Confirm multiple paths passes"""
task_vars = {
"vars": {"a": {"b1": [1, 2, 3], "b2": {"c": "123", "d": False}}}
}
expected = {"a": {"b1": [1, 2, 3, 4], "b2": {"c": 456, "d": True}}}
expected.update({"changed": True})
self._plugin._task.args = {
"updates": [
{"path": "a.b1.3", "value": 4},
{"path": "a.b2.c", "value": 456},
{"path": "a.b2.d", "value": True},
]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_replace_in_list(self):
"""Replace in list"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"][1] = 5
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b.1", "value": 5}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_append_to_list(self):
"""Append to list"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4)
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b.3", "value": 4}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_bracket_single_quote(self):
"""Bracket notation sigle quote"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4)
expected.update({"changed": True})
self._plugin._task.args = {
"updates": [{"path": "a['b'][3]", "value": 4}]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_bracket_double_quote(self):
"""Bracket notation double quote"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4)
expected.update({"changed": True})
self._plugin._task.args = {
"updates": [{"path": 'a["b"][3]', "value": 4}]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_int_dict_keys(self):
"""Integer dict keys"""
task_vars = {"vars": {"a": {0: [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"][0][0] = 0
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.0.0", "value": 0}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_int_as_string(self):
"""Integer dict keys as string"""
task_vars = {"vars": {"a": {"0": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["0"][0] = 0
expected.update({"changed": True})
self._plugin._task.args = {
"updates": [{"path": 'a["0"].0', "value": 0}]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_invalid_path_quote_after_dot(self):
"""Invalid path format"""
self._plugin._task.args = {"updates": [{"path": "a.'b'", "value": 0}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception))
def test_run_invalid_path_bracket_after_dot(self):
"""Invalid path format"""
self._plugin._task.args = {
"updates": [{"path": "a.['b']", "value": 0}]
}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception))
def test_run_invalid_key_start_with_dot(self):
"""Invalid key format"""
self._plugin._task.args = {"updates": [{"path": ".abc", "value": 0}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception))
def test_run_no_update_list(self):
"""Confirm no change when same"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"] = [1, 2, 3]
expected.update({"changed": False})
self._plugin._task.args = {"updates": [{"path": "a.b.0", "value": 1}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_no_update_dict(self):
"""Confirm no change when same"""
task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"] = [1, 2, 3]
expected.update({"changed": False})
self._plugin._task.args = {
"updates": [{"path": "a.b", "value": [1, 2, 3]}]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_missing_key(self):
"""Confirm error when key not found"""
task_vars = {"vars": {"a": {"b": 1}}}
self._plugin._task.args = {"updates": [{"path": "a.c.d", "value": 1}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars)
self.assertIn("the key 'c' was not found", str(error.exception))
def test_run_list_not_int(self):
"""Confirm error when key not found"""
task_vars = {"vars": {"a": {"b": [1]}}}
self._plugin._task.args = {
"updates": [{"path": "a.b['0']", "value": 2}]
}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars)
self.assertIn(
"index provided was not an integer", str(error.exception)
)
def test_run_list_not_long(self):
"""List not long enough"""
task_vars = {"vars": {"a": {"b": [0]}}}
self._plugin._task.args = {"updates": [{"path": "a.b.2", "value": 2}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars)
self.assertIn(
"not long enough for item #2 to be set", str(error.exception)
)
def test_not_mutable_sequence_or_mapping(self):
"""Confirm graceful fail when immutable object
This should never happen in the real world
"""
obj = {"a": frozenset([1, 2, 3])}
path = ["a", 0]
val = 9
with self.assertRaises(Exception) as error:
self._plugin.set_value(obj, path, val)
self.assertIn("can only modify mutable objects", str(error.exception))
def test_run_not_dotted_success_one(self):
"""Test with a not dotted key"""
task_vars = {"vars": {"a": 0}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"] = 1
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a", "value": 1}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_not_dotted_success_three(self):
"""Test with a not dotted key longer"""
task_vars = {"vars": {"abc": 0}}
expected = copy.deepcopy(task_vars["vars"])
expected["abc"] = 1
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "abc", "value": 1}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_not_dotted_fail_missing(self):
"""Test with a not dotted key, missing"""
task_vars = {"vars": {"abc": 0}}
self._plugin._task.args = {"updates": [{"path": "123", "value": 1}]}
with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars)
self.assertIn(
"'123' was not found in the current facts", str(error.exception)
)
def test_run_not_dotted_success_same(self):
"""Test with a not dotted key, no change"""
task_vars = {"vars": {"a": 0}}
expected = copy.deepcopy(task_vars["vars"])
expected.update({"changed": False})
self._plugin._task.args = {"updates": [{"path": "a", "value": 0}]}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)
def test_run_looks_like_a_bool(self):
"""Test with a key that looks like a bool"""
task_vars = {"vars": {"a": {"True": 0}}}
expected = copy.deepcopy(task_vars["vars"])
expected["a"]["True"] = 1
expected.update({"changed": True})
self._plugin._task.args = {
"updates": [{"path": "a['True']", "value": 1}]
}
result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected)