feat(anta.tests): Add support for IPv6 for VerifyReachability test (#…

…1019) * Fix VerifyReachability for do not accept IPv6 input * Added validator for source and destination should beling to same IP add family * Revert Host ClassVar --------- Co-authored-by: Guillaume Mulocher <[email protected]> Co-authored-by: Carl Baillargeon <[email protected]>
aristanetworks · Feb 6, 2025 · f0ae643 · f0ae643
1 parent 7897ec5
commit f0ae643
Show file tree

Hide file tree

Showing 7 changed files with 133 additions and 7 deletions.
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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: