aristanetworks · ClausHolbechArista · Oct 12, 2023 · Oct 5, 2023 · Oct 5, 2023 · Oct 6, 2023
@@ -73,14 +73,6 @@ anta.tests.interfaces:
       - Interface State
       custom_field: Port-Channel1 - DC1_L3_LEAF1_Po8
       description: Port-Channel Interface & Line Protocol == \"up\"
-anta.tests.routing.generic:
-- VerifyRoutingProtocolModel:
-    model: multi-agent
-    result_overwrite:
-      categories:
-      - BGP
-      custom_field: ArBGP
-      description: ArBGP is configured and operating
 anta.tests.system:
 - VerifyNTP:
     result_overwrite:

@@ -73,14 +73,6 @@ anta.tests.interfaces:
       - Interface State
       custom_field: Port-Channel1 - DC1_L3_LEAF2_Po8
       description: Port-Channel Interface & Line Protocol == \"up\"
-anta.tests.routing.generic:
-- VerifyRoutingProtocolModel:
-    model: multi-agent
-    result_overwrite:
-      categories:
-      - BGP
-      custom_field: ArBGP
-      description: ArBGP is configured and operating
 anta.tests.system:
 - VerifyNTP:
     result_overwrite:

@@ -73,14 +73,6 @@ anta.tests.interfaces:
       - Interface State
       custom_field: Port-Channel1 - DC2_L3_LEAF1_Po8
       description: Port-Channel Interface & Line Protocol == \"up\"
-anta.tests.routing.generic:
-- VerifyRoutingProtocolModel:
-    model: multi-agent
-    result_overwrite:
-      categories:
-      - BGP
-      custom_field: ArBGP
-      description: ArBGP is configured and operating
 anta.tests.system:
 - VerifyNTP:
     result_overwrite:

@@ -73,14 +73,6 @@ anta.tests.interfaces:
       - Interface State
       custom_field: Port-Channel1 - DC2_L3_LEAF2_Po8
       description: Port-Channel Interface & Line Protocol == \"up\"
-anta.tests.routing.generic:
-- VerifyRoutingProtocolModel:
-    model: multi-agent
-    result_overwrite:
-      categories:
-      - BGP
-      custom_field: ArBGP
-      description: ArBGP is configured and operating
 anta.tests.system:
 - VerifyNTP:
     result_overwrite:

@@ -2,6 +2,7 @@
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
 from .ansible_eos_device import AnsibleEOSDevice
+from .avdtestbase import AvdTestBase
 from .get_anta_results import get_anta_results
 
-__all__ = ["AnsibleEOSDevice", "get_anta_results"]
+__all__ = ["AnsibleEOSDevice", "get_anta_results", "AvdTestBase"]
@@ -24,13 +24,18 @@
 ## Known limitations
 
 - Loose mode to ignore playbook errors is no longer supported in ANTA mode.
+- ANTA mode exclusively supports the newer "list-of-dicts" data models in the structured configuration file input. For further details, consult the AVD 4.x.x [porting guides](https://avd.sh/en/stable/docs/porting-guides/4.x.x.html#data-model-changes-from-dict-of-dicts-to-list-of-dicts).
 
 ## Expected changes
 
 - You should expect faster execution, and if not please report on the GitHub [discussions board](https://github.com/aristanetworks/ansible-avd/discussions)
 - Hardware tests are now collapsed.
 - Some description of tests have been updated to be more precise.
 - Sorting of the test results is now done per device as opposed to per category.
+- BGP tests will only run if `service_routing_protocols_model` is set to `multi-agent` in the structured configuration file.
+
+!!! note
+    Starting from version 4.30.1F, `service_routing_protocols_model` is preset to `multi-agent` by default on EOS devices.
 
 ## How to run eos_validate_state in ANTA mode
 
@@ -103,7 +108,6 @@
 - AvdTestBGP (Ansible tags: `bgp_check`)
   - VerifyBGPSpecificPeers: Validate IP BGP and BGP EVPN sessions state.
   - VerifyRoutingProtocolModel: Validate ArBGP is configured and operating.
-    - *This test is currently being run on devices that do not have a BGP configuration, which is not the actual behavior of `eos_validate_state`.*
 
 - AvdTestReloadCause (Ansible tags: `reload_cause`, `optional`, `never`)
   - VerifyReloadCause: Validate last reload cause. (Optional)

@@ -106,41 +106,62 @@ def add_verify_peers_test(description: str, afi: str, bgp_neighbor_ip: str, safi
                 }
             )
 
-        # Add test to check service_routing_protocol_model
-        if (
-            service_routing_protocols_model := get(self.hostvars[self.device_name], "service_routing_protocols_model")
-        ) is None or service_routing_protocols_model != "multi-agent":
-            LOGGER.warning(
-                "Variable 'service_routing_protocols_model' is missing from structured_config or is NOT set to 'multi-agent'. %s is skipped.",
-                self.__class__.__name__,
-            )
+        try:
+            get(self.hostvars[self.device_name], "router_bgp", required=True)
+        except AristaAvdMissingVariableError as e:
+            LOGGER.info("Variable '%s' is missing from the structured_config. %s is skipped.", str(e), self.__class__.__name__)
             return None
 
-        anta_tests.setdefault("generic", []).append(
-            {
-                "VerifyRoutingProtocolModel": {
-                    "model": "multi-agent",
-                    "result_overwrite": {"categories": self.categories, "description": "ArBGP is configured and operating", "custom_field": "ArBGP"},
-                }
-            }
-        )
+        try:
+            service_routing_protocols_model = get(self.hostvars[self.device_name], "service_routing_protocols_model", required=True)
+        except AristaAvdMissingVariableError as e:
+            LOGGER.warning("Variable '%s' is missing from the structured_config. %s is skipped.", str(e), self.__class__.__name__)
+            return None
 
-        bgp_peer_groups = get(self.hostvars[self.device_name], "router_bgp.peer_groups", [])
+        if service_routing_protocols_model == "multi-agent":
+            anta_tests.setdefault("generic", []).append(
+                {
+                    "VerifyRoutingProtocolModel": {
+                        "model": "multi-agent",
+                        "result_overwrite": {"categories": self.categories, "description": "ArBGP is configured and operating", "custom_field": "ArBGP"},
+                    }
+                }
+            )
 
-        for bgp_neighbor in get(self.hostvars[self.device_name], "router_bgp.neighbors", []):
-            # TODO - this matches legacy eos_validate_state BUT works only for neighbors in peer groups...
-            try:
-                neighbor_peer_group = get_item(
-                    bgp_peer_groups, "name", bgp_neighbor["peer_group"], required=True, var_name=f"name: {bgp_neighbor['peer_group']}"
-                )
-                bgp_neighbor_ip = str(get(bgp_neighbor, "ip_address", required=True))
+            bgp_peer_groups = get(self.hostvars[self.device_name], "router_bgp.peer_groups", [])
+            bgp_neighbors = get(self.hostvars[self.device_name], "router_bgp.neighbors", [])
+
+            for idx, bgp_neighbor in enumerate(bgp_neighbors, start=1):
+                # TODO - this matches legacy eos_validate_state BUT works only for neighbors in peer groups...
+                try:
+                    neighbor_peer_group = get_item(bgp_peer_groups, "name", bgp_neighbor["peer_group"], required=True, var_name=bgp_neighbor["peer_group"])
+                except AristaAvdMissingVariableError as e:
+                    LOGGER.warning("Peer group '%s' dictionary is missing from the 'peer_groups' list of the 'router_bgp' data model.", str(e))
+                    continue
+
+                try:
+                    bgp_neighbor_ip = str(get(bgp_neighbor, "ip_address", required=True))
+                except AristaAvdMissingVariableError as e:
+                    LOGGER.warning("Neighbor entry #%d from the 'neighbors' list of the 'router_bgp' data model is missing the variable '%s'.", idx, str(e))
+                    continue
+
+                try:
+                    neighbor_peer_group_type = get(neighbor_peer_group, "type", required=True)
+                except AristaAvdMissingVariableError as e:
+                    LOGGER.warning(
+                        "Peer group '%s' from the 'peer_groups' list of the 'router_bgp' data model is missing the variable '%s'.",
+                        bgp_neighbor["peer_group"],
+                        str(e),
+                    )
+                    continue
 
-                if get(neighbor_peer_group, "type", required=True) == "ipv4":
+                if neighbor_peer_group_type == "ipv4":
                     add_verify_peers_test(description="ip bgp peer state established (ipv4)", afi="ipv4", safi="unicast", bgp_neighbor_ip=bgp_neighbor_ip)
-                elif get(neighbor_peer_group, "type", required=True) == "evpn":
+                elif neighbor_peer_group_type == "evpn":
                     add_verify_peers_test(description="bgp evpn peer state established (evpn)", afi="evpn", bgp_neighbor_ip=bgp_neighbor_ip)
 
-            except AristaAvdMissingVariableError as e:
-                LOGGER.warning("Variable '%s' is missing. Please validate the Router BGP data model of this host.", str(e))
+        else:
+            LOGGER.warning("Variable 'service_routing_protocols_model' is NOT set to 'multi-agent'. %s is skipped.", self.__class__.__name__)
+            return None
 
         return {self.anta_module: anta_tests}
@@ -0,0 +1,29 @@
+# Copyright (c) 2023 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 annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from ansible_collections.arista.avd.plugins.plugin_utils.eos_validate_state_utils import AvdTestBase
+from ansible_collections.arista.avd.plugins.plugin_utils.utils import load_python_class
+
+if TYPE_CHECKING:
+    from pytest import LogCaptureFixture
+
+
+def test_avd_tests(caplog: LogCaptureFixture, data: dict) -> None:
+    """
+    Generic function for all AvdTestBase subclasses unit tests.
+    """
+    caplog.set_level(logging.INFO)
+    avd_test_class = load_python_class("ansible_collections.arista.avd.roles.eos_validate_state.python_modules.tests", data["test_module"], AvdTestBase)
+    avd_test_bgp = avd_test_class(device_name="DC1-SPINE1", hostvars=data["hostvars"])
+    result = avd_test_bgp.render()
+    assert result == data["expected_result"]
+
+    if data["expected_log"] and data["expected_log_level"]:
+        found_expected_log = any(record.message == data["expected_log"] and record.levelname == data["expected_log_level"] for record in caplog.records)
+        assert found_expected_log, f"Expected log message not found: {data['expected_log']}"
@@ -0,0 +1,35 @@
+# Copyright (c) 2023 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 annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pytest import Metafunc
+
+
+def build_test_id(val: dict) -> str:
+    """
+    Build a proper test ID for AvdTestBase subclasses unit tests: `<test_module>-<test_name>`
+
+    Examples output:
+
+    `test_avd_tests[AvdTestBGP-missing-router-bgp]`
+
+    `test_avd_tests[AvdTestBGP-missing-service-routing-protocols-model]`
+    """
+    return f"{val['test_module']}-{val['test_name']}"
+
+
+def pytest_generate_tests(metafunc: Metafunc) -> None:
+    """
+    This function is called during test collection.
+
+    It will parametrize test cases based on the `DATA` data structure defined in `tests.unit.roles.eos_validate_state.tests` modules.
+
+    Test IDs are generated using the `build_test_id` function above.
+    """
+    if "tests.unit.roles.eos_validate_state.tests" in metafunc.module.__package__:
+        metafunc.parametrize("data", metafunc.module.DATA, ids=build_test_id)