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

fix(anta.tests): VerifySpecificPath testcase with the latest changes #965

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 2 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
28 changes: 28 additions & 0 deletions anta/input_models/path_selection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# Copyright (c) 2023-2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""Module containing input models for path-selection tests."""

from __future__ import annotations

from ipaddress import IPv4Address

from pydantic import BaseModel, ConfigDict


class RouterPath(BaseModel):
"""Model for a list of router path entries."""

model_config = ConfigDict(extra="forbid")
peer: IPv4Address
"""Static peer IPv4 address."""
path_group: str
"""Router path group name."""
source_address: IPv4Address
"""Source IPv4 address of path."""
destination_address: IPv4Address
"""Destination IPv4 address of path."""

def __str__(self) -> str:
"""Return a human-readable string representation of the RouterPath for reporting."""
return f"Peer: {self.peer} PathGroup: {self.path_group} Source: {self.source_address} Destination: {self.destination_address}"
96 changes: 46 additions & 50 deletions anta/tests/path_selection.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,10 @@
# mypy: disable-error-code=attr-defined
from __future__ import annotations

from ipaddress import IPv4Address
from typing import ClassVar

from pydantic import BaseModel

from anta.decorators import skip_on_platforms
from anta.input_models.path_selection import RouterPath
from anta.models import AntaCommand, AntaTemplate, AntaTest
from anta.tools import get_value

Expand Down Expand Up @@ -70,16 +68,21 @@ def test(self) -> None:


class VerifySpecificPath(AntaTest):
"""Verifies the path and telemetry state of a specific path for an IPv4 peer under router path-selection.
"""Verifies the path and telemetry state of a specific path for an IPv4 peer.
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved

The expected states are 'IPsec established', 'Resolved' for path and 'active' for telemetry.
This test performs the following checks for each specified routerpath:
1. Verifies that the specified peer is configured.
2. Verifies that the specified path group is found.
3. Verifies that the expected source and destination address match the path group.
4. Verifies that the state of the path is `ipsecEstablished` or `routeResolved`.
5. Verifies that the telemetry state is `active`.

Expected Results
----------------
* Success: The test will pass if the path state under router path-selection is either 'IPsec established' or 'Resolved'
* Success: The test will pass if the path state under router path selection is either 'IPsecEstablished' or 'Resolved'
and telemetry state as 'active'.
* Failure: The test will fail if router path-selection is not configured or if the path state is not 'IPsec established' or 'Resolved',
or if the telemetry state is 'inactive'.
* Failure: The test will fail if router path selection is not configured, the path state is not 'IPsec established' or 'Resolved',
or the telemetry state is 'inactive'.

Examples
--------
Expand All @@ -95,65 +98,58 @@ class VerifySpecificPath(AntaTest):
"""

categories: ClassVar[list[str]] = ["path-selection"]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
AntaTemplate(template="show path-selection paths peer {peer} path-group {group} source {source} destination {destination}", revision=1)
]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show path-selection paths", revision=1)]

class Input(AntaTest.Input):
"""Input model for the VerifySpecificPath test."""

paths: list[RouterPath]
"""List of router paths to verify."""

class RouterPath(BaseModel):
"""Detail of a router path."""

peer: IPv4Address
"""Static peer IPv4 address."""

path_group: str
"""Router path group name."""

source_address: IPv4Address
"""Source IPv4 address of path."""

destination_address: IPv4Address
"""Destination IPv4 address of path."""

def render(self, template: AntaTemplate) -> list[AntaCommand]:
"""Render the template for each router path."""
return [
template.render(peer=path.peer, group=path.path_group, source=path.source_address, destination=path.destination_address) for path in self.inputs.paths
]
RouterPath: ClassVar[type[RouterPath]] = RouterPath

@skip_on_platforms(["cEOSLab", "vEOS-lab"])
@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifySpecificPath."""
self.result.is_success()

# Check the state of each path
for command in self.instance_commands:
peer = command.params.peer
path_group = command.params.group
source = command.params.source
destination = command.params.destination
command_output = command.json_output.get("dpsPeers", [])
command_output = self.instance_commands[0].json_output

# If the dpsPeers details are not found in the command output, the test fails.
if not (dps_peers_details := get_value(command_output, "dpsPeers")):
self.result.is_failure("Router path not configured")
return

# Iterating on each router path mentioned in the inputs.
for router_path in self.inputs.paths:
peer = str(router_path.peer)
path_group = router_path.path_group
source = str(router_path.source_address)
destination = str(router_path.destination_address)
peer_details = dps_peers_details.get(peer, {})

# If the peer is not configured for the path group, the test fails
if not command_output:
self.result.is_failure(f"Path `peer: {peer} source: {source} destination: {destination}` is not configured for path-group `{path_group}`.")
if not peer_details:
self.result.is_failure(f"{router_path} - Peer not found")
continue

path_group_details = get_value(peer_details, f"dpsGroups..{path_group}..dpsPaths", separator="..")
# If the expected pathgroup is not found for the peer, the test fails.
if not path_group_details:
self.result.is_failure(f"{router_path} - Path-group not found")
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved
continue

# Extract the state of the path
path_output = get_value(command_output, f"{peer}..dpsGroups..{path_group}..dpsPaths", separator="..")
path_state = next(iter(path_output.values())).get("state")
session = get_value(next(iter(path_output.values())), "dpsSessions.0.active")
path_data = next((path for path in path_group_details.values() if (path.get("source") == source and path.get("destination") == destination)), None)
# If the expected and actual source and destion address of the pathgroup are not matched, test fails.
if not path_data:
self.result.is_failure(f"{router_path} - Source and/or Destination address not found")
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved
continue

path_state = path_data.get("state")
session = get_value(path_data, "dpsSessions.0.active")
expected_state = ["ipsecEstablished", "routeResolved"]
# If the state of the path is not 'ipsecEstablished' or 'routeResolved', or the telemetry state is 'inactive', the test fails
if path_state not in ["ipsecEstablished", "routeResolved"]:
self.result.is_failure(f"Path state for `peer: {peer} source: {source} destination: {destination}` in path-group {path_group} is `{path_state}`.")
if path_state not in expected_state:
self.result.is_failure(f"{router_path} - Incorrect path state - Expected: {' or '.join(expected_state)} Actual: {path_state}")
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved
elif not session:
self.result.is_failure(
f"Telemetry state for path `peer: {peer} source: {source} destination: {destination}` in path-group {path_group} is `inactive`."
)
self.result.is_failure(f"{router_path} - Incorrect telemetry state - Expected: active Actual: inactive")
vitthalmagadum marked this conversation as resolved.
Show resolved Hide resolved
15 changes: 15 additions & 0 deletions docs/api/tests.path_selection.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ anta_title: ANTA catalog for Router path-selection tests
~ that can be found in the LICENSE file.
-->

# Tests

::: anta.tests.path_selection
options:
show_root_heading: false
Expand All @@ -18,3 +20,16 @@ anta_title: ANTA catalog for Router path-selection tests
filters:
- "!test"
- "!render"

# Input models

::: anta.input_models.path_selection

options:
show_root_heading: false
show_root_toc_entry: false
show_bases: false
merge_init_into_class: false
anta_hide_test_module_description: true
show_labels: true
filters: ["!^__str__"]
2 changes: 1 addition & 1 deletion examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -325,7 +325,7 @@ anta.tests.path_selection:
- VerifyPathsHealth:
# Verifies the path and telemetry state of all paths under router path-selection.
- VerifySpecificPath:
# Verifies the path and telemetry state of a specific path for an IPv4 peer under router path-selection.
# Verifies the path and telemetry state of a specific path for an IPv4 peer.
paths:
- peer: 10.255.0.1
path_group: internet
Expand Down
Loading
Loading