Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor(anta.tests): Refactor VerifyISISSegmentRoutingTunnels test case #1037

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 61 additions & 1 deletion anta/input_models/routing/isis.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -122,3 +122,63 @@ def __init__(self, **data: Any) -> None: # noqa: ANN401
stacklevel=2,
)
super().__init__(**data)


class SRTunnelEntry(BaseModel):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are already using Segment + TunnelPath so I think Tunnel makes sense.

Suggested change
class SRTunnelEntry(BaseModel):
class Tunnel(BaseModel):

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You also need to add a ClassVar for the deprecated Vias model for backward compatibility.

"""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."""
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
"""Optional list of path to reach endpoint."""
"""Optional list of paths to reach the endpoint."""


def __str__(self) -> str:
"""Return a human-readable string representation of the SRTunnelEntry for reporting."""
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
"""Return a human-readable string representation of the SRTunnelEntry for reporting."""
"""Return a human-readable string representation of the Tunnel 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}"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
base_string += f" TunnelID: {self.tunnel_id}"
base_string += f" Tunnel ID: {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)
48 changes: 14 additions & 34 deletions anta/tests/routing/isis.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need that function? Can we use get_item from our tools module instead?

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,
)

Expand All @@ -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.
Expand Down
20 changes: 10 additions & 10 deletions tests/units/anta_tests/routing/test_isis.py
Original file line number Diff line number Diff line change
Expand Up @@ -1812,7 +1812,7 @@
},
"expected": {
"result": "skipped",
"messages": ["IS-IS-SR is not running on device."],
"messages": ["IS-IS-SR not configured"],
},
},
{
Expand Down Expand Up @@ -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"],
},
},
{
Expand Down Expand Up @@ -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"],
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is incorrect here?

},
},
{
Expand Down Expand Up @@ -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"],
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same thing here. The user doesn't know what is incorrect about the tunnel.

},
},
{
Expand Down Expand Up @@ -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"],
},
},
{
Expand Down Expand Up @@ -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"],
},
},
{
Expand Down Expand Up @@ -2251,7 +2251,7 @@
"vias": [
{
"type": "tunnel",
"tunnelId": {"type": "TI-LFA", "index": 4},
"tunnelId": {"type": "unset", "index": 4},
"labels": ["3"],
}
],
Expand All @@ -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"],
},
},
{
Expand All @@ -2294,7 +2294,7 @@
},
"expected": {
"result": "skipped",
"messages": ["IS-IS-SR is not running on device."],
"messages": ["IS-IS-SR not configured"],
},
},
]
35 changes: 33 additions & 2 deletions tests/units/input_models/routing/test_isis.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Loading