xref: /aosp_15_r20/external/openthread/tests/toranj/cli/test-032-leader-take-over.py (revision cfb92d1480a9e65faed56933e9c12405f45898b4)
1#!/usr/bin/env python3
2#
3#  Copyright (c) 2023, The OpenThread Authors.
4#  All rights reserved.
5#
6#  Redistribution and use in source and binary forms, with or without
7#  modification, are permitted provided that the following conditions are met:
8#  1. Redistributions of source code must retain the above copyright
9#     notice, this list of conditions and the following disclaimer.
10#  2. Redistributions in binary form must reproduce the above copyright
11#     notice, this list of conditions and the following disclaimer in the
12#     documentation and/or other materials provided with the distribution.
13#  3. Neither the name of the copyright holder nor the
14#     names of its contributors may be used to endorse or promote products
15#     derived from this software without specific prior written permission.
16#
17#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
18#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
21#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
22#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
23#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
24#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
25#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
26#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27#  POSSIBILITY OF SUCH DAMAGE.
28
29from cli import verify
30from cli import verify_within
31import cli
32import time
33
34# -----------------------------------------------------------------------------------------------------------------------
35# Test description:
36
37# This test covers behavior of leader take over (an already attached device
38# trying to form their own partition and taking over the leader role).
39#
40
41test_name = __file__[:-3] if __file__.endswith('.py') else __file__
42print('-' * 120)
43print('Starting \'{}\''.format(test_name))
44
45# -----------------------------------------------------------------------------------------------------------------------
46# Creating `cli.Nodes` instances
47
48speedup = 25
49cli.Node.set_time_speedup_factor(speedup)
50
51node1 = cli.Node()
52node2 = cli.Node()
53node3 = cli.Node()
54child2 = cli.Node()
55
56# -----------------------------------------------------------------------------------------------------------------------
57# Form topology
58
59node1.form('lto')
60node2.join(node1)
61node3.join(node1)
62
63child2.allowlist_node(node2)
64child2.join(node2, cli.JOIN_TYPE_REED)
65
66verify(node1.get_state() == 'leader')
67verify(node2.get_state() == 'router')
68verify(node3.get_state() == 'router')
69verify(child2.get_state() == 'child')
70
71# -----------------------------------------------------------------------------------------------------------------------
72# Test Implementation
73
74node1.set_router_selection_jitter(1)
75
76n1_weight = int(node1.get_leader_weight())
77
78node2.set_leader_weight(n1_weight)
79
80#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
81# Make sure we get `NonCapable` if local leader weight same as current leader's weight
82
83error = None
84try:
85    node2.cli('state leader')
86except cli.CliError as e:
87    error = e
88
89verify(error.message == 'NotCapable')
90
91#- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
92# Update local leader weight and try to take over the leader role on `node2`.
93
94node2.set_leader_weight(n1_weight + 1)
95
96old_partition_id = int(node2.get_partition_id())
97
98node2.cli('state leader')
99
100new_partition_id = int(node2.get_partition_id())
101verify(new_partition_id != old_partition_id)
102
103
104def check_leader_switch():
105    for node in [node1, node2, node3, child2]:
106        verify(int(node.get_partition_id()) == new_partition_id)
107    verify(node1.get_state() == 'router')
108    verify(node2.get_state() == 'leader')
109    verify(node3.get_state() == 'router')
110    verify(child2.get_state() == 'child')
111
112
113verify_within(check_leader_switch, 30)
114
115# -----------------------------------------------------------------------------------------------------------------------
116# Test finished
117
118cli.Node.finalize_all_nodes()
119
120print('\'{}\' passed.'.format(test_name))
121