From 3c91d9e47f9b8b1714e46e4ef70361d63c423106 Mon Sep 17 00:00:00 2001 From: vitthalmagadum Date: Tue, 11 Feb 2025 05:50:49 -0500 Subject: [PATCH 1/2] Updated input models in SR Tunnel test --- anta/input_models/routing/isis.py | 62 ++++++++++++++++++++- anta/tests/routing/isis.py | 48 +++++----------- tests/units/anta_tests/routing/test_isis.py | 20 +++---- 3 files changed, 85 insertions(+), 45 deletions(-) diff --git a/anta/input_models/routing/isis.py b/anta/input_models/routing/isis.py index efeefe604..8ec78a7d0 100644 --- a/anta/input_models/routing/isis.py +++ b/anta/input_models/routing/isis.py @@ -5,7 +5,7 @@ from __future__ import annotations -from ipaddress import IPv4Address +from ipaddress import IPv4Address, IPv4Network from typing import Any, Literal from warnings import warn @@ -122,3 +122,63 @@ def __init__(self, **data: Any) -> None: # noqa: ANN401 stacklevel=2, ) super().__init__(**data) + + +class SRTunnelEntry(BaseModel): + """Model for a IS-IS SR tunnel.""" + + model_config = ConfigDict(extra="forbid") + endpoint: IPv4Network + """Endpoint of the tunnel.""" + vias: list[TunnelPath] | None = None + """Optional list of path to reach endpoint.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the SRTunnelEntry for reporting.""" + return f"Endpoint: {self.endpoint}" + + +class TunnelPath(BaseModel): + """Model for a IS-IS tunnel path.""" + + model_config = ConfigDict(extra="forbid") + nexthop: IPv4Address | None = None + """Nexthop of the tunnel.""" + type: Literal["ip", "tunnel"] | None = None + """Type of the tunnel.""" + interface: Interface | None = None + """Interface of the tunnel.""" + tunnel_id: Literal["TI-LFA", "ti-lfa", "unset"] | None = None + """Computation method of the tunnel.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the TunnelPath for reporting.""" + base_string = "" + if self.nexthop: + base_string += f" Next-hop: {self.nexthop}" + if self.type: + base_string += f" Type: {self.type}" + if self.interface: + base_string += f" Interface: {self.interface}" + if self.tunnel_id: + base_string += f" TunnelID: {self.tunnel_id}" + + return base_string.lstrip() + + +class Entry(SRTunnelEntry): # pragma: no cover + """Alias for the SRTunnelEntry model to maintain backward compatibility. + + When initialized, it will emit a deprecation warning and call the SRTunnelEntry model. + + TODO: Remove this class in ANTA v2.0.0. + """ + + def __init__(self, **data: Any) -> None: # noqa: ANN401 + """Initialize the Entry class, emitting a deprecation warning.""" + warn( + message="Entry model is deprecated and will be removed in ANTA v2.0.0. Use the SRTunnelEntry model instead.", + category=DeprecationWarning, + stacklevel=2, + ) + super().__init__(**data) diff --git a/anta/tests/routing/isis.py b/anta/tests/routing/isis.py index 47803e3a2..b994b41e0 100644 --- a/anta/tests/routing/isis.py +++ b/anta/tests/routing/isis.py @@ -7,13 +7,11 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from ipaddress import IPv4Address, IPv4Network -from typing import Any, ClassVar, Literal +from typing import Any, ClassVar -from pydantic import BaseModel, field_validator +from pydantic import field_validator -from anta.custom_types import Interface -from anta.input_models.routing.isis import InterfaceCount, InterfaceState, ISISInstance, IsisInstance, ISISInterface +from anta.input_models.routing.isis import Entry, InterfaceCount, InterfaceState, ISISInstance, IsisInstance, ISISInterface, SRTunnelEntry, TunnelPath from anta.models import AntaCommand, AntaTemplate, AntaTest from anta.tools import get_item, get_value @@ -391,32 +389,13 @@ class VerifyISISSegmentRoutingTunnels(AntaTest): class Input(AntaTest.Input): """Input model for the VerifyISISSegmentRoutingTunnels test.""" - entries: list[Entry] + entries: list[SRTunnelEntry] """List of tunnels to check on device.""" + Entry: ClassVar[type[Entry]] = Entry - class Entry(BaseModel): - """Definition of a tunnel entry.""" - - endpoint: IPv4Network - """Endpoint IP of the tunnel.""" - vias: list[Vias] | None = None - """Optional list of path to reach endpoint.""" - - class Vias(BaseModel): - """Definition of a tunnel path.""" - - nexthop: IPv4Address | None = None - """Nexthop of the tunnel. If None, then it is not tested. Default: None""" - type: Literal["ip", "tunnel"] | None = None - """Type of the tunnel. If None, then it is not tested. Default: None""" - interface: Interface | None = None - """Interface of the tunnel. If None, then it is not tested. Default: None""" - tunnel_id: Literal["TI-LFA", "ti-lfa", "unset"] | None = None - """Computation method of the tunnel. If None, then it is not tested. Default: None""" - - def _eos_entry_lookup(self, search_value: IPv4Network, entries: dict[str, Any], search_key: str = "endpoint") -> dict[str, Any] | None: + def _eos_entry_lookup(self, search_value: str, entries: dict[str, Any], search_key: str = "endpoint") -> dict[str, Any] | None: return next( - (entry_value for entry_id, entry_value in entries.items() if str(entry_value[search_key]) == str(search_value)), + (entry_value for entry_id, entry_value in entries.items() if str(entry_value[search_key]) == search_value), None, ) @@ -431,25 +410,26 @@ def test(self) -> None: self.result.is_success() if len(command_output["entries"]) == 0: - self.result.is_skipped("IS-IS-SR is not running on device.") + self.result.is_skipped("IS-IS-SR not configured") return for input_entry in self.inputs.entries: - eos_entry = self._eos_entry_lookup(search_value=input_entry.endpoint, entries=command_output["entries"]) + eos_entry = self._eos_entry_lookup(search_value=str(input_entry.endpoint), entries=command_output["entries"]) if eos_entry is None: - self.result.is_failure(f"Tunnel to {input_entry.endpoint!s} is not found.") + self.result.is_failure(f"{input_entry} - Tunnel not found") + elif input_entry.vias is not None: for via_input in input_entry.vias: via_search_result = any(self._via_matches(via_input, eos_via) for eos_via in eos_entry["vias"]) if not via_search_result: - self.result.is_failure(f"Tunnel to {input_entry.endpoint!s} is incorrect.") + self.result.is_failure(f"{input_entry} {via_input} - Tunnel is incorrect") - def _via_matches(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_via: dict[str, Any]) -> bool: + def _via_matches(self, via_input: TunnelPath, eos_via: dict[str, Any]) -> bool: """Check if the via input matches the eos via. Parameters ---------- - via_input : VerifyISISSegmentRoutingTunnels.Input.Entry.Vias + via_input : TunnelPath The input via to check. eos_via : dict[str, Any] The EOS via to compare against. diff --git a/tests/units/anta_tests/routing/test_isis.py b/tests/units/anta_tests/routing/test_isis.py index 733f5710b..87e80bcf2 100644 --- a/tests/units/anta_tests/routing/test_isis.py +++ b/tests/units/anta_tests/routing/test_isis.py @@ -1812,7 +1812,7 @@ }, "expected": { "result": "skipped", - "messages": ["IS-IS-SR is not running on device."], + "messages": ["IS-IS-SR not configured"], }, }, { @@ -1841,7 +1841,7 @@ }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.122/32 is not found."], + "messages": ["Endpoint: 1.0.0.122/32 - Tunnel not found"], }, }, { @@ -1922,7 +1922,7 @@ }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.13/32 is incorrect."], + "messages": ["Endpoint: 1.0.0.13/32 Type: tunnel - Tunnel is incorrect"], }, }, { @@ -2010,7 +2010,7 @@ }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.122/32 is incorrect."], + "messages": ["Endpoint: 1.0.0.122/32 Next-hop: 10.0.1.2 Type: ip Interface: Ethernet1 - Tunnel is incorrect"], }, }, { @@ -2098,7 +2098,7 @@ }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.122/32 is incorrect."], + "messages": ["Endpoint: 1.0.0.122/32 Next-hop: 10.0.1.1 Type: ip Interface: Ethernet4 - Tunnel is incorrect"], }, }, { @@ -2186,7 +2186,7 @@ }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.122/32 is incorrect."], + "messages": ["Endpoint: 1.0.0.122/32 Next-hop: 10.0.1.2 Type: ip Interface: Ethernet1 - Tunnel is incorrect"], }, }, { @@ -2251,7 +2251,7 @@ "vias": [ { "type": "tunnel", - "tunnelId": {"type": "TI-LFA", "index": 4}, + "tunnelId": {"type": "unset", "index": 4}, "labels": ["3"], } ], @@ -2266,14 +2266,14 @@ { "endpoint": "1.0.0.111/32", "vias": [ - {"type": "tunnel", "tunnel_id": "unset"}, + {"type": "tunnel", "tunnel_id": "ti-lfa"}, ], }, ] }, "expected": { "result": "failure", - "messages": ["Tunnel to 1.0.0.111/32 is incorrect."], + "messages": ["Endpoint: 1.0.0.111/32 Type: tunnel TunnelID: ti-lfa - Tunnel is incorrect"], }, }, { @@ -2294,7 +2294,7 @@ }, "expected": { "result": "skipped", - "messages": ["IS-IS-SR is not running on device."], + "messages": ["IS-IS-SR not configured"], }, }, ] From 5842f0cac7d58f353654578aad221cdb8977b709 Mon Sep 17 00:00:00 2001 From: vitthalmagadum Date: Tue, 11 Feb 2025 07:15:38 -0500 Subject: [PATCH 2/2] Updated unit tests for input models --- tests/units/input_models/routing/test_isis.py | 35 +++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/tests/units/input_models/routing/test_isis.py b/tests/units/input_models/routing/test_isis.py index f22bfa6fd..ddf5e623c 100644 --- a/tests/units/input_models/routing/test_isis.py +++ b/tests/units/input_models/routing/test_isis.py @@ -5,15 +5,18 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal import pytest from pydantic import ValidationError +from anta.input_models.routing.isis import ISISInstance, TunnelPath from anta.tests.routing.isis import VerifyISISSegmentRoutingAdjacencySegments, VerifyISISSegmentRoutingDataplane if TYPE_CHECKING: - from anta.input_models.routing.isis import ISISInstance + from ipaddress import IPv4Address + + from anta.custom_types import Interface class TestVerifyISISSegmentRoutingAdjacencySegmentsInput: @@ -68,3 +71,31 @@ def test_invalid(self, instances: list[ISISInstance]) -> None: """Test VerifyISISSegmentRoutingDataplane.Input invalid inputs.""" with pytest.raises(ValidationError): VerifyISISSegmentRoutingDataplane.Input(instances=instances) + + +class TestTunnelPath: + """Test anta.input_models.routing.isis.TestTunnelPath.""" + + # pylint: disable=too-few-public-methods + + @pytest.mark.parametrize( + ("nexthop", "type", "interface", "tunnel_id", "expected"), + [ + pytest.param("1.1.1.1", None, None, None, "Next-hop: 1.1.1.1", id="nexthop"), + pytest.param(None, "ip", None, None, "Type: ip", id="type"), + pytest.param(None, None, "Et1", None, "Interface: Ethernet1", id="interface"), + pytest.param(None, None, None, "TI-LFA", "TunnelID: TI-LFA", id="tunnel_id"), + pytest.param("1.1.1.1", "ip", "Et1", "TI-LFA", "Next-hop: 1.1.1.1 Type: ip Interface: Ethernet1 TunnelID: TI-LFA", id="all"), + pytest.param(None, None, None, None, "", id="None"), + ], + ) + def test_valid__str__( + self, + nexthop: IPv4Address | None, + type: Literal["ip", "tunnel"] | None, # noqa: A002 + interface: Interface | None, + tunnel_id: Literal["TI-LFA", "ti-lfa", "unset"] | None, + expected: str, + ) -> None: + """Test TunnelPath __str__.""" + assert str(TunnelPath(nexthop=nexthop, type=type, interface=interface, tunnel_id=tunnel_id)) == expected