Skip to content

MLAG

ANTA catalog for mlag tests

Test functions related to Multi-chassis Link Aggregation (MLAG)

VerifyMlagConfigSanity

Bases: 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.
Source code in anta/tests/mlag.py 
 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
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 = "This test 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:
        """
        Run VerifyMlagConfigSanity validation
        """

        command_output = self.instance_commands[0].json_output

        if (mlag_status := get_value(command_output, "mlagActive")) is None:
            self.result.is_error("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}")

test 

test() -> None

Run VerifyMlagConfigSanity validation

Source code in anta/tests/mlag.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@AntaTest.anta_test
def test(self) -> None:
    """
    Run VerifyMlagConfigSanity validation
    """

    command_output = self.instance_commands[0].json_output

    if (mlag_status := get_value(command_output, "mlagActive")) is None:
        self.result.is_error("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}")

VerifyMlagDualPrimary

Bases: 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 the dual-primary parameters are NOT provided or if MLAG is ‘disabled’.
Source code in anta/tests/mlag.py 
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
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 the dual-primary parameters are NOT provided or if MLAG is 'disabled'.
    """

    name = "VerifyMlagDualPrimary"
    description = "This test verifies the dual-primary detection and its parameters of the MLAG configuration."
    categories = ["mlag"]
    commands = [AntaCommand(command="show mlag detail", ofmt="json")]

    @AntaTest.anta_test
    def test(
        self, detection_delay: Optional[int] = None, errdisabled: bool = False, recovery_delay: Optional[int] = None, recovery_delay_non_mlag: Optional[int] = None
    ) -> None:
        """
        Run VerifyMlagDualPrimary validation

        Args:
            detection_delay: Delay detection for <N> seconds.
            errdisabled: Errdisabled all interfaces when dual-primary is detected. Defaults to False.
            recovery_delay: Delay (seconds) after dual-primary detection resolves until non peer-link ports that are part of an MLAG are enabled.
            recovery_delay_non_mlag: Delay (seconds) after dual-primary detection resolves until ports that are not part of an MLAG are enabled.
        """

        if detection_delay is None or errdisabled is None or recovery_delay is None or recovery_delay_non_mlag is None:
            self.result.is_skipped(
                f"{self.__class__.name} did not run because detection_delay, errdisabled, recovery_delay or recovery_delay_non_mlag were not supplied"
            )
            return

        errdisabled_action = "errdisableAllInterfaces" if 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"] == detection_delay
            and verified_output["detail.dualPrimaryAction"] == errdisabled_action
            and verified_output["dualPrimaryMlagRecoveryDelay"] == recovery_delay
            and verified_output["dualPrimaryNonMlagRecoveryDelay"] == recovery_delay_non_mlag
        ):
            self.result.is_success()

        else:
            self.result.is_failure(f"The dual-primary parameters are not configured properly: {verified_output}")

test 

test(
    detection_delay: Optional[int] = None,
    errdisabled: bool = False,
    recovery_delay: Optional[int] = None,
    recovery_delay_non_mlag: Optional[int] = None,
) -> None

Run VerifyMlagDualPrimary validation

Parameters:

Name Type Description Default
detection_delay Optional[int]

Delay detection for seconds.

 None
errdisabled bool

Errdisabled all interfaces when dual-primary is detected. Defaults to False.

 False
recovery_delay Optional[int]

Delay (seconds) after dual-primary detection resolves until non peer-link ports that are part of an MLAG are enabled.

 None
recovery_delay_non_mlag Optional[int]

Delay (seconds) after dual-primary detection resolves until ports that are not part of an MLAG are enabled.

 None
Source code in anta/tests/mlag.py 
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
@AntaTest.anta_test
def test(
    self, detection_delay: Optional[int] = None, errdisabled: bool = False, recovery_delay: Optional[int] = None, recovery_delay_non_mlag: Optional[int] = None
) -> None:
    """
    Run VerifyMlagDualPrimary validation

    Args:
        detection_delay: Delay detection for <N> seconds.
        errdisabled: Errdisabled all interfaces when dual-primary is detected. Defaults to False.
        recovery_delay: Delay (seconds) after dual-primary detection resolves until non peer-link ports that are part of an MLAG are enabled.
        recovery_delay_non_mlag: Delay (seconds) after dual-primary detection resolves until ports that are not part of an MLAG are enabled.
    """

    if detection_delay is None or errdisabled is None or recovery_delay is None or recovery_delay_non_mlag is None:
        self.result.is_skipped(
            f"{self.__class__.name} did not run because detection_delay, errdisabled, recovery_delay or recovery_delay_non_mlag were not supplied"
        )
        return

    errdisabled_action = "errdisableAllInterfaces" if 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"] == detection_delay
        and verified_output["detail.dualPrimaryAction"] == errdisabled_action
        and verified_output["dualPrimaryMlagRecoveryDelay"] == recovery_delay
        and verified_output["dualPrimaryNonMlagRecoveryDelay"] == recovery_delay_non_mlag
    ):
        self.result.is_success()

    else:
        self.result.is_failure(f"The dual-primary parameters are not configured properly: {verified_output}")

VerifyMlagInterfaces

Bases: 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’.
Source code in anta/tests/mlag.py 
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
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 = "This test 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:
        """
        Run VerifyMlagInterfaces validation
        """

        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']}")

test 

test() -> None

Run VerifyMlagInterfaces validation

Source code in anta/tests/mlag.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@AntaTest.anta_test
def test(self) -> None:
    """
    Run VerifyMlagInterfaces validation
    """

    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']}")

VerifyMlagReloadDelay

Bases: 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 the reload-delay parameters are NOT provided or if MLAG is ‘disabled’.
Source code in anta/tests/mlag.py 
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
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 the reload-delay parameters are NOT provided or if MLAG is 'disabled'.
    """

    name = "VerifyMlagReloadDelay"
    description = "This test verifies the reload-delay parameters of the MLAG configuration."
    categories = ["mlag"]
    commands = [AntaCommand(command="show mlag", ofmt="json")]

    @AntaTest.anta_test
    def test(self, reload_delay: Optional[int] = None, reload_delay_non_mlag: Optional[int] = None) -> None:
        """
        Run VerifyMlagReloadDelay validation

        Args:
            reload_delay: Delay (seconds) after reboot until non peer-link ports that are part of an MLAG are enabled.
            reload_delay_non_mlag: Delay (seconds) after reboot until ports that are not part of an MLAG are enabled.
        """

        if not reload_delay or not reload_delay_non_mlag:
            self.result.is_skipped(f"{self.__class__.name} did not run because reload_delay or reload_delay_non_mlag were not supplied")
            return

        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"] == reload_delay and verified_output["reloadDelayNonMlag"] == reload_delay_non_mlag:
            self.result.is_success()

        else:
            self.result.is_failure(f"The reload-delay parameters are not configured properly: {verified_output}")

test 

test(
    reload_delay: Optional[int] = None,
    reload_delay_non_mlag: Optional[int] = None,
) -> None

Run VerifyMlagReloadDelay validation

Parameters:

Name Type Description Default
reload_delay Optional[int]

Delay (seconds) after reboot until non peer-link ports that are part of an MLAG are enabled.

 None
reload_delay_non_mlag Optional[int]

Delay (seconds) after reboot until ports that are not part of an MLAG are enabled.

 None
Source code in anta/tests/mlag.py 
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
@AntaTest.anta_test
def test(self, reload_delay: Optional[int] = None, reload_delay_non_mlag: Optional[int] = None) -> None:
    """
    Run VerifyMlagReloadDelay validation

    Args:
        reload_delay: Delay (seconds) after reboot until non peer-link ports that are part of an MLAG are enabled.
        reload_delay_non_mlag: Delay (seconds) after reboot until ports that are not part of an MLAG are enabled.
    """

    if not reload_delay or not reload_delay_non_mlag:
        self.result.is_skipped(f"{self.__class__.name} did not run because reload_delay or reload_delay_non_mlag were not supplied")
        return

    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"] == reload_delay and verified_output["reloadDelayNonMlag"] == reload_delay_non_mlag:
        self.result.is_success()

    else:
        self.result.is_failure(f"The reload-delay parameters are not configured properly: {verified_output}")

VerifyMlagStatus

Bases: 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’.
Source code in anta/tests/mlag.py 
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
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 = "This test verifies the health status of the MLAG configuration."
    categories = ["mlag"]
    commands = [AntaCommand(command="show mlag", ofmt="json")]

    @AntaTest.anta_test
    def test(self) -> None:
        """
        Run VerifyMlagStatus validation
        """

        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}")

test 

test() -> None

Run VerifyMlagStatus validation

Source code in anta/tests/mlag.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@AntaTest.anta_test
def test(self) -> None:
    """
    Run VerifyMlagStatus validation
    """

    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}")
Last update: July 19, 2023