1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
|
# 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.
"""
Test functions related to Multi-chassis Link Aggregation (MLAG)
"""
# Mypy does not understand AntaTest.Input typing
# mypy: disable-error-code=attr-defined
from __future__ import annotations
from pydantic import conint
from anta.custom_types import MlagPriority
from anta.models import AntaCommand, AntaTest
from anta.tools.get_value import get_value
class VerifyMlagStatus(AntaTest):
"""
This test verifies the health status of the MLAG configuration.
Expected Results:
* success: The test will pass if the MLAG state is 'active', negotiation status is 'connected',
peer-link status and local interface status are 'up'.
* failure: The test will fail if the MLAG state is not 'active', negotiation status is not 'connected',
peer-link status or local interface status are not 'up'.
* skipped: The test will be skipped if MLAG is 'disabled'.
"""
name = "VerifyMlagStatus"
description = "Verifies the health status of the MLAG configuration."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag", ofmt="json")]
@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
if command_output["state"] == "disabled":
self.result.is_skipped("MLAG is disabled")
return
keys_to_verify = ["state", "negStatus", "localIntfStatus", "peerLinkStatus"]
verified_output = {key: get_value(command_output, key) for key in keys_to_verify}
if (
verified_output["state"] == "active"
and verified_output["negStatus"] == "connected"
and verified_output["localIntfStatus"] == "up"
and verified_output["peerLinkStatus"] == "up"
):
self.result.is_success()
else:
self.result.is_failure(f"MLAG status is not OK: {verified_output}")
class VerifyMlagInterfaces(AntaTest):
"""
This test verifies there are no inactive or active-partial MLAG ports.
Expected Results:
* success: The test will pass if there are NO inactive or active-partial MLAG ports.
* failure: The test will fail if there are inactive or active-partial MLAG ports.
* skipped: The test will be skipped if MLAG is 'disabled'.
"""
name = "VerifyMlagInterfaces"
description = "Verifies there are no inactive or active-partial MLAG ports."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag", ofmt="json")]
@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
if command_output["state"] == "disabled":
self.result.is_skipped("MLAG is disabled")
return
if command_output["mlagPorts"]["Inactive"] == 0 and command_output["mlagPorts"]["Active-partial"] == 0:
self.result.is_success()
else:
self.result.is_failure(f"MLAG status is not OK: {command_output['mlagPorts']}")
class VerifyMlagConfigSanity(AntaTest):
"""
This test verifies there are no MLAG config-sanity inconsistencies.
Expected Results:
* success: The test will pass if there are NO MLAG config-sanity inconsistencies.
* failure: The test will fail if there are MLAG config-sanity inconsistencies.
* skipped: The test will be skipped if MLAG is 'disabled'.
* error: The test will give an error if 'mlagActive' is not found in the JSON response.
"""
name = "VerifyMlagConfigSanity"
description = "Verifies there are no MLAG config-sanity inconsistencies."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag config-sanity", ofmt="json")]
@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
if (mlag_status := get_value(command_output, "mlagActive")) is None:
self.result.is_error(message="Incorrect JSON response - 'mlagActive' state was not found")
return
if mlag_status is False:
self.result.is_skipped("MLAG is disabled")
return
keys_to_verify = ["globalConfiguration", "interfaceConfiguration"]
verified_output = {key: get_value(command_output, key) for key in keys_to_verify}
if not any(verified_output.values()):
self.result.is_success()
else:
self.result.is_failure(f"MLAG config-sanity returned inconsistencies: {verified_output}")
class VerifyMlagReloadDelay(AntaTest):
"""
This test verifies the reload-delay parameters of the MLAG configuration.
Expected Results:
* success: The test will pass if the reload-delay parameters are configured properly.
* failure: The test will fail if the reload-delay parameters are NOT configured properly.
* skipped: The test will be skipped if MLAG is 'disabled'.
"""
name = "VerifyMlagReloadDelay"
description = "Verifies the MLAG reload-delay parameters."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag", ofmt="json")]
class Input(AntaTest.Input): # pylint: disable=missing-class-docstring
reload_delay: conint(ge=0) # type: ignore
"""Delay (seconds) after reboot until non peer-link ports that are part of an MLAG are enabled"""
reload_delay_non_mlag: conint(ge=0) # type: ignore
"""Delay (seconds) after reboot until ports that are not part of an MLAG are enabled"""
@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
if command_output["state"] == "disabled":
self.result.is_skipped("MLAG is disabled")
return
keys_to_verify = ["reloadDelay", "reloadDelayNonMlag"]
verified_output = {key: get_value(command_output, key) for key in keys_to_verify}
if verified_output["reloadDelay"] == self.inputs.reload_delay and verified_output["reloadDelayNonMlag"] == self.inputs.reload_delay_non_mlag:
self.result.is_success()
else:
self.result.is_failure(f"The reload-delay parameters are not configured properly: {verified_output}")
class VerifyMlagDualPrimary(AntaTest):
"""
This test verifies the dual-primary detection and its parameters of the MLAG configuration.
Expected Results:
* success: The test will pass if the dual-primary detection is enabled and its parameters are configured properly.
* failure: The test will fail if the dual-primary detection is NOT enabled or its parameters are NOT configured properly.
* skipped: The test will be skipped if MLAG is 'disabled'.
"""
name = "VerifyMlagDualPrimary"
description = "Verifies the MLAG dual-primary detection parameters."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag detail", ofmt="json")]
class Input(AntaTest.Input): # pylint: disable=missing-class-docstring
detection_delay: conint(ge=0) # type: ignore
"""Delay detection (seconds)"""
errdisabled: bool = False
"""Errdisabled all interfaces when dual-primary is detected"""
recovery_delay: conint(ge=0) # type: ignore
"""Delay (seconds) after dual-primary detection resolves until non peer-link ports that are part of an MLAG are enabled"""
recovery_delay_non_mlag: conint(ge=0) # type: ignore
"""Delay (seconds) after dual-primary detection resolves until ports that are not part of an MLAG are enabled"""
@AntaTest.anta_test
def test(self) -> None:
errdisabled_action = "errdisableAllInterfaces" if self.inputs.errdisabled else "none"
command_output = self.instance_commands[0].json_output
if command_output["state"] == "disabled":
self.result.is_skipped("MLAG is disabled")
return
if command_output["dualPrimaryDetectionState"] == "disabled":
self.result.is_failure("Dual-primary detection is disabled")
return
keys_to_verify = ["detail.dualPrimaryDetectionDelay", "detail.dualPrimaryAction", "dualPrimaryMlagRecoveryDelay", "dualPrimaryNonMlagRecoveryDelay"]
verified_output = {key: get_value(command_output, key) for key in keys_to_verify}
if (
verified_output["detail.dualPrimaryDetectionDelay"] == self.inputs.detection_delay
and verified_output["detail.dualPrimaryAction"] == errdisabled_action
and verified_output["dualPrimaryMlagRecoveryDelay"] == self.inputs.recovery_delay
and verified_output["dualPrimaryNonMlagRecoveryDelay"] == self.inputs.recovery_delay_non_mlag
):
self.result.is_success()
else:
self.result.is_failure(f"The dual-primary parameters are not configured properly: {verified_output}")
class VerifyMlagPrimaryPriority(AntaTest):
"""
Test class to verify the MLAG (Multi-Chassis Link Aggregation) primary priority.
Expected Results:
* Success: The test will pass if the MLAG state is set as 'primary' and the priority matches the input.
* Failure: The test will fail if the MLAG state is not 'primary' or the priority doesn't match the input.
* Skipped: The test will be skipped if MLAG is 'disabled'.
"""
name = "VerifyMlagPrimaryPriority"
description = "Verifies the configuration of the MLAG primary priority."
categories = ["mlag"]
commands = [AntaCommand(command="show mlag detail")]
class Input(AntaTest.Input):
"""Inputs for the VerifyMlagPrimaryPriority test."""
primary_priority: MlagPriority
"""The expected MLAG primary priority."""
@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
self.result.is_success()
# Skip the test if MLAG is disabled
if command_output["state"] == "disabled":
self.result.is_skipped("MLAG is disabled")
return
mlag_state = get_value(command_output, "detail.mlagState")
primary_priority = get_value(command_output, "detail.primaryPriority")
# Check MLAG state
if mlag_state != "primary":
self.result.is_failure("The device is not set as MLAG primary.")
# Check primary priority
if primary_priority != self.inputs.primary_priority:
self.result.is_failure(
f"The primary priority does not match expected. Expected `{self.inputs.primary_priority}`, but found `{primary_priority}` instead."
)
|
