Refactor(plugins): Move jinja test code for arista.avd.defined to PyAVD

ansible_collections/arista/avd/plugins/plugin_utils/pyavd_wrappers.py

@@ -5,13 +5,17 @@
 
 __metaclass__ = type
 
+import warnings
 from functools import partial, wraps
 from typing import Callable, Literal
 
 from ansible.errors import AnsibleFilterError, AnsibleInternalError, AnsibleTemplateError, AnsibleUndefinedVariable
 from ansible.module_utils.basic import to_native
+from ansible.utils.display import Display
 from jinja2.exceptions import UndefinedError
 
+display = Display()
+
 
 class RaiseOnUse:
     """
@@ -39,8 +43,17 @@ def wrap_plugin(plugin_type: Literal["filter", "test"], name: str) -> Callable:
     def wrap_plugin_decorator(func: Callable) -> Callable:
         @wraps(func)
         def plugin_wrapper(*args, **kwargs):
+            """Wrapper function for plugins.
+
+            NOTE: if the same warning is raised multiple times, Ansible Display() will print only one
+            """
             try:
-                return func(*args, **kwargs)
+                with warnings.catch_warnings(record=True) as w:
+                    result = func(*args, **kwargs)
+                    if w:
+                        for warning in w:
+                            display.warning(str(warning.message))
+                return result
             except UndefinedError as e:
                 raise AnsibleUndefinedVariable(f"{plugin_type.capitalize()} '{name}' failed: {to_native(e)}", orig_exc=e) from e
             except Exception as e:

ansible_collections/arista/avd/plugins/test/defined.py

@@ -25,9 +25,22 @@
 
 __metaclass__ = type
 
-from ansible.errors import AnsibleError
-from ansible.utils.display import Display
-from jinja2.runtime import Undefined
+from ansible.errors import AnsibleTemplateError
+
+from ansible_collections.arista.avd.plugins.plugin_utils.pyavd_wrappers import RaiseOnUse, wrap_test
+
+PLUGIN_NAME = "arista.avd.defined"
+
+try:
+    from pyavd.j2tests.defined import defined
+except ImportError as e:
+    defined = RaiseOnUse(
+        AnsibleTemplateError(
+            f"The '{PLUGIN_NAME}' plugin requires the 'pyavd' Python library. Got import error",
+            orig_exc=e,
+        )
+    )
+
 
 DOCUMENTATION = r"""
 ---
@@ -84,106 +97,6 @@
 """
 
 
-def defined(value, test_value=None, var_type=None, fail_action=None, var_name=None, run_tests=False):
-    """
-    defined - Ansible test plugin to test if a variable is defined and not none
-
-    Arista.avd.defined will test value if defined and is not none and return true or false.
-    If test_value is supplied, the value must also pass == test_value to return true.
-    If var_type is supplied, the value must also be of the specified class/type
-    If fail_action is 'warning' a warning will be emitted on failure.
-    If fail_action is 'error' an error will be emitted on failure and the task will fail.
-    If var_name is supplied it will be used in the warning and error messages to ease troubleshooting.
-
-    Examples:
-    1. Test if var is defined and not none:
-    {% if spanning_tree is arista.avd.defined %}
-    ...
-    {% endif %}
-
-    2. Test if variable is defined, not none and has value "something"
-    {% if extremely_long_variable_name is arista.avd.defined("something") %}
-    ...
-    {% endif %}
-
-    3. Test if variable is defined and of not print a warning message with the variable name
-    {% if my_dict.my_list[12].my_var is arista.avd.defined(fail_action='warning', var_name='my_dict.my_list[12].my_var' %}
-
-    Parameters
-    ----------
-    value : any
-        Value to test from ansible
-    test_value : any, optional
-        Value to test in addition of defined and not none, by default None
-    var_type : ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'], optional
-        Type or Class to test for
-    fail_action : ['warning', 'error'], optional
-        Optional action if test fails to emit a Warning or Error
-    var_name : <string>, optional
-        Optional string to use as variable name in warning or error messages
-
-    Returns
-    -------
-    boolean
-        True if variable matches criteria, False in other cases.
-    """
-    display = Display()
-    if isinstance(value, Undefined) or value is None:
-        # Invalid value - return false
-        if str(fail_action).lower() == "warning":
-            display._warns = {}
-            if var_name is not None:
-                display.warning(f"{var_name} was expected but not set. Output may be incorrect or incomplete!")
-            else:
-                display.warning("A variable was expected but not set. Output may be incorrect or incomplete!")
-        elif str(fail_action).lower() == "error":
-            if var_name is not None:
-                raise AnsibleError(f"{var_name} was expected but not set!")
-            else:
-                raise AnsibleError("A variable was expected but not set!")
-        if run_tests:
-            return False, display._warns
-        return False
-
-    elif test_value is not None and value != test_value:
-        # Valid value but not matching the optional argument
-        if str(fail_action).lower() == "warning":
-            display._warns = {}
-            if var_name is not None:
-                display.warning(f"{var_name} was set to {value} but we expected {test_value}. Output may be incorrect or incomplete!")
-            else:
-                display.warning(f"A variable was set to {value} but we expected {test_value}. Output may be incorrect or incomplete!")
-        elif str(fail_action).lower() == "error":
-            if var_name is not None:
-                raise AnsibleError(f"{var_name} was set to {value} but we expected {test_value}!")
-            else:
-                raise AnsibleError(f"A variable was set to {value} but we expected {test_value}!")
-        if run_tests:
-            return False, display._warns
-        return False
-    elif str(var_type).lower() in ["float", "int", "str", "list", "dict", "tuple", "bool"] and str(var_type).lower() != type(value).__name__:
-        # Invalid class - return false
-        if str(fail_action).lower() == "warning":
-            display._warns = {}
-            if var_name is not None:
-                display.warning(f"{var_name} was a {type(value).__name__} but we expected a {str(var_type).lower()}. Output may be incorrect or incomplete!")
-            else:
-                display.warning(f"A variable was a {type(value).__name__} but we expected a {str(var_type).lower()}. Output may be incorrect or incomplete!")
-        elif str(fail_action).lower() == "error":
-            if var_name is not None:
-                raise AnsibleError(f"{var_name} was a {type(value).__name__} but we expected a {str(var_type).lower()}!")
-            else:
-                raise AnsibleError(f"A variable was a {type(value).__name__} but we expected a {str(var_type).lower()}!")
-        if run_tests:
-            return False, display._warns
-        return False
-    else:
-        # Valid value and is matching optional argument if provided - return true
-        return True
-
-
 class TestModule(object):
     def tests(self):
-        return {
-            "defined": defined,
-        }
+        return {"defined": wrap_test(PLUGIN_NAME)(defined)}

python-avd/pyavd/j2tests/defined.py

@@ -0,0 +1,118 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""AVD Jinja2 test defined.
+
+The test checks if a passed variable is defined, and if specified, its value and its type.
+"""
+
+from __future__ import annotations
+
+import warnings
+
+from jinja2.runtime import Undefined
+
+
+def defined(value, test_value=None, var_type=None, fail_action=None, var_name=None, run_tests=False):
+    """
+    defined - Ansible test plugin to test if a variable is defined and not none
+
+    Arista.avd.defined will test value if defined and is not none and return true or false.
+    If test_value is supplied, the value must also pass == test_value to return true.
+    If var_type is supplied, the value must also be of the specified class/type
+    If fail_action is 'warning' a warning will be emitted on failure.
+    If fail_action is 'error' an error will be emitted on failure and the task will fail.
+    If var_name is supplied it will be used in the warning and error messages to ease troubleshooting.
+
+    Examples:
+    1. Test if var is defined and not none:
+    {% if spanning_tree is arista.avd.defined %}
+    ...
+    {% endif %}
+
+    2. Test if variable is defined, not none and has value "something"
+    {% if extremely_long_variable_name is arista.avd.defined("something") %}
+    ...
+    {% endif %}
+
+    3. Test if variable is defined and of not print a warning message with the variable name
+    {% if my_dict.my_list[12].my_var is arista.avd.defined(fail_action='warning', var_name='my_dict.my_list[12].my_var' %}
+
+    Parameters
+    ----------
+    value : any
+        Value to test from ansible
+    test_value : any, optional
+        Value to test in addition of defined and not none, by default None
+    var_type : ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'], optional
+        Type or Class to test for
+    fail_action : ['warning', 'error'], optional
+        Optional action if test fails to emit a Warning or Error
+    var_name : <string>, optional
+        Optional string to use as variable name in warning or error messages
+
+    Returns
+    -------
+    boolean
+        True if variable matches criteria, False in other cases.
+    """
+    if isinstance(value, Undefined) or value is None:
+        # Invalid value - return false
+        if str(fail_action).lower() == "warning":
+            warnings_count = {}
+            if var_name is not None:
+                warning_msg = f"{var_name} was expected but not set. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+            else:
+                warning_msg = "A variable was expected but not set. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+        elif str(fail_action).lower() == "error":
+            if var_name is not None:
+                raise ValueError(f"{var_name} was expected but not set!")
+            raise ValueError("A variable was expected but not set!")
+        if run_tests:
+            return False, warnings_count
+        return False
+
+    if test_value is not None and value != test_value:
+        # Valid value but not matching the optional argument
+        if str(fail_action).lower() == "warning":
+            warnings_count = {}
+            if var_name is not None:
+                warning_msg = f"{var_name} was set to {value} but we expected {test_value}. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+            else:
+                warning_msg = f"A variable was set to {value} but we expected {test_value}. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+        elif str(fail_action).lower() == "error":
+            if var_name is not None:
+                raise ValueError(f"{var_name} was set to {value} but we expected {test_value}!")
+            raise ValueError(f"A variable was set to {value} but we expected {test_value}!")
+        if run_tests:
+            return False, warnings_count
+        return False
+    if str(var_type).lower() in ["float", "int", "str", "list", "dict", "tuple", "bool"] and str(var_type).lower() != type(value).__name__:
+        # Invalid class - return false
+        if str(fail_action).lower() == "warning":
+            warnings_count = {}
+            if var_name is not None:
+                warning_msg = f"{var_name} was a {type(value).__name__} but we expected a {str(var_type).lower()}. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+            else:
+                warning_msg = f"A variable was a {type(value).__name__} but we expected a {str(var_type).lower()}. Output may be incorrect or incomplete!"
+                warnings.warn(warning_msg)
+                warnings_count["[WARNING]: " + warning_msg] = warnings_count.get("[WARNING]: " + warning_msg, 0) + 1
+        elif str(fail_action).lower() == "error":
+            if var_name is not None:
+                raise ValueError(f"{var_name} was a {type(value).__name__} but we expected a {str(var_type).lower()}!")
+            raise ValueError(f"A variable was a {type(value).__name__} but we expected a {str(var_type).lower()}!")
+        if run_tests:
+            return False, warnings_count
+        return False
+    # Valid value and is matching optional argument if provided - return true
+    return True

python-avd/pyavd/templater.py

@@ -82,7 +82,7 @@ def import_filters_and_tests(self) -> None:
         from .j2filters.snmp_hash import snmp_hash
         from .j2filters.status_render import status_render
         from .j2tests.contains import contains
-        from .vendor.j2.test.defined import defined
+        from .j2tests.defined import defined
 
         # pylint: enable=import-outside-toplevel
 

ansible_collections/arista/avd/tests/unit/test/test_defined_plugin.py → python-avd/tests/pyavd/j2tests/test_defined_plugin.py

@@ -1,15 +1,13 @@
 # Copyright (c) 2023-2024 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
-from __future__ import absolute_import, division, print_function
+from __future__ import annotations
 
-__metaclass__ = type
+import warnings
 
 import pytest
-from ansible.errors import AnsibleError
 from jinja2.runtime import Undefined
-
-from ansible_collections.arista.avd.plugins.test.defined import TestModule, defined
+from pyavd.j2tests.defined import defined
 
 VALUE_LIST = ["ab", None, 1, True, {"key": "value"}]
 TEST_VALUE_LIST = [None, "ab", True, 1, True]
@@ -18,20 +16,22 @@
 VAR_TYPE_LIST = ["int", "str", "integer", "aaa", None, dict]
 INVALID_FAIL_ACTION_LIST = [None, "aaaa"]
 
-f = TestModule()
-
 
 class TestDefinedPlugin:
     def defined_function(self, value, test_value=None, var_type=None, fail_action=None, var_name=None, err_msg=None, warn_msg=None):
         if str(fail_action).lower() == "warning":
-            resp, warning = defined(value, test_value=test_value, var_type=var_type, fail_action=fail_action, var_name=var_name, run_tests=True)
+            with warnings.catch_warnings(record=True) as w:
+                resp, warning = defined(value, test_value=test_value, var_type=var_type, fail_action=fail_action, var_name=var_name, run_tests=True)
+            assert len(w) == 1
+            assert isinstance(w[0].message, UserWarning)
             if warn_msg:
                 assert warning is not None
                 warn = str(list(warning.keys())[0]).replace("[WARNING]: ", "").strip().replace("\n", " ")
                 assert warn == warn_msg
-                assert resp is False
+                assert str(w[0].message) == warn_msg
+            assert resp is False
         elif str(fail_action).lower() == "error":
-            with pytest.raises(AnsibleError) as e:
+            with pytest.raises(ValueError) as e:
                 resp, warning = defined(value, test_value=test_value, var_type=var_type, fail_action=fail_action, var_name=var_name, run_tests=True)
             assert str(e.value) == err_msg