1#!/usr/bin/env python3
2#
3#  Copyright (c) 2022, 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#
29
30import unittest
31
32import thread_cert
33import config
34from pktverify.consts import MLE_CHILD_UPDATE_REQUEST, TIMEOUT_TLV, ADDR_REL_URI
35from pktverify.packet_verifier import PacketVerifier
36
37# Test description:
38#   This test verifies that detaching function can send correct "goodbye" messages.
39#
40# Topology:
41#
42#   CHILD_1 ----- ROUTER_1 ----- LEADER
43#
44#
45
46LEADER = 1
47ROUTER_1 = 2
48CHILD_1 = 3
49
50
51class TestDetach(thread_cert.TestCase):
52    USE_MESSAGE_FACTORY = False
53    SUPPORT_NCP = False
54
55    TOPOLOGY = {
56        LEADER: {
57            'name': 'Leader',
58            'allowlist': [ROUTER_1],
59            'mode': 'rdn',
60        },
61        ROUTER_1: {
62            'name': 'Router_1',
63            'allowlist': [LEADER, CHILD_1],
64            'mode': 'rdn',
65        },
66        CHILD_1: {
67            'name': 'Child_1',
68            'is_mtd': True,
69            'allowlist': [ROUTER_1],
70            'mode': '-',
71            'timeout': 10,
72        },
73    }
74
75    def test(self):
76        leader = self.nodes[LEADER]
77        router1 = self.nodes[ROUTER_1]
78        child1 = self.nodes[CHILD_1]
79
80        leader.start()
81        self.simulator.go(config.LEADER_STARTUP_DELAY)
82        self.assertEqual(leader.get_state(), 'leader')
83
84        router1.start()
85        self.simulator.go(config.ROUTER_STARTUP_DELAY)
86        self.assertEqual(router1.get_state(), 'router')
87        router1_rloc16 = router1.get_addr16()
88        self.assertTrue(list(filter(lambda x: x[1]['rloc16'] == router1_rloc16, leader.router_table().items())))
89
90        self.collect_rloc16s()
91
92        child1.start()
93        self.simulator.go(7)
94        self.assertEqual(child1.get_state(), 'child')
95        child_table = router1.get_child_table()
96        self.assertEqual(len(child_table), 1)
97        self.assertEqual(child_table[1]['timeout'], 10)
98
99        child1.detach()
100        self.assertEqual(child1.get_state(), 'disabled')
101        self.assertFalse(router1.get_child_table())
102
103        router1.detach()
104        self.assertEqual(router1.get_state(), 'disabled')
105        self.assertFalse(list(filter(lambda x: x[1]['rloc16'] == router1_rloc16, leader.router_table().items())))
106
107        router1.start()
108        self.simulator.go(5)
109        self.assertEqual(router1.get_state(), 'router')
110
111        child1.start()
112        self.simulator.go(7)
113        self.assertEqual(child1.get_state(), 'child')
114        child_table = router1.get_child_table()
115        self.assertEqual(len(child_table), 1)
116        self.assertEqual(child_table[2]['timeout'], 10)
117
118        router1.thread_stop()
119        self.assertEqual(router1.get_state(), 'disabled')
120        child1.detach()
121        self.assertEqual(child1.get_state(), 'disabled')
122
123        router1.start()
124        self.simulator.go(5)
125        self.assertEqual(router1.get_state(), 'router')
126
127        child1.start()
128        self.simulator.go(7)
129        self.assertEqual(child1.get_state(), 'child')
130
131        leader.detach()
132        self.assertEqual(leader.get_state(), 'disabled')
133
134        self.assertTrue(child1.ping(router1.get_mleid(), timeout=20))
135
136        router1.detach()
137        self.assertEqual(router1.get_state(), 'disabled')
138
139        leader.detach()
140        self.assertEqual(leader.get_state(), 'disabled')
141
142        leader.start()
143        self.assertEqual(leader.get_state(), 'detached')
144        leader.detach()
145        self.assertEqual(leader.get_state(), 'disabled')
146
147        leader.start()
148        # leader didn't become leader after the last start(), so it re-syncs in a non-critical manner thus taking ROUTER_RESET_DELAY to recover
149        self.simulator.go(config.ROUTER_RESET_DELAY / 2)
150        self.assertEqual(leader.get_state(), 'detached')
151        self.simulator.go(config.ROUTER_RESET_DELAY / 2)
152        self.assertEqual(leader.get_state(), 'leader')
153        router1.start()
154        self.simulator.go(config.ROUTER_RESET_DELAY)
155        self.assertEqual(router1.get_state(), 'router')
156
157        leader.thread_stop()
158        router1.detach(is_async=True)
159        self.assertEqual(router1.get_state(), 'router')
160        router1.thread_stop()
161        self.assertEqual(router1.get_state(), 'disabled')
162        router1.detach()
163        self.assertEqual(router1.get_state(), 'disabled')
164
165    def verify(self, pv: PacketVerifier):
166        pkts = pv.pkts
167        pv.summary.show()
168
169        leader = pv.vars['Leader']
170        router1 = pv.vars['Router_1']
171        child1 = pv.vars['Child_1']
172        leader_rloc16 = pv.vars['Leader_RLOC16']
173
174        pkts.filter_wpan_src64(child1).filter_mle_cmd(MLE_CHILD_UPDATE_REQUEST).filter_wpan_dst64(
175            router1).must_next().must_verify(lambda p: TIMEOUT_TLV in set(p.mle.tlv.type) and p.mle.tlv.timeout == 0)
176        pkts.filter_wpan_src64(router1).filter_coap_request(ADDR_REL_URI).filter_wpan_dst16(leader_rloc16).must_next()
177        pkts.filter_wpan_src64(child1).filter_mle_cmd(MLE_CHILD_UPDATE_REQUEST).filter_wpan_dst64(
178            router1).must_next().must_verify(lambda p: TIMEOUT_TLV in set(p.mle.tlv.type) and p.mle.tlv.timeout == 0)
179        pkts.filter_wpan_src64(leader).filter_coap_request(ADDR_REL_URI).must_not_next()
180        pkts.filter_wpan_src64(router1).filter_coap_request(ADDR_REL_URI).filter_wpan_dst16(leader_rloc16).must_next()
181
182
183if __name__ == '__main__':
184    unittest.main()
185