ANTA workflows

pyproject.toml

@@ -82,6 +82,10 @@ convention = "google"
   "INP001",  # implicit namespace package. Add an `__init__.py` - Tests are not in packages
 ]
 
+"python-avd/pyavd/_anta/utils/index.py" = [
+  "F403",  # Allow wildcard imports to avoid cluttering the index
+  "F405",  # Allow names defined in via a wildcard import
+]
 
 [tool.ruff.lint.pylint]
 max-args = 12
@@ -94,3 +98,7 @@ known-first-party = ["pyavd", "schema_tools"]
 
 [tool.ruff.format]
 docstring-code-format = true
+
+[tool.ruff.lint.flake8-type-checking]
+# These classes require that type annotations be available at runtime
+runtime-evaluated-base-classes = ["pydantic.BaseModel"]

python-avd/pyavd/__init__.py

@@ -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 .get_avd_facts import get_avd_facts
+from .get_device_anta_catalog import get_device_anta_catalog
 from .get_device_config import get_device_config
 from .get_device_doc import get_device_doc
 from .get_device_structured_config import get_device_structured_config
@@ -21,6 +22,7 @@
 
 __all__ = [
     "get_avd_facts",
+    "get_device_anta_catalog",
     "get_device_config",
     "get_device_doc",
     "get_device_structured_config",

python-avd/pyavd/_anta/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (c) 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.

python-avd/pyavd/_anta/input_factories/__init__.py

@@ -0,0 +1,23 @@
+# 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.
+"""Input factories for the ANTA tests."""
+
+from __future__ import annotations
+
+from .connectivity import VerifyLLDPNeighborsInputFactory, VerifyReachabilityInputFactory
+from .hardware import VerifyEnvironmentCoolingInputFactory, VerifyEnvironmentPowerInputFactory, VerifyTransceiversManufacturersInputFactory
+from .interfaces import VerifyInterfacesStatusInputFactory
+from .routing_bgp import VerifyBGPSpecificPeersInputFactory
+from .routing_generic import VerifyRoutingTableEntryInputFactory
+
+__all__ = [
+    "VerifyLLDPNeighborsInputFactory",
+    "VerifyReachabilityInputFactory",
+    "VerifyEnvironmentCoolingInputFactory",
+    "VerifyEnvironmentPowerInputFactory",
+    "VerifyTransceiversManufacturersInputFactory",
+    "VerifyInterfacesStatusInputFactory",
+    "VerifyBGPSpecificPeersInputFactory",
+    "VerifyRoutingTableEntryInputFactory",
+]

python-avd/pyavd/_anta/input_factories/connectivity.py

@@ -0,0 +1,170 @@
+# 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 annotations
+
+from ipaddress import ip_interface
+from typing import TYPE_CHECKING
+
+from pyavd._anta.utils import LogMessage
+from pyavd._utils import get, validate_dict
+
+if TYPE_CHECKING:
+    from anta.tests.connectivity import VerifyLLDPNeighbors, VerifyReachability
+
+    from pyavd._anta.utils import TestLoggerAdapter
+    from pyavd._anta.utils.config_manager import ConfigManager
+
+
+class VerifyLLDPNeighborsInputFactory:
+    """Input factory class for the VerifyLLDPNeighbors test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyLLDPNeighbors], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyLLDPNeighbors.Input | None:
+        """Create Input for the VerifyLLDPNeighbors test."""
+        ethernet_interfaces = get(manager.structured_config, "ethernet_interfaces", [])
+
+        neighbors = []
+        required_keys = ["peer", "peer_interface"]
+        required_key_values = {"shutdown": False}
+
+        for interface in ethernet_interfaces:
+            if manager.is_subinterface(interface):
+                logger.info(LogMessage.SUBINTERFACE, entity=interface["name"])
+                continue
+
+            manager.update_interface_shutdown(interface)
+
+            is_valid, issues = validate_dict(interface, required_keys, required_key_values)
+            if not is_valid:
+                logger.info(LogMessage.INVALID_DATA, entity=interface["name"], issues=issues)
+                continue
+
+            if not manager.is_peer_available(peer := interface["peer"]):
+                logger.info(LogMessage.UNAVAILABLE_PEER, entity=interface["name"], peer=peer)
+                continue
+
+            if (dns_domain := get(manager.fabric_data.structured_configs[peer], "dns_domain")) is not None:
+                peer = f"{peer}.{dns_domain}"
+
+            neighbors.append(
+                test.Input.Neighbor(
+                    port=interface["name"],
+                    neighbor_device=peer,
+                    neighbor_port=interface["peer_interface"],
+                ),
+            )
+
+        return test.Input(neighbors=neighbors) if neighbors else None
+
+
+class VerifyReachabilityInputFactory:
+    """Input factory class for the VerifyReachability test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyReachability], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyReachability.Input | None:
+        """Create Input for the VerifyReachability test."""
+        # Get the eligible source IPs and VRFs
+        inband_mgmt_svis = cls._get_inband_mgmt_svis(manager, logger=logger.add_context(context="Inband MGMT"))
+        vtep_loopback0s = cls._get_vtep_loopback0s(manager, logger=logger.add_context(context="VTEP Loopback0"))
+
+        # Generate the hosts from the eligible sources and remote loopback0 interfaces from the mapping
+        hosts = []
+        for dst_node, dst_ip in manager.fabric_data.loopback0_mapping.items():
+            if not manager.is_peer_available(dst_node):
+                logger.info(LogMessage.UNAVAILABLE_PEER, entity=f"Destination {dst_ip}", peer=dst_node)
+                continue
+
+            hosts.extend([test.Input.Host(**source_vrf, destination=dst_ip, repeat=1) for source_vrf in inband_mgmt_svis + vtep_loopback0s])
+
+        # Add the P2P hosts
+        hosts.extend(cls._get_p2p_hosts(test, manager, logger=logger.add_context(context="P2P")))
+
+        return test.Input(hosts=hosts) if hosts else None
+
+    @staticmethod
+    def _get_p2p_hosts(test: type[VerifyReachability], manager: ConfigManager, logger: TestLoggerAdapter) -> list[VerifyReachability.Input.Host]:
+        """Generate the P2P hosts for the VerifyReachability test."""
+        ethernet_interfaces = get(manager.structured_config, "ethernet_interfaces", default=[])
+
+        hosts = []
+        required_keys = ["peer", "peer_interface", "ip_address"]
+        required_key_values = {"type": "routed", "shutdown": False}
+
+        for interface in ethernet_interfaces:
+            manager.update_interface_shutdown(interface)
+
+            is_valid, issues = validate_dict(interface, required_keys, required_key_values)
+            if not is_valid:
+                logger.info(LogMessage.INVALID_DATA, entity=interface["name"], issues=issues)
+                continue
+
+            if not manager.is_peer_available(peer := interface["peer"]):
+                logger.info(LogMessage.UNAVAILABLE_PEER, entity=interface["name"], peer=peer)
+                continue
+
+            if (
+                peer_interface_ip := manager.get_interface_ip(interface_model="ethernet_interfaces", interface_name=interface["peer_interface"], device=peer)
+            ) is None:
+                logger.info(LogMessage.UNAVAILABLE_PEER_IP, entity=interface["name"], peer=peer, peer_interface=interface["peer_interface"])
+                continue
+
+            hosts.append(
+                test.Input.Host(
+                    source=ip_interface(interface["ip_address"]).ip,
+                    destination=ip_interface(peer_interface_ip).ip,
+                    vrf="default",
+                    repeat=1,
+                ),
+            )
+
+        if not hosts:
+            logger.info(LogMessage.NO_SOURCES, entity="P2P")
+
+        return hosts
+
+    @staticmethod
+    def _get_inband_mgmt_svis(manager: ConfigManager, logger: TestLoggerAdapter) -> list[dict]:
+        """Generate the source IPs and VRFs from inband management SVIs for the VerifyReachability test."""
+        vlan_interfaces = get(manager.structured_config, "vlan_interfaces", default=[])
+
+        svis = []
+        required_keys = ["ip_address"]
+        required_key_values = {"type": "inband_mgmt", "shutdown": False}
+
+        for svi in vlan_interfaces:
+            manager.update_interface_shutdown(svi)
+
+            is_valid, issues = validate_dict(svi, required_keys, required_key_values)
+            if not is_valid:
+                logger.info(LogMessage.INVALID_DATA, entity=svi["name"], issues=issues)
+                continue
+
+            vrf = get(svi, "vrf", default="default")
+
+            svis.append({"source": ip_interface(svi["ip_address"]).ip, "vrf": vrf})
+
+        if not svis:
+            logger.info(LogMessage.NO_SOURCES, entity="inband management SVI")
+
+        return svis
+
+    @staticmethod
+    def _get_vtep_loopback0s(manager: ConfigManager, logger: TestLoggerAdapter) -> list[dict]:
+        """Generate the source IPs and VRFs from loopback0 interfaces of VTEPs for the VerifyReachability test."""
+        vtep_loopback0s = []
+
+        # TODO: Improve the VTEP logic
+        if not manager.is_vtep():
+            logger.info(LogMessage.NOT_VTEP)
+        elif manager.is_wan_vtep():
+            logger.info(LogMessage.WAN_VTEP)
+        elif (loopback0_ip := manager.get_interface_ip(interface_model="loopback_interfaces", interface_name="Loopback0")) is None:
+            logger.info(LogMessage.UNAVAILABLE_IP, entity="Loopback0")
+        else:
+            vtep_loopback0s.append({"source": ip_interface(loopback0_ip).ip, "vrf": "default"})
+
+        if not vtep_loopback0s:
+            logger.info(LogMessage.NO_SOURCES, entity="Loopback0")
+
+        return vtep_loopback0s

python-avd/pyavd/_anta/input_factories/constants.py

@@ -0,0 +1,27 @@
+# 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 annotations
+
+INTERFACE_MODELS = [
+    "ethernet_interfaces",
+    "port_channel_interfaces",
+    "vlan_interfaces",
+    "loopback_interfaces",
+    "dps_interfaces",
+]
+"""List of interface models from the structured configurations that are used for testing."""
+
+BGP_MAPPINGS = [
+    {"afi": "evpn", "safi": None, "description": "EVPN", "avd_key": "address_family_evpn"},
+    {"afi": "path-selection", "safi": None, "description": "Path-Selection", "avd_key": "address_family_path_selection"},
+    {"afi": "link-state", "safi": None, "description": "Link-State", "avd_key": "address_family_link_state"},
+    {"afi": "ipv4", "safi": "unicast", "description": "IPv4 Unicast", "avd_key": "address_family_ipv4"},
+    {"afi": "ipv6", "safi": "unicast", "description": "IPv6 Unicast", "avd_key": "address_family_ipv6"},
+    {"afi": "ipv4", "safi": "sr-te", "description": "IPv4 SR-TE", "avd_key": "address_family_ipv4_sr_te"},
+    {"afi": "ipv6", "safi": "sr-te", "description": "IPv6 SR-TE", "avd_key": "address_family_ipv6_sr_te"},
+]
+"""
+List of dictionaries that maps the BGP Address Family Identifier (AFI) and the Subsequent Address Family Identifier (SAFI) for validation.
+Each dictionary includes a description formatted for input messages in the report and an avd_key to access the address families in the structured configuration.
+"""

python-avd/pyavd/_anta/input_factories/hardware.py

@@ -0,0 +1,53 @@
+# 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 annotations
+
+from typing import TYPE_CHECKING
+
+from pyavd._utils import get
+
+if TYPE_CHECKING:
+    from anta.tests.hardware import VerifyEnvironmentCooling, VerifyEnvironmentPower, VerifyTransceiversManufacturers
+
+    from pyavd._anta.utils import ConfigManager, TestLoggerAdapter
+
+
+class VerifyEnvironmentPowerInputFactory:
+    """Input factory class for the VerifyEnvironmentPower test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyEnvironmentPower], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyEnvironmentPower.Input:
+        """Create Input for the VerifyEnvironmentPower test."""
+        _ = logger
+
+        pwr_supply_states = get(manager.structured_config, "accepted_pwr_supply_states", ["ok"])
+
+        return test.Input(states=pwr_supply_states)
+
+
+class VerifyEnvironmentCoolingInputFactory:
+    """Input factory class for the VerifyEnvironmentCooling test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyEnvironmentCooling], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyEnvironmentCooling.Input:
+        """Create Input for the VerifyEnvironmentCooling test."""
+        _ = logger
+
+        fan_states = get(manager.structured_config, "accepted_fan_states", ["ok"])
+
+        return test.Input(states=fan_states)
+
+
+class VerifyTransceiversManufacturersInputFactory:
+    """Input factory class for the VerifyTransceiversManufacturers test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyTransceiversManufacturers], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyTransceiversManufacturers.Input:
+        """Create Input for the VerifyTransceiversManufacturers test."""
+        _ = logger
+
+        xcvr_manufacturers = get(manager.structured_config, "accepted_xcvr_manufacturers", ["Arista Networks", "Arastra, Inc."])
+        xcvr_manufacturers.append("Not Present")
+
+        return test.Input(manufacturers=xcvr_manufacturers)

python-avd/pyavd/_anta/input_factories/interfaces.py

@@ -0,0 +1,58 @@
+# 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 annotations
+
+from typing import TYPE_CHECKING
+
+from pyavd._anta.utils import LogMessage
+from pyavd._utils import get
+
+from .constants import INTERFACE_MODELS
+
+if TYPE_CHECKING:
+    from anta.tests.interfaces import VerifyInterfacesStatus
+
+    from pyavd._anta.utils import ConfigManager, TestLoggerAdapter
+
+
+class VerifyInterfacesStatusInputFactory:
+    """Input factory class for the VerifyInterfacesStatus test."""
+
+    @classmethod
+    def create(cls, test: type[VerifyInterfacesStatus], manager: ConfigManager, logger: TestLoggerAdapter) -> VerifyInterfacesStatus.Input | None:
+        """Create Input for the VerifyInterfacesStatus test."""
+        inputs = []
+
+        for interface_model in INTERFACE_MODELS:
+            if (interfaces := get(manager.structured_config, interface_model)) is None:
+                logger.info(LogMessage.NO_DATA_MODEL, entity=interface_model)
+                continue
+
+            for interface in interfaces:
+                manager.update_interface_shutdown(interface)
+
+                if not manager.to_be_validated(interface):
+                    logger.info(LogMessage.SKIP_INTERFACE, entity=interface["name"])
+                    continue
+
+                status = "adminDown" if interface["shutdown"] else "up"
+
+                inputs.append(
+                    test.Input.InterfaceState(
+                        name=interface["name"],
+                        status=status,
+                    ),
+                )
+
+        # If the device is a VTEP, add the Vxlan1 interface to the list of interfaces to check
+        # TODO: Check if we want to add log here
+        if manager.is_vtep():
+            inputs.append(
+                test.Input.InterfaceState(
+                    name="Vxlan1",
+                    status="up",
+                ),
+            )
+
+        return test.Input(interfaces=inputs) if inputs else None

python-avd/pyavd/_anta/input_factories/protocols.py

@@ -0,0 +1,19 @@
+# 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 annotations
+
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from anta.models import AntaTest
+
+    from pyavd._anta.utils import ConfigManager, TestLoggerAdapter
+
+
+@runtime_checkable
+class AntaTestInputFactory(Protocol):
+    """Protocol for all AntaTest.Input factories available in this package."""
+
+    @classmethod
+    def create(cls, test: type[AntaTest], manager: ConfigManager, logger: TestLoggerAdapter) -> AntaTest.Input | None: ...