Skip to content

Commit

Permalink
Merge branch 'main' into fix_interface_module_failure_msg
Browse files Browse the repository at this point in the history
  • Loading branch information
geetanjalimanegslab authored Feb 21, 2025
2 parents 8e572a2 + 7c9bef4 commit 09d891a
Show file tree
Hide file tree
Showing 5 changed files with 140 additions and 56 deletions.
4 changes: 2 additions & 2 deletions .github/workflows/sonar.yml
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ jobs:
- uses: actions/checkout@v4
with:
ref: ${{ github.event.pull_request.head.sha }}
fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
- name: Setup Python
uses: actions/setup-python@v5
with:
Expand All @@ -30,7 +30,7 @@ jobs:
- name: "Run pytest via tox for ${{ matrix.python }}"
run: tox
- name: SonarCloud Scan
uses: SonarSource/sonarcloud-github-action@master
uses: SonarSource/sonarqube-scan-action@v5.0.0
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
Expand Down
80 changes: 79 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,81 @@ def __init__(self, **data: Any) -> None: # noqa: ANN401
stacklevel=2,
)
super().__init__(**data)


class Tunnel(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 paths to reach the endpoint."""

def __str__(self) -> str:
"""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" Tunnel ID: {self.tunnel_id}"

return base_string.lstrip()


class Entry(Tunnel): # pragma: no cover
"""Alias for the Tunnel model to maintain backward compatibility.
When initialized, it will emit a deprecation warning and call the Tunnel 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 Tunnel model instead.",
category=DeprecationWarning,
stacklevel=2,
)
super().__init__(**data)


class Vias(TunnelPath): # pragma: no cover
"""Alias for the TunnelPath model to maintain backward compatibility.
When initialized, it will emit a deprecation warning and call the TunnelPath model.
TODO: Remove this class in ANTA v2.0.0.
"""

def __init__(self, **data: Any) -> None: # noqa: ANN401
"""Initialize the Vias class, emitting a deprecation warning."""
warn(
message="Vias model is deprecated and will be removed in ANTA v2.0.0. Use the TunnelPath model instead.",
category=DeprecationWarning,
stacklevel=2,
)
super().__init__(**data)
57 changes: 16 additions & 41 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, Tunnel, TunnelPath
from anta.models import AntaCommand, AntaTemplate, AntaTest
from anta.tools import get_item, get_value

Expand Down Expand Up @@ -391,34 +389,9 @@ class VerifyISISSegmentRoutingTunnels(AntaTest):
class Input(AntaTest.Input):
"""Input model for the VerifyISISSegmentRoutingTunnels test."""

entries: list[Entry]
entries: list[Tunnel]
"""List of tunnels to check on device."""

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:
return next(
(entry_value for entry_id, entry_value in entries.items() if str(entry_value[search_key]) == str(search_value)),
None,
)
Entry: ClassVar[type[Entry]] = Entry

@AntaTest.anta_test
def test(self) -> None:
Expand All @@ -427,29 +400,31 @@ def test(self) -> None:
This method performs the main test logic for verifying ISIS Segment Routing tunnels.
It checks the command output, initiates defaults, and performs various checks on the tunnels.
"""
command_output = self.instance_commands[0].json_output
self.result.is_success()

command_output = self.instance_commands[0].json_output
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"])
if eos_entry is None:
self.result.is_failure(f"Tunnel to {input_entry.endpoint!s} is not found.")
elif input_entry.vias is not None:
entries = list(command_output["entries"].values())
if (eos_entry := get_item(entries, "endpoint", str(input_entry.endpoint))) is None:
self.result.is_failure(f"{input_entry} - Tunnel not found")
continue

if 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"],
},
},
{
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"],
},
},
{
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 Tunnel ID: 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", "Tunnel ID: 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 Tunnel ID: 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

0 comments on commit 09d891a

Please sign in to comment.