Skip to content

Commit

Permalink
fix(anta.tests): Implement logic to not test against all inputs for V…
Browse files Browse the repository at this point in the history 
…erifyISISSegmentRoutingTunnels (#1030)

* fix(anta.tests): Implement logic to not test against all inputs for VerifyISISSegmentRoutingTunnels

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(anta.tests): revert code linting and reduce to VerifyISISSegmentRoutingTunnels only

* fix(anta.tests): revert code linting and reduce to VerifyISISSegmentRoutingTunnels only

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(anta.tests): code linting

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
  • Loading branch information
@titom73 @pre-commit-ci
titom73 and pre-commit-ci[bot] authored Feb 6, 2025
1 parent f0ae643 commit aaf4b9b
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 123 deletions.
133 changes: 16 additions & 117 deletions anta/tests/routing/isis.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -592,139 +592,38 @@ def test(self) -> None:
command_output = self.instance_commands[0].json_output
self.result.is_success()

# initiate defaults
failure_message = []

if len(command_output["entries"]) == 0:
self.result.is_skipped("IS-IS-SR is not running on device.")
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:
failure_message.append(f"Tunnel to {input_entry} is not found.")
self.result.is_failure(f"Tunnel to {input_entry.endpoint!s} is not found.")
elif input_entry.vias is not None:
failure_src = []
for via_input in input_entry.vias:
if not self._check_tunnel_type(via_input, eos_entry):
failure_src.append("incorrect tunnel type")
if not self._check_tunnel_nexthop(via_input, eos_entry):
failure_src.append("incorrect nexthop")
if not self._check_tunnel_interface(via_input, eos_entry):
failure_src.append("incorrect interface")
if not self._check_tunnel_id(via_input, eos_entry):
failure_src.append("incorrect tunnel ID")

if failure_src:
failure_message.append(f"Tunnel to {input_entry.endpoint!s} is incorrect: {', '.join(failure_src)}")

if failure_message:
self.result.is_failure("\n".join(failure_message))

def _check_tunnel_type(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""Check if the tunnel type specified in `via_input` matches any of the tunnel types in `eos_entry`.
Parameters
----------
via_input : VerifyISISSegmentRoutingTunnels.Input.Entry.Vias
The input tunnel type to check.
eos_entry : dict[str, Any]
The EOS entry containing the tunnel types.
Returns
-------
bool
True if the tunnel type matches any of the tunnel types in `eos_entry`, False otherwise.
"""
if via_input.type is not None:
return any(
via_input.type
== get_value(
dictionary=eos_via,
key="type",
default="undefined",
)
for eos_via in eos_entry["vias"]
)
return True

def _check_tunnel_nexthop(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""Check if the tunnel nexthop matches the given input.
Parameters
----------
via_input : VerifyISISSegmentRoutingTunnels.Input.Entry.Vias
The input via object.
eos_entry : dict[str, Any]
The EOS entry dictionary.
Returns
-------
bool
True if the tunnel nexthop matches, False otherwise.
"""
if via_input.nexthop is not None:
return any(
str(via_input.nexthop)
== get_value(
dictionary=eos_via,
key="nexthop",
default="undefined",
)
for eos_via in eos_entry["vias"]
)
return True

def _check_tunnel_interface(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""Check if the tunnel interface exists in the given EOS entry.
Parameters
----------
via_input : VerifyISISSegmentRoutingTunnels.Input.Entry.Vias
The input via object.
eos_entry : dict[str, Any]
The EOS entry dictionary.
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.")

Returns
-------
bool
True if the tunnel interface exists, False otherwise.
"""
if via_input.interface is not None:
return any(
via_input.interface
== get_value(
dictionary=eos_via,
key="interface",
default="undefined",
)
for eos_via in eos_entry["vias"]
)
return True

def _check_tunnel_id(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""Check if the tunnel ID matches any of the tunnel IDs in the EOS entry's vias.
def _via_matches(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_via: dict[str, Any]) -> bool:
"""Check if the via input matches the eos via.
Parameters
----------
via_input : VerifyISISSegmentRoutingTunnels.Input.Entry.Vias
The input vias to check.
eos_entry : dict[str, Any])
The EOS entry to compare against.
The input via to check.
eos_via : dict[str, Any]
The EOS via to compare against.
Returns
-------
bool
True if the tunnel ID matches any of the tunnel IDs in the EOS entry's vias, False otherwise.
True if the via input matches the eos via, False otherwise.
"""
if via_input.tunnel_id is not None:
return any(
via_input.tunnel_id.upper()
== get_value(
dictionary=eos_via,
key="tunnelId.type",
default="undefined",
).upper()
for eos_via in eos_entry["vias"]
)
return True
return (
(via_input.type is None or via_input.type == eos_via.get("type"))
and (via_input.nexthop is None or str(via_input.nexthop) == eos_via.get("nexthop"))
and (via_input.interface is None or via_input.interface == eos_via.get("interface"))
and (via_input.tunnel_id is None or via_input.tunnel_id.upper() == get_value(eos_via, "tunnelId.type", default="").upper())
)
33 changes: 27 additions & 6 deletions tests/units/anta_tests/routing/test_isis.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1405,7 +1405,7 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to endpoint=IPv4Network('1.0.0.122/32') vias=None is not found."],
"messages": ["Tunnel to 1.0.0.122/32 is not found."],
},
},
{
Expand Down Expand Up @@ -1486,7 +1486,7 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to 1.0.0.13/32 is incorrect: incorrect tunnel type"],
"messages": ["Tunnel to 1.0.0.13/32 is incorrect."],
},
},
{
Expand Down Expand Up @@ -1574,7 +1574,7 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to 1.0.0.122/32 is incorrect: incorrect nexthop"],
"messages": ["Tunnel to 1.0.0.122/32 is incorrect."],
},
},
{
Expand Down Expand Up @@ -1662,7 +1662,7 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to 1.0.0.122/32 is incorrect: incorrect interface"],
"messages": ["Tunnel to 1.0.0.122/32 is incorrect."],
},
},
{
Expand Down Expand Up @@ -1750,7 +1750,7 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to 1.0.0.122/32 is incorrect: incorrect nexthop"],
"messages": ["Tunnel to 1.0.0.122/32 is incorrect."],
},
},
{
Expand Down Expand Up @@ -1837,7 +1837,28 @@
},
"expected": {
"result": "failure",
"messages": ["Tunnel to 1.0.0.111/32 is incorrect: incorrect tunnel ID"],
"messages": ["Tunnel to 1.0.0.111/32 is incorrect."],
},
},
{
"test": VerifyISISSegmentRoutingTunnels,
"name": "skipped with ISIS-SR not running",
"eos_data": [{"entries": {}}],
"inputs": {
"entries": [
{"endpoint": "1.0.0.122/32"},
{"endpoint": "1.0.0.13/32", "vias": [{"type": "ip"}]},
{
"endpoint": "1.0.0.111/32",
"vias": [
{"type": "tunnel", "tunnel_id": "unset"},
],
},
]
},
"expected": {
"result": "skipped",
"messages": ["IS-IS-SR is not running on device."],
},
},
]
Expand Down

0 comments on commit aaf4b9b

Please sign in to comment.