diff --git a/anta/custom_types.py b/anta/custom_types.py index c9cb1789f..8ca3c1c2f 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -55,7 +55,7 @@ def interface_autocomplete(v: str) -> str: raise ValueError(msg) intf_id = m[0] - alias_map = {"et": "Ethernet", "eth": "Ethernet", "po": "Port-Channel", "lo": "Loopback"} + alias_map = {"et": "Ethernet", "eth": "Ethernet", "po": "Port-Channel", "lo": "Loopback", "vl": "Vlan"} return next((f"{full_name}{intf_id}" for alias, full_name in alias_map.items() if v.lower().startswith(alias)), v) diff --git a/anta/input_models/connectivity.py b/anta/input_models/connectivity.py index 53581ea3c..1a904ac1d 100644 --- a/anta/input_models/connectivity.py +++ b/anta/input_models/connectivity.py @@ -5,7 +5,7 @@ from __future__ import annotations -from ipaddress import IPv4Address +from ipaddress import IPv4Address, IPv6Address from typing import Any from warnings import warn @@ -18,10 +18,10 @@ class Host(BaseModel): """Model for a remote host to ping.""" model_config = ConfigDict(extra="forbid") - destination: IPv4Address - """IPv4 address to ping.""" - source: IPv4Address | Interface - """IPv4 address source IP or egress interface to use.""" + destination: IPv4Address | IPv6Address + """Destination address to ping.""" + source: IPv4Address | IPv6Address | Interface + """Source address IP or egress interface to use.""" vrf: str = "default" """VRF context. Defaults to `default`.""" repeat: int = 2 diff --git a/anta/tests/connectivity.py b/anta/tests/connectivity.py index 245baf46d..2fd58c1bb 100644 --- a/anta/tests/connectivity.py +++ b/anta/tests/connectivity.py @@ -7,11 +7,16 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from typing import ClassVar +from typing import ClassVar, TypeVar + +from pydantic import field_validator from anta.input_models.connectivity import Host, LLDPNeighbor, Neighbor from anta.models import AntaCommand, AntaTemplate, AntaTest +# Using a TypeVar for the Host model since mypy thinks it's a ClassVar and not a valid type when used in field validators +T = TypeVar("T", bound=Host) + class VerifyReachability(AntaTest): """Test network reachability to one or many destination IP(s). @@ -37,6 +42,11 @@ class VerifyReachability(AntaTest): vrf: MGMT df_bit: True size: 100 + - source: fd12:3456:789a:1::1 + destination: fd12:3456:789a:1::2 + vrf: default + df_bit: True + size: 100 ``` """ @@ -54,6 +64,16 @@ class Input(AntaTest.Input): Host: ClassVar[type[Host]] = Host """To maintain backward compatibility.""" + @field_validator("hosts") + @classmethod + def validate_hosts(cls, hosts: list[T]) -> list[T]: + """Validate the 'destination' and 'source' IP address family in each host.""" + for host in hosts: + if not isinstance(host.source, str) and host.destination.version != host.source.version: + msg = f"{host} IP address family for destination does not match source" + raise ValueError(msg) + return hosts + def render(self, template: AntaTemplate) -> list[AntaCommand]: """Render the template for each host in the input list.""" return [ diff --git a/examples/tests.yaml b/examples/tests.yaml index 058166295..b190ec3ce 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -135,6 +135,11 @@ anta.tests.connectivity: vrf: MGMT df_bit: True size: 100 + - source: fd12:3456:789a:1::1 + destination: fd12:3456:789a:1::2 + vrf: default + df_bit: True + size: 100 anta.tests.cvx: - VerifyActiveCVXConnections: # Verifies the number of active CVX Connections. diff --git a/tests/units/anta_tests/test_connectivity.py b/tests/units/anta_tests/test_connectivity.py index 0e37e053b..d367258e1 100644 --- a/tests/units/anta_tests/test_connectivity.py +++ b/tests/units/anta_tests/test_connectivity.py @@ -45,6 +45,46 @@ ], "expected": {"result": "success"}, }, + { + "name": "success-ipv6", + "test": VerifyReachability, + "eos_data": [ + { + "messages": [ + """PING fd12:3456:789a:1::2(fd12:3456:789a:1::2) from fd12:3456:789a:1::1 : 52 data bytes + 60 bytes from fd12:3456:789a:1::2: icmp_seq=1 ttl=64 time=0.097 ms + 60 bytes from fd12:3456:789a:1::2: icmp_seq=2 ttl=64 time=0.033 ms + + --- fd12:3456:789a:1::2 ping statistics --- + 2 packets transmitted, 2 received, 0% packet loss, time 0ms + rtt min/avg/max/mdev = 0.033/0.065/0.097/0.032 ms, ipg/ewma 0.148/0.089 ms + """, + ], + }, + ], + "inputs": {"hosts": [{"destination": "fd12:3456:789a:1::2", "source": "fd12:3456:789a:1::1"}]}, + "expected": {"result": "success"}, + }, + { + "name": "success-ipv6-vlan", + "test": VerifyReachability, + "eos_data": [ + { + "messages": [ + """PING fd12:3456:789a:1::2(fd12:3456:789a:1::2) 52 data bytes + 60 bytes from fd12:3456:789a:1::2: icmp_seq=1 ttl=64 time=0.094 ms + 60 bytes from fd12:3456:789a:1::2: icmp_seq=2 ttl=64 time=0.027 ms + + --- fd12:3456:789a:1::2 ping statistics --- + 2 packets transmitted, 2 received, 0% packet loss, time 0ms + rtt min/avg/max/mdev = 0.027/0.060/0.094/0.033 ms, ipg/ewma 0.152/0.085 ms + """, + ], + }, + ], + "inputs": {"hosts": [{"destination": "fd12:3456:789a:1::2", "source": "vl110"}]}, + "expected": {"result": "success"}, + }, { "name": "success-interface", "test": VerifyReachability, @@ -155,6 +195,23 @@ ], "expected": {"result": "failure", "messages": ["Host 10.0.0.11 (src: 10.0.0.5, vrf: default, size: 100B, repeat: 2) - Unreachable"]}, }, + { + "name": "failure-ipv6", + "test": VerifyReachability, + "eos_data": [ + { + "messages": [ + """PING fd12:3456:789a:1::2(fd12:3456:789a:1::2) from fd12:3456:789a:1::1 : 52 data bytes + + --- fd12:3456:789a:1::3 ping statistics --- + 2 packets transmitted, 0 received, 100% packet loss, time 10ms + """, + ], + }, + ], + "inputs": {"hosts": [{"destination": "fd12:3456:789a:1::2", "source": "fd12:3456:789a:1::1"}]}, + "expected": {"result": "failure", "messages": ["Host fd12:3456:789a:1::2 (src: fd12:3456:789a:1::1, vrf: default, size: 100B, repeat: 2) - Unreachable"]}, + }, { "name": "failure-interface", "test": VerifyReachability, diff --git a/tests/units/input_models/test_connectivity.py b/tests/units/input_models/test_connectivity.py new file mode 100644 index 000000000..9e4288cf8 --- /dev/null +++ b/tests/units/input_models/test_connectivity.py @@ -0,0 +1,43 @@ +# Copyright (c) 2023-2025 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Tests for anta.input_models.connectivity.py.""" + +# pylint: disable=C0302 +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from pydantic import ValidationError + +from anta.tests.connectivity import VerifyReachability + +if TYPE_CHECKING: + from anta.input_models.connectivity import Host + + +class TestVerifyReachabilityInput: + """Test anta.tests.connectivity.VerifyReachability.Input.""" + + @pytest.mark.parametrize( + ("hosts"), + [ + pytest.param([{"destination": "fd12:3456:789a:1::2", "source": "fd12:3456:789a:1::1"}], id="valid"), + ], + ) + def test_valid(self, hosts: list[Host]) -> None: + """Test VerifyReachability.Input valid inputs.""" + VerifyReachability.Input(hosts=hosts) + + @pytest.mark.parametrize( + ("hosts"), + [ + pytest.param([{"destination": "fd12:3456:789a:1::2", "source": "192.168.0.10"}], id="invalid-source"), + pytest.param([{"destination": "192.168.0.10", "source": "fd12:3456:789a:1::2"}], id="invalid-destination"), + ], + ) + def test_invalid(self, hosts: list[Host]) -> None: + """Test VerifyReachability.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyReachability.Input(hosts=hosts) diff --git a/tests/units/test_custom_types.py b/tests/units/test_custom_types.py index b82686b3f..5a92092c7 100644 --- a/tests/units/test_custom_types.py +++ b/tests/units/test_custom_types.py @@ -189,6 +189,7 @@ def test_interface_autocomplete_success() -> None: assert interface_autocomplete("lo4") == "Loopback4" assert interface_autocomplete("Po1000") == "Port-Channel1000" assert interface_autocomplete("Po 1000") == "Port-Channel1000" + assert interface_autocomplete("Vl1000") == "Vlan1000" def test_interface_autocomplete_no_alias() -> None: