Utils fix issues with cli_parse and facts (#428)

* fix utils

* Update fix for tpp

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update changes for tests

* fix uts again

* sanittyyy

* test for utils ipaddress_utils

* Check

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: KB-perByte

changelogs/fragments/402-flat_list.yml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - cli_parse - Honor ttp_results.results flat_list in TTP parser so output is a single-level list instead of double-wrapped (https://github.com/ansible-collections/ansible.utils/issues/402).

changelogs/fragments/426-update_fact_task_vars.yml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - update_fact - Use task_vars at top-level instead of the deprecated ``vars`` key for compatibility with ansible-core 2.24 (ansible/ansible issue #426).

changelogs/fragments/cidr_merge-macro_context.yml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - cidr_merge - Fix filter failing when used inside a Jinja2 macro called with ``with context`` by unwrapping Ansible lazy template lists before validation.

changelogs/fragments/ipaddress_utils-python314.yml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - ipaddress_utils - Support Python 3.14+ by using the public ``version`` attribute instead of the removed private ``_version`` on ``ipaddress`` network objects (bpo-118710).

plugins/action/update_fact.py

@@ -163,23 +163,30 @@ def run(self, tmp=None, task_vars=None):
         self._result["changed"] = False
         self._check_argspec()
         results = set()
+        full_replaces = set()  # keys that were fully replaced (no path)
         self._ensure_valid_jinja()
+        # Use task_vars (top-level) instead of task_vars["vars"] to avoid the
+        # deprecated internal "vars" dictionary (ansible-core 2.24, issue #426).
         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"]:
+            if obj not in task_vars:
                 msg = "'{obj}' was not found in the current facts.".format(obj=obj)
                 raise AnsibleActionFail(msg)
-            retrieved = task_vars["vars"].get(obj)
+            retrieved = task_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"]
+                if retrieved != entry["value"]:
+                    self._result.setdefault("ansible_facts", {})[obj] = entry["value"]
+                    full_replaces.add(obj)
                     self._result["changed"] = True
 
         for key in results:
-            value = task_vars["vars"].get(key)
+            if key in full_replaces:
+                value = self._result.get("ansible_facts", {}).get(key)
+            else:
+                value = task_vars.get(key)
             self._result[key] = value
         return self._result

plugins/filter/cidr_merge.py

@@ -129,12 +129,26 @@ class mac_linux(netaddr.mac_unix):
 """
 
 
+def _unwrap_lazy_list(value):
+    """
+    If value is an Ansible lazy template list, return a list of raw elements without
+    triggering templar.template() resolution. This avoids Template.__new__() errors
+    when the filter is used inside a Jinja2 macro called with ``with context``.
+    """
+    if value is not None and hasattr(value, "_yield_non_lazy_list_items"):
+        return list(value._yield_non_lazy_list_items())
+    return value
+
+
 @pass_environment
 def _cidr_merge(*args, **kwargs):
-    """Convert the given data from json to xml."""
+    """Merge CIDRs; compatible with pipe syntax (value | cidr_merge)."""
     keys = ["value", "action"]
     data = dict(zip(keys, args[1:]))
     data.update(kwargs)
+    # Unwrap lazy list when present to avoid resolution bug in macro context (no template.j2 change).
+    if "value" in data:
+        data["value"] = _unwrap_lazy_list(data["value"])
     aav = AnsibleArgSpecValidator(data=data, schema=DOCUMENTATION, name="cidr_merge")
     valid, errors, updated_data = aav.validate()
     if not valid:

plugins/plugin_utils/base/ipaddress_utils.py

@@ -70,14 +70,19 @@ def _get_network_version(network):
 
 
 def _is_subnet_of(network_a, network_b):
+    """
+    Return True if network_a is a subnet of network_b (same logic as ipaddress).
+    Uses the public .version attribute for compatibility with Python 3.14+ where
+    the private _version was removed (see bpo-118710 / cpython@c530ce1).
+    """
     try:
         if _get_network_version(network_a) != _get_network_version(network_b):
             return False
         return (
             network_b.network_address <= network_a.network_address
             and network_b.broadcast_address >= network_a.broadcast_address
         )
-    except Exception:
+    except (AttributeError, TypeError):
         return False
 
 

plugins/sub_plugins/cli_parser/ttp_parser.py

@@ -119,7 +119,23 @@ def parse(self, *_args, **_kwargs):
                 if parser_param.get("vars")
                 else {}
             )
-            results = parser.result(**ttp_results)
+            # TTP's result() uses keyword "structure", but we accept "results" from users
+            result_kwargs = dict(ttp_results)
+            if "results" in result_kwargs:
+                result_kwargs["structure"] = result_kwargs.pop("results")
+            results = parser.result(**result_kwargs)
+            # When flat_list is requested, flatten one level if we got [[...]]
+            requested_flat = (
+                ttp_results.get("results") == "flat_list"
+                or ttp_results.get("structure") == "flat_list"
+            )
+            if (
+                requested_flat
+                and isinstance(results, list)
+                and len(results) == 1
+                and isinstance(results[0], list)
+            ):
+                results = results[0]
         except Exception as exc:
             msg = "Template Text Parser returned an error while parsing. Error: {err}"
             return {"errors": [msg.format(err=to_native(exc))]}

tests/unit/plugins/action/test_update_fact.py

@@ -137,24 +137,23 @@ 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._plugin.run(task_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"])
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["b"] = 5
-        expected.update({"changed": True})
+        expected["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})
+        task_vars = {"a": {"b1": [1, 2, 3], "b2": {"c": "123", "d": False}}}
+        expected = {"a": {"b1": [1, 2, 3, 4], "b2": {"c": 456, "d": True}}, "changed": True}
         self._plugin._task.args = {
             "updates": [
                 {"path": "a.b1.3", "value": 4},
@@ -167,60 +166,60 @@ def test_run_multiple(self):
 
     def test_run_replace_in_list(self):
         """Replace in list"""
-        task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
-        expected = copy.deepcopy(task_vars["vars"])
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["b"][1] = 5
-        expected.update({"changed": True})
+        expected["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"])
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["b"].append(4)
-        expected.update({"changed": True})
+        expected["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"])
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["b"].append(4)
-        expected.update({"changed": True})
+        expected["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"])
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["b"].append(4)
-        expected.update({"changed": True})
+        expected["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"])
+        task_vars = {"a": {0: [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"][0][0] = 0
-        expected.update({"changed": True})
+        expected["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"])
+        task_vars = {"a": {"0": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["0"][0] = 0
-        expected.update({"changed": True})
+        expected["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)
@@ -229,62 +228,60 @@ 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._plugin.run(task_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._plugin.run(task_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._plugin.run(task_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})
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
+        expected["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})
+        task_vars = {"a": {"b": [1, 2, 3]}}
+        expected = copy.deepcopy(task_vars)
+        expected["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}}}
+        task_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]}}}
+        task_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]}}}
+        task_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)
@@ -303,47 +300,43 @@ def test_not_mutable_sequence_or_mapping(self):
 
     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})
+        task_vars = {"a": 0}
+        expected = {"a": 1, "ansible_facts": {"a": 1}, "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})
+        task_vars = {"abc": 0}
+        expected = {"abc": 1, "ansible_facts": {"abc": 1}, "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}}
+        task_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})
+        task_vars = {"a": 0}
+        expected = copy.deepcopy(task_vars)
+        expected["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"])
+        task_vars = {"a": {"True": 0}}
+        expected = copy.deepcopy(task_vars)
         expected["a"]["True"] = 1
-        expected.update({"changed": True})
+        expected["changed"] = True
         self._plugin._task.args = {"updates": [{"path": "a['True']", "value": 1}]}
         result = self._plugin.run(task_vars=task_vars)
         self.assertEqual(result, expected)