Repository: aristanetworks/anta (python)

Issue:
feat(anta.tests): The source field of VerifyReachability Host model should be optional
When `source` is `None`, that means we don't provide a source to the ping command and let EOS decides the source.

Pull request:
feat(anta.tests): The source field of VerifyReachability Host model should be optional
# Description

When source is None, that means we don't provide a source to the ping command and let EOS decides the source.

Fixes #1112

# Checklist:

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [X] I have made corresponding changes to the documentation
- [X] I have added tests that prove my fix is effective or that my feature works
- [X] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 4b734f4b..499528ff 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -56,7 +56,7 @@ repos:
         name: Run Ruff formatter
 
   - repo: https://github.com/pycqa/pylint
-    rev: "v3.3.5"
+    rev: "v3.3.6"
     hooks:
       - id: pylint
         name: Check code style with pylint
diff --git a/anta/input_models/connectivity.py b/anta/input_models/connectivity.py
index 464d22a5..2d4f0e9f 100644
--- a/anta/input_models/connectivity.py
+++ b/anta/input_models/connectivity.py
@@ -20,8 +20,8 @@ class Host(BaseModel):
     model_config = ConfigDict(extra="forbid")
     destination: IPv4Address | IPv6Address
     """Destination address to ping."""
-    source: IPv4Address | IPv6Address | Interface
-    """Source address IP or egress interface to use."""
+    source: IPv4Address | IPv6Address | Interface | None = None
+    """Source address IP or egress interface to use. Can be provided in the `VerifyReachability` test."""
     vrf: str = "default"
     """VRF context."""
     repeat: int = 2
@@ -38,10 +38,15 @@ class Host(BaseModel):
 
         Examples
         --------
-        Host: 10.1.1.1 Source: 10.2.2.2 VRF: mgmt
+        - Host: 10.1.1.1 Source: 10.2.2.2 VRF: mgmt
+        - Host: 10.1.1.1 VRF: mgmt
 
         """
-        return f"Host: {self.destination} Source: {self.source} VRF: {self.vrf}"
+        base_string = f"Host: {self.destination}"
+        if self.source:
+            base_string += f" Source: {self.source}"
+        base_string += f" VRF: {self.vrf}"
+        return base_string
 
 
 class LLDPNeighbor(BaseModel):


Test patch:
diff --git a/anta/tests/connectivity.py b/anta/tests/connectivity.py
index a5ba5fff..17e7eaf2 100644
--- a/anta/tests/connectivity.py
+++ b/anta/tests/connectivity.py
@@ -38,8 +38,7 @@ class VerifyReachability(AntaTest):
               df_bit: True
               size: 100
               reachable: true
-            - source: Management0
-              destination: 8.8.8.8
+            - destination: 8.8.8.8
               vrf: MGMT
               df_bit: True
               size: 100
@@ -55,7 +54,7 @@ class VerifyReachability(AntaTest):
     categories: ClassVar[list[str]] = ["connectivity"]
     # Template uses '{size}{df_bit}' without space since df_bit includes leading space when enabled
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
-        AntaTemplate(template="ping vrf {vrf} {destination} source {source} size {size}{df_bit} repeat {repeat}", revision=1)
+        AntaTemplate(template="ping vrf {vrf} {destination}{source} size {size}{df_bit} repeat {repeat}", revision=1)
     ]
 
     class Input(AntaTest.Input):
@@ -71,7 +70,7 @@ class VerifyReachability(AntaTest):
         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:
+                if host.source and 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
@@ -80,7 +79,12 @@ class VerifyReachability(AntaTest):
         """Render the template for each host in the input list."""
         return [
             template.render(
-                destination=host.destination, source=host.source, vrf=host.vrf, repeat=host.repeat, size=host.size, df_bit=" df-bit" if host.df_bit else ""
+                destination=host.destination,
+                source=f" source {host.source}" if host.source else "",
+                vrf=host.vrf,
+                repeat=host.repeat,
+                size=host.size,
+                df_bit=" df-bit" if host.df_bit else "",
             )
             for host in self.inputs.hosts
         ]
diff --git a/anta/tests/field_notices.py b/anta/tests/field_notices.py
index cc7fab9f..ed8022fa 100644
--- a/anta/tests/field_notices.py
+++ b/anta/tests/field_notices.py
@@ -38,7 +38,7 @@ class VerifyFieldNotice44Resolution(AntaTest):
     categories: ClassVar[list[str]] = ["field notices"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show version detail", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyFieldNotice44Resolution."""
@@ -142,7 +142,7 @@ class VerifyFieldNotice72Resolution(AntaTest):
     categories: ClassVar[list[str]] = ["field notices"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show version detail", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyFieldNotice72Resolution."""
diff --git a/anta/tests/hardware.py b/anta/tests/hardware.py
index 7edd41b1..9e0453f9 100644
--- a/anta/tests/hardware.py
+++ b/anta/tests/hardware.py
@@ -45,7 +45,7 @@ class VerifyTransceiversManufacturers(AntaTest):
         manufacturers: list[str]
         """List of approved transceivers manufacturers."""
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyTransceiversManufacturers."""
@@ -78,7 +78,7 @@ class VerifyTemperature(AntaTest):
     categories: ClassVar[list[str]] = ["hardware"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment temperature", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyTemperature."""
@@ -108,7 +108,7 @@ class VerifyTransceiversTemperature(AntaTest):
     categories: ClassVar[list[str]] = ["hardware"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment temperature transceiver", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyTransceiversTemperature."""
@@ -141,7 +141,7 @@ class VerifyEnvironmentSystemCooling(AntaTest):
     categories: ClassVar[list[str]] = ["hardware"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment cooling", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyEnvironmentSystemCooling."""
@@ -179,7 +179,7 @@ class VerifyEnvironmentCooling(AntaTest):
         states: list[str]
         """List of accepted states of fan status."""
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyEnvironmentCooling."""
@@ -228,7 +228,7 @@ class VerifyEnvironmentPower(AntaTest):
         states: list[str]
         """List of accepted states list of power supplies status."""
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyEnvironmentPower."""
@@ -259,7 +259,7 @@ class VerifyAdverseDrops(AntaTest):
     categories: ClassVar[list[str]] = ["hardware"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show hardware counter drop", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyAdverseDrops."""
diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py
index e291bd68..282b9456 100644
--- a/anta/tests/interfaces.py
+++ b/anta/tests/interfaces.py
@@ -29,13 +29,16 @@ class VerifyInterfaceUtilization(AntaTest):
     """Verifies that the utilization of interfaces is below a certain threshold.
 
     Load interval (default to 5 minutes) is defined in device configuration.
-    This test has been implemented for full-duplex interfaces only.
+
+    !!! warning
+        This test has been implemented for full-duplex interfaces only.
 
     Expected Results
     ----------------
     * Success: The test will pass if all interfaces have a usage below the threshold.
-    * Failure: The test will fail if one or more interfaces have a usage above the threshold.
-    * Error: The test will error out if the device has at least one non full-duplex interface.
+    * Failure: If any of the following occur:
+        - One or more interfaces have a usage above the threshold.
+        - The device has at least one non full-duplex interface.
 
     Examples
     --------
@@ -56,7 +59,7 @@ class VerifyInterfaceUtilization(AntaTest):
         """Input model for the VerifyInterfaceUtilization test."""
 
         threshold: Percent = 75.0
-        """Interface utilization threshold above which the test will fail. Defaults to 75%."""
+        """Interface utilization threshold above which the test will fail."""
 
     @AntaTest.anta_test
     def test(self) -> None:
@@ -67,12 +70,19 @@ class VerifyInterfaceUtilization(AntaTest):
         interfaces = self.instance_commands[1].json_output
 
         for intf, rate in rates["interfaces"].items():
+            interface_data = []
             # The utilization logic has been implemented for full-duplex interfaces only
-            if ((duplex := (interface := interfaces["interfaces"][intf]).get("duplex", None)) is not None and duplex != duplex_full) or (
-                (members := interface.get("memberInterfaces", None)) is not None and any(stats["duplex"] != duplex_full for stats in members.values())
-            ):
-                self.result.is_failure(f"Interface {intf} or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented.")
-                return
+            if not all([duplex := (interface := interfaces["interfaces"][intf]).get("duplex", None), duplex == duplex_full]):
+                if (members := interface.get("memberInterfaces", None)) is None:
+                    self.result.is_failure(f"Interface: {intf} - Test not implemented for non-full-duplex interfaces - Expected: {duplex_full} Actual: {duplex}")
+                    continue
+                interface_data = [(member_interface, state) for member_interface, stats in members.items() if (state := stats["duplex"]) != duplex_full]
+
+            for member_interface in interface_data:
+                self.result.is_failure(
+                    f"Interface: {intf} Member Interface: {member_interface[0]} - Test not implemented for non-full-duplex interfaces - Expected: {duplex_full}"
+                    f" Actual: {member_interface[1]}"
+                )
 
             if (bandwidth := interfaces["interfaces"][intf]["bandwidth"]) == 0:
                 self.logger.debug("Interface %s has been ignored due to null bandwidth value", intf)
@@ -276,7 +286,7 @@ class VerifyStormControlDrops(AntaTest):
     categories: ClassVar[list[str]] = ["interfaces"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show storm-control", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyStormControlDrops."""
diff --git a/anta/tests/lanz.py b/anta/tests/lanz.py
index 33e5472c..0ace171b 100644
--- a/anta/tests/lanz.py
+++ b/anta/tests/lanz.py
@@ -34,7 +34,7 @@ class VerifyLANZ(AntaTest):
     categories: ClassVar[list[str]] = ["lanz"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show queue-monitor length status", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyLANZ."""
diff --git a/anta/tests/profiles.py b/anta/tests/profiles.py
index 12799480..7a22b7a0 100644
--- a/anta/tests/profiles.py
+++ b/anta/tests/profiles.py
@@ -43,7 +43,7 @@ class VerifyUnifiedForwardingTableMode(AntaTest):
         mode: Literal[0, 1, 2, 3, 4, "flexible"]
         """Expected UFT mode. Valid values are 0, 1, 2, 3, 4, or "flexible"."""
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyUnifiedForwardingTableMode."""
@@ -81,7 +81,7 @@ class VerifyTcamProfile(AntaTest):
         profile: str
         """Expected TCAM profile."""
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyTcamProfile."""
diff --git a/anta/tests/ptp.py b/anta/tests/ptp.py
index 309871be..ef8c0263 100644
--- a/anta/tests/ptp.py
+++ b/anta/tests/ptp.py
@@ -36,7 +36,7 @@ class VerifyPtpModeStatus(AntaTest):
     categories: ClassVar[list[str]] = ["ptp"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyPtpModeStatus."""
@@ -81,7 +81,7 @@ class VerifyPtpGMStatus(AntaTest):
     categories: ClassVar[list[str]] = ["ptp"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyPtpGMStatus."""
@@ -116,7 +116,7 @@ class VerifyPtpLockStatus(AntaTest):
     categories: ClassVar[list[str]] = ["ptp"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyPtpLockStatus."""
@@ -155,7 +155,7 @@ class VerifyPtpOffset(AntaTest):
     categories: ClassVar[list[str]] = ["ptp"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp monitor", revision=1)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyPtpOffset."""
@@ -196,7 +196,7 @@ class VerifyPtpPortModeStatus(AntaTest):
     categories: ClassVar[list[str]] = ["ptp"]
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)]
 
-    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"])
+    @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab", "vEOS"])
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyPtpPortModeStatus."""
diff --git a/examples/tests.yaml b/examples/tests.yaml
index 45fe99ab..e7a67852 100644
--- a/examples/tests.yaml
+++ b/examples/tests.yaml
@@ -131,8 +131,7 @@ anta.tests.connectivity:
           df_bit: True
           size: 100
           reachable: true
-        - source: Management0
-          destination: 8.8.8.8
+        - destination: 8.8.8.8
           vrf: MGMT
           df_bit: True
           size: 100
diff --git a/tests/units/anta_tests/test_connectivity.py b/tests/units/anta_tests/test_connectivity.py
index 7fb9a114..e51c9fbd 100644
--- a/tests/units/anta_tests/test_connectivity.py
+++ b/tests/units/anta_tests/test_connectivity.py
@@ -178,6 +178,26 @@ DATA: list[dict[str, Any]] = [
         ],
         "expected": {"result": "success"},
     },
+    {
+        "name": "success-without-source",
+        "test": VerifyReachability,
+        "inputs": {"hosts": [{"destination": "10.0.0.1", "repeat": 1}]},
+        "eos_data": [
+            {
+                "messages": [
+                    """PING 10.0.0.1 (10.0.0.1) : 72(100) bytes of data.
+                80 bytes from 10.0.0.1: icmp_seq=1 ttl=64 time=0.247 ms
+
+                --- 10.0.0.1 ping statistics ---
+                1 packets transmitted, 1 received, 0% packet loss, time 0ms
+                rtt min/avg/max/mdev = 0.072/0.159/0.247/0.088 ms, ipg/ewma 0.370/0.225 ms
+
+                """,
+                ],
+            },
+        ],
+        "expected": {"result": "success"},
+    },
     {
         "name": "failure-ip",
         "test": VerifyReachability,
@@ -309,6 +329,25 @@ DATA: list[dict[str, Any]] = [
             "messages": ["Host: 10.0.0.1 Source: 10.0.0.5 VRF: default - Destination is expected to be unreachable but found reachable"],
         },
     },
+    {
+        "name": "failure-without-source",
+        "test": VerifyReachability,
+        "inputs": {"hosts": [{"destination": "10.0.0.1", "repeat": 1}]},
+        "eos_data": [
+            {
+                "messages": [
+                    """ping: sendmsg: Network is unreachable
+                PING 10.0.0.1 (10.0.0.1) : 72(100) bytes of data.
+
+                --- 10.0.0.11 ping statistics ---
+                2 packets transmitted, 0 received, 100% packet loss, time 10ms
+
+                """,
+                ],
+            },
+        ],
+        "expected": {"result": "failure", "messages": ["Host: 10.0.0.1 VRF: default - Unreachable"]},
+    },
     {
         "name": "success",
         "test": VerifyLLDPNeighbors,
diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py
index 9f4ab116..778133b2 100644
--- a/tests/units/anta_tests/test_interfaces.py
+++ b/tests/units/anta_tests/test_interfaces.py
@@ -656,7 +656,7 @@ DATA: list[dict[str, Any]] = [
         "inputs": {"threshold": 70.0},
         "expected": {
             "result": "failure",
-            "messages": ["Interface Ethernet1/1 or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented"],
+            "messages": ["Interface: Ethernet1/1 - Test not implemented for non-full-duplex interfaces - Expected: duplexFull Actual: duplexHalf"],
         },
     },
     {
@@ -801,7 +801,12 @@ DATA: list[dict[str, Any]] = [
         "inputs": {"threshold": 70.0},
         "expected": {
             "result": "failure",
-            "messages": ["Interface Port-Channel31 or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented"],
+            "messages": [
+                "Interface: Port-Channel31 Member Interface: Ethernet3/1 - Test not implemented for non-full-duplex interfaces - Expected: duplexFull "
+                "Actual: duplexHalf",
+                "Interface: Port-Channel31 Member Interface: Ethernet4/1 - Test not implemented for non-full-duplex interfaces - Expected: duplexFull "
+                "Actual: duplexHalf",
+            ],
         },
     },
     {