2025-02-05 11:32:35 +01:00
|
|
|
# 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.
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Module related to various Spanning Tree Protocol (STP) tests."""
|
|
|
|
|
2025-02-05 11:32:35 +01:00
|
|
|
# Mypy does not understand AntaTest.Input typing
|
|
|
|
# mypy: disable-error-code=attr-defined
|
|
|
|
from __future__ import annotations
|
|
|
|
|
2025-02-05 11:39:09 +01:00
|
|
|
from typing import ClassVar, Literal
|
|
|
|
|
|
|
|
from pydantic import Field
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
from anta.custom_types import Vlan
|
|
|
|
from anta.models import AntaCommand, AntaTemplate, AntaTest
|
2025-02-05 11:39:09 +01:00
|
|
|
from anta.tools import get_value
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
|
|
|
|
class VerifySTPMode(AntaTest):
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Verifies the configured STP mode for a provided list of VLAN(s).
|
|
|
|
|
|
|
|
Expected Results
|
|
|
|
----------------
|
|
|
|
* Success: The test will pass if the STP mode is configured properly in the specified VLAN(s).
|
|
|
|
* Failure: The test will fail if the STP mode is NOT configured properly for one or more specified VLAN(s).
|
|
|
|
|
|
|
|
Examples
|
|
|
|
--------
|
|
|
|
```yaml
|
|
|
|
anta.tests.stp:
|
|
|
|
- VerifySTPMode:
|
|
|
|
mode: rapidPvst
|
|
|
|
vlans:
|
|
|
|
- 10
|
|
|
|
- 20
|
|
|
|
```
|
2025-02-05 11:32:35 +01:00
|
|
|
"""
|
|
|
|
|
|
|
|
name = "VerifySTPMode"
|
|
|
|
description = "Verifies the configured STP mode for a provided list of VLAN(s)."
|
2025-02-05 11:39:09 +01:00
|
|
|
categories: ClassVar[list[str]] = ["stp"]
|
|
|
|
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show spanning-tree vlan {vlan}", revision=1)]
|
|
|
|
|
|
|
|
class Input(AntaTest.Input):
|
|
|
|
"""Input model for the VerifySTPMode test."""
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
mode: Literal["mstp", "rstp", "rapidPvst"] = "mstp"
|
2025-02-05 11:39:09 +01:00
|
|
|
"""STP mode to verify. Supported values: mstp, rstp, rapidPvst. Defaults to mstp."""
|
|
|
|
vlans: list[Vlan]
|
|
|
|
"""List of VLAN on which to verify STP mode."""
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
def render(self, template: AntaTemplate) -> list[AntaCommand]:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Render the template for each VLAN in the input list."""
|
2025-02-05 11:32:35 +01:00
|
|
|
return [template.render(vlan=vlan) for vlan in self.inputs.vlans]
|
|
|
|
|
|
|
|
@AntaTest.anta_test
|
|
|
|
def test(self) -> None:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Main test function for VerifySTPMode."""
|
2025-02-05 11:32:35 +01:00
|
|
|
not_configured = []
|
|
|
|
wrong_stp_mode = []
|
|
|
|
for command in self.instance_commands:
|
2025-02-05 11:39:09 +01:00
|
|
|
vlan_id = command.params.vlan
|
|
|
|
if not (
|
|
|
|
stp_mode := get_value(
|
|
|
|
command.json_output,
|
|
|
|
f"spanningTreeVlanInstances.{vlan_id}.spanningTreeVlanInstance.protocol",
|
|
|
|
)
|
|
|
|
):
|
2025-02-05 11:32:35 +01:00
|
|
|
not_configured.append(vlan_id)
|
|
|
|
elif stp_mode != self.inputs.mode:
|
|
|
|
wrong_stp_mode.append(vlan_id)
|
|
|
|
if not_configured:
|
|
|
|
self.result.is_failure(f"STP mode '{self.inputs.mode}' not configured for the following VLAN(s): {not_configured}")
|
|
|
|
if wrong_stp_mode:
|
|
|
|
self.result.is_failure(f"Wrong STP mode configured for the following VLAN(s): {wrong_stp_mode}")
|
|
|
|
if not not_configured and not wrong_stp_mode:
|
|
|
|
self.result.is_success()
|
|
|
|
|
|
|
|
|
|
|
|
class VerifySTPBlockedPorts(AntaTest):
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Verifies there is no STP blocked ports.
|
|
|
|
|
|
|
|
Expected Results
|
|
|
|
----------------
|
|
|
|
* Success: The test will pass if there are NO ports blocked by STP.
|
|
|
|
* Failure: The test will fail if there are ports blocked by STP.
|
|
|
|
|
|
|
|
Examples
|
|
|
|
--------
|
|
|
|
```yaml
|
|
|
|
anta.tests.stp:
|
|
|
|
- VerifySTPBlockedPorts:
|
|
|
|
```
|
2025-02-05 11:32:35 +01:00
|
|
|
"""
|
|
|
|
|
|
|
|
name = "VerifySTPBlockedPorts"
|
|
|
|
description = "Verifies there is no STP blocked ports."
|
2025-02-05 11:39:09 +01:00
|
|
|
categories: ClassVar[list[str]] = ["stp"]
|
|
|
|
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show spanning-tree blockedports", revision=1)]
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
@AntaTest.anta_test
|
|
|
|
def test(self) -> None:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Main test function for VerifySTPBlockedPorts."""
|
2025-02-05 11:32:35 +01:00
|
|
|
command_output = self.instance_commands[0].json_output
|
|
|
|
if not (stp_instances := command_output["spanningTreeInstances"]):
|
|
|
|
self.result.is_success()
|
|
|
|
else:
|
|
|
|
for key, value in stp_instances.items():
|
|
|
|
stp_instances[key] = value.pop("spanningTreeBlockedPorts")
|
|
|
|
self.result.is_failure(f"The following ports are blocked by STP: {stp_instances}")
|
|
|
|
|
|
|
|
|
|
|
|
class VerifySTPCounters(AntaTest):
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Verifies there is no errors in STP BPDU packets.
|
|
|
|
|
|
|
|
Expected Results
|
|
|
|
----------------
|
|
|
|
* Success: The test will pass if there are NO STP BPDU packet errors under all interfaces participating in STP.
|
|
|
|
* Failure: The test will fail if there are STP BPDU packet errors on one or many interface(s).
|
|
|
|
|
|
|
|
Examples
|
|
|
|
--------
|
|
|
|
```yaml
|
|
|
|
anta.tests.stp:
|
|
|
|
- VerifySTPCounters:
|
|
|
|
```
|
2025-02-05 11:32:35 +01:00
|
|
|
"""
|
|
|
|
|
|
|
|
name = "VerifySTPCounters"
|
|
|
|
description = "Verifies there is no errors in STP BPDU packets."
|
2025-02-05 11:39:09 +01:00
|
|
|
categories: ClassVar[list[str]] = ["stp"]
|
|
|
|
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show spanning-tree counters", revision=1)]
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
@AntaTest.anta_test
|
|
|
|
def test(self) -> None:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Main test function for VerifySTPCounters."""
|
2025-02-05 11:32:35 +01:00
|
|
|
command_output = self.instance_commands[0].json_output
|
|
|
|
interfaces_with_errors = [
|
|
|
|
interface for interface, counters in command_output["interfaces"].items() if counters["bpduTaggedError"] or counters["bpduOtherError"] != 0
|
|
|
|
]
|
|
|
|
if interfaces_with_errors:
|
|
|
|
self.result.is_failure(f"The following interfaces have STP BPDU packet errors: {interfaces_with_errors}")
|
|
|
|
else:
|
|
|
|
self.result.is_success()
|
|
|
|
|
|
|
|
|
|
|
|
class VerifySTPForwardingPorts(AntaTest):
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Verifies that all interfaces are in a forwarding state for a provided list of VLAN(s).
|
|
|
|
|
|
|
|
Expected Results
|
|
|
|
----------------
|
|
|
|
* Success: The test will pass if all interfaces are in a forwarding state for the specified VLAN(s).
|
|
|
|
* Failure: The test will fail if one or many interfaces are NOT in a forwarding state in the specified VLAN(s).
|
|
|
|
|
|
|
|
Examples
|
|
|
|
--------
|
|
|
|
```yaml
|
|
|
|
anta.tests.stp:
|
|
|
|
- VerifySTPForwardingPorts:
|
|
|
|
vlans:
|
|
|
|
- 10
|
|
|
|
- 20
|
|
|
|
```
|
2025-02-05 11:32:35 +01:00
|
|
|
"""
|
|
|
|
|
|
|
|
name = "VerifySTPForwardingPorts"
|
|
|
|
description = "Verifies that all interfaces are forwarding for a provided list of VLAN(s)."
|
2025-02-05 11:39:09 +01:00
|
|
|
categories: ClassVar[list[str]] = ["stp"]
|
|
|
|
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show spanning-tree topology vlan {vlan} status", revision=1)]
|
|
|
|
|
|
|
|
class Input(AntaTest.Input):
|
|
|
|
"""Input model for the VerifySTPForwardingPorts test."""
|
2025-02-05 11:32:35 +01:00
|
|
|
|
2025-02-05 11:39:09 +01:00
|
|
|
vlans: list[Vlan]
|
|
|
|
"""List of VLAN on which to verify forwarding states."""
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
def render(self, template: AntaTemplate) -> list[AntaCommand]:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Render the template for each VLAN in the input list."""
|
2025-02-05 11:32:35 +01:00
|
|
|
return [template.render(vlan=vlan) for vlan in self.inputs.vlans]
|
|
|
|
|
|
|
|
@AntaTest.anta_test
|
|
|
|
def test(self) -> None:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Main test function for VerifySTPForwardingPorts."""
|
2025-02-05 11:32:35 +01:00
|
|
|
not_configured = []
|
|
|
|
not_forwarding = []
|
|
|
|
for command in self.instance_commands:
|
2025-02-05 11:39:09 +01:00
|
|
|
vlan_id = command.params.vlan
|
2025-02-05 11:32:35 +01:00
|
|
|
if not (topologies := get_value(command.json_output, "topologies")):
|
|
|
|
not_configured.append(vlan_id)
|
|
|
|
else:
|
|
|
|
for value in topologies.values():
|
2025-02-05 11:39:09 +01:00
|
|
|
if vlan_id and int(vlan_id) in value["vlans"]:
|
2025-02-05 11:32:35 +01:00
|
|
|
interfaces_not_forwarding = [interface for interface, state in value["interfaces"].items() if state["state"] != "forwarding"]
|
|
|
|
if interfaces_not_forwarding:
|
|
|
|
not_forwarding.append({f"VLAN {vlan_id}": interfaces_not_forwarding})
|
|
|
|
if not_configured:
|
|
|
|
self.result.is_failure(f"STP instance is not configured for the following VLAN(s): {not_configured}")
|
|
|
|
if not_forwarding:
|
2025-02-05 11:39:09 +01:00
|
|
|
self.result.is_failure(f"The following VLAN(s) have interface(s) that are not in a forwarding state: {not_forwarding}")
|
2025-02-05 11:32:35 +01:00
|
|
|
if not not_configured and not interfaces_not_forwarding:
|
|
|
|
self.result.is_success()
|
|
|
|
|
|
|
|
|
|
|
|
class VerifySTPRootPriority(AntaTest):
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Verifies the STP root priority for a provided list of VLAN or MST instance ID(s).
|
|
|
|
|
|
|
|
Expected Results
|
|
|
|
----------------
|
|
|
|
* Success: The test will pass if the STP root priority is configured properly for the specified VLAN or MST instance ID(s).
|
|
|
|
* Failure: The test will fail if the STP root priority is NOT configured properly for the specified VLAN or MST instance ID(s).
|
|
|
|
|
|
|
|
Examples
|
|
|
|
--------
|
|
|
|
```yaml
|
|
|
|
anta.tests.stp:
|
|
|
|
- VerifySTPRootPriority:
|
|
|
|
priority: 32768
|
|
|
|
instances:
|
|
|
|
- 10
|
|
|
|
- 20
|
|
|
|
```
|
2025-02-05 11:32:35 +01:00
|
|
|
"""
|
|
|
|
|
|
|
|
name = "VerifySTPRootPriority"
|
|
|
|
description = "Verifies the STP root priority for a provided list of VLAN or MST instance ID(s)."
|
2025-02-05 11:39:09 +01:00
|
|
|
categories: ClassVar[list[str]] = ["stp"]
|
|
|
|
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show spanning-tree root detail", revision=1)]
|
|
|
|
|
|
|
|
class Input(AntaTest.Input):
|
|
|
|
"""Input model for the VerifySTPRootPriority test."""
|
2025-02-05 11:32:35 +01:00
|
|
|
|
|
|
|
priority: int
|
2025-02-05 11:39:09 +01:00
|
|
|
"""STP root priority to verify."""
|
|
|
|
instances: list[Vlan] = Field(default=[])
|
2025-02-05 11:32:35 +01:00
|
|
|
"""List of VLAN or MST instance ID(s). If empty, ALL VLAN or MST instance ID(s) will be verified."""
|
|
|
|
|
|
|
|
@AntaTest.anta_test
|
|
|
|
def test(self) -> None:
|
2025-02-05 11:39:09 +01:00
|
|
|
"""Main test function for VerifySTPRootPriority."""
|
2025-02-05 11:32:35 +01:00
|
|
|
command_output = self.instance_commands[0].json_output
|
|
|
|
if not (stp_instances := command_output["instances"]):
|
|
|
|
self.result.is_failure("No STP instances configured")
|
|
|
|
return
|
|
|
|
# Checking the type of instances based on first instance
|
2025-02-05 11:39:09 +01:00
|
|
|
first_name = next(iter(stp_instances))
|
2025-02-05 11:32:35 +01:00
|
|
|
if first_name.startswith("MST"):
|
|
|
|
prefix = "MST"
|
|
|
|
elif first_name.startswith("VL"):
|
|
|
|
prefix = "VL"
|
|
|
|
else:
|
|
|
|
self.result.is_failure(f"Unsupported STP instance type: {first_name}")
|
|
|
|
return
|
|
|
|
check_instances = [f"{prefix}{instance_id}" for instance_id in self.inputs.instances] if self.inputs.instances else command_output["instances"].keys()
|
|
|
|
wrong_priority_instances = [
|
|
|
|
instance for instance in check_instances if get_value(command_output, f"instances.{instance}.rootBridge.priority") != self.inputs.priority
|
|
|
|
]
|
|
|
|
if wrong_priority_instances:
|
|
|
|
self.result.is_failure(f"The following instance(s) have the wrong STP root priority configured: {wrong_priority_instances}")
|
|
|
|
else:
|
|
|
|
self.result.is_success()
|