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
ansible.utils/tests/unit/plugins/action/test_update_fact.py

350 lines
13 KiB
Python
Raw Blame History

# -*- 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 copy
from unittest import TestCase
from ansible.playbook.task import Task
from ansible.template import Templar
from jinja2 import Template, TemplateSyntaxError
from ansible_collections.ansible.utils.plugins.action.update_fact import ActionModule
try:
from unittest.mock import MagicMock # pylint:disable=syntax-error
except ImportError:
from mock import MagicMock # pyright: ignore[reportMissingModuleSource]
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(TestCase):
def setUp(self):
task = MagicMock(Task)
# Ansible > 2.13 looks for check_mode in task
task.check_mode = False
play_context = MagicMock()
# Ansible <= 2.13 looks for check_mode in play_context
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)
Reference in New Issue View Git Blame Copy Permalink