1#!/usr/bin/env python3
2#
3#  Copyright (c) 2021, 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 os
31import unittest
32import sys
33
34import config
35import thread_cert
36
37# Test description:
38#   This test verifies History Tracker behavior.
39#
40# Topology:
41#
42#     LEADER
43#       |
44#       |
45#     CHILD
46#
47
48LEADER = 1
49CHILD = 2
50
51SHORT_WAIT = 5
52ONE_DAY = 24 * 60 * 60
53MAX_AGE_IN_DAYS = 49
54
55
56class TestHistoryTracker(thread_cert.TestCase):
57    USE_MESSAGE_FACTORY = False
58    SUPPORT_NCP = False
59
60    TOPOLOGY = {
61        LEADER: {
62            'name': 'Leader',
63            'mode': 'rdn',
64        },
65        CHILD: {
66            'name': 'Child',
67            'mode': 'n',
68        },
69    }
70
71    def test(self):
72        leader = self.nodes[LEADER]
73        child = self.nodes[CHILD]
74
75        # Start the leader and verify that 'netinfo' history
76        # is updated correctly.
77
78        leader.start()
79        self.simulator.go(SHORT_WAIT * 2)
80        self.assertEqual(leader.get_state(), 'leader')
81
82        netinfo = leader.history_netinfo()
83        self.assertEqual(len(netinfo), 2)
84        self.assertEqual(netinfo[0]['role'], 'leader')
85        self.assertEqual(netinfo[0]['mode'], 'rdn')
86        self.assertEqual(int(netinfo[0]['rloc16'], 16), leader.get_addr16())
87        self.assertEqual(netinfo[0]['partition-id'], leader.get_partition_id())
88        self.assertEqual(netinfo[1]['role'], 'detached')
89
90        # Stop the leader
91
92        leader.thread_stop()
93        leader.interface_down()
94        self.simulator.go(SHORT_WAIT)
95        netinfo = leader.history_netinfo(2)
96        self.assertEqual(len(netinfo), 2)
97        self.assertEqual(netinfo[0]['role'], 'disabled')
98        self.assertEqual(netinfo[1]['role'], 'leader')
99
100        # Wait for one day, two days, then up to max age and verify that
101        # `netinfo` entry age is updated correctly.
102        #
103        # Since we want to wait for long duration (49 days), to speed up
104        # the simulation time, we disable leader to avoid the need to
105        # to simulate all the message/events (e.g. MLE adv) while thread
106        # is operational.
107
108        self.simulator.go(ONE_DAY)
109        netinfo = leader.history_netinfo(1)
110        self.assertTrue(netinfo[0]['age'].startswith('1 day'))
111
112        self.simulator.go(ONE_DAY)
113        netinfo = leader.history_netinfo(1)
114        self.assertTrue(netinfo[0]['age'].startswith('2 days'))
115
116        self.simulator.go((MAX_AGE_IN_DAYS - 3) * ONE_DAY)
117        netinfo = leader.history_netinfo(1)
118        self.assertTrue(netinfo[0]['age'].startswith('{} days'.format(MAX_AGE_IN_DAYS - 1)))
119
120        self.simulator.go(ONE_DAY)
121        netinfo = leader.history_netinfo(1)
122        self.assertTrue(netinfo[0]['age'].startswith('more than {} days'.format(MAX_AGE_IN_DAYS)))
123
124        self.simulator.go(2 * ONE_DAY)
125        netinfo = leader.history_netinfo(1)
126        self.assertTrue(netinfo[0]['age'].startswith('more than {} days'.format(MAX_AGE_IN_DAYS)))
127
128        # Start leader and child
129
130        leader.start()
131        self.simulator.go(config.LEADER_RESET_DELAY)
132        self.assertEqual(leader.get_state(), 'leader')
133
134        child.start()
135        self.simulator.go(SHORT_WAIT)
136        self.assertEqual(child.get_state(), 'child')
137
138        child_rloc16 = child.get_addr16()
139        leader_rloc16 = leader.get_addr16()
140
141        # Verify the `netinfo` history on child
142
143        netinfo = child.history_netinfo(2)
144        self.assertEqual(len(netinfo), 2)
145        self.assertEqual(netinfo[0]['role'], 'child')
146        self.assertEqual(netinfo[0]['mode'], 'n')
147        self.assertEqual(int(netinfo[0]['rloc16'], 16), child_rloc16)
148        self.assertEqual(netinfo[0]['partition-id'], leader.get_partition_id())
149        self.assertEqual(netinfo[1]['role'], 'detached')
150
151        # Change the child mode and verify that `netinfo` history
152        # records this change.
153
154        child.set_mode('rn')
155        self.simulator.go(SHORT_WAIT)
156        netinfo = child.history_netinfo(1)
157        self.assertEqual(len(netinfo), 1)
158        self.assertEqual(netinfo[0]['mode'], 'rn')
159
160        # Ping from leader to child and check the RX and TX history
161        # on child and leader.
162
163        child_mleid = child.get_mleid()
164        leader_mleid = leader.get_mleid()
165
166        ping_sizes = [10, 100, 1000]
167        num_msgs = len(ping_sizes)
168
169        for size in ping_sizes:
170            leader.ping(child_mleid, size=size)
171
172        leader_tx = leader.history_tx(num_msgs)
173        leader_rx = leader.history_rx(num_msgs)
174        child_tx = child.history_tx(num_msgs)
175        child_rx = child.history_rx(num_msgs)
176
177        for index in range(num_msgs):
178            self.assertEqual(leader_tx[index]['type'], 'ICMP6(EchoReqst)')
179            self.assertEqual(leader_tx[index]['sec'], 'yes')
180            self.assertEqual(leader_tx[index]['prio'], 'norm')
181            self.assertEqual(leader_tx[index]['tx-success'], 'yes')
182            self.assertEqual(leader_tx[index]['radio'], '15.4')
183            self.assertEqual(int(leader_tx[index]['to'], 16), child_rloc16)
184            self.assertEqual(leader_tx[index]['src'][1:-3], leader_mleid)
185            self.assertEqual(leader_tx[index]['dst'][1:-3], child_mleid)
186
187            self.assertEqual(child_rx[index]['type'], 'ICMP6(EchoReqst)')
188            self.assertEqual(child_rx[index]['sec'], 'yes')
189            self.assertEqual(child_rx[index]['prio'], 'norm')
190            self.assertEqual(child_rx[index]['radio'], '15.4')
191            self.assertEqual(int(child_rx[index]['from'], 16), leader_rloc16)
192            self.assertEqual(child_rx[index]['src'][1:-3], leader_mleid)
193            self.assertEqual(child_rx[index]['dst'][1:-3], child_mleid)
194
195            self.assertEqual(leader_rx[index]['type'], 'ICMP6(EchoReply)')
196            self.assertEqual(child_tx[index]['type'], 'ICMP6(EchoReply)')
197
198            self.assertEqual(leader_tx[index]['len'], child_rx[index]['len'])
199            self.assertEqual(leader_rx[index]['len'], child_tx[index]['len'])
200
201
202if __name__ == '__main__':
203    #  FIXME: We skip the test under distcheck build (the simulation
204    #  under this build for some reason cannot seem to handle longer
205    #  wait times - days up to 50 days in this test). We return error
206    #  code 77 which indicates that this test case was skipped (in
207    #  automake).
208
209    if os.getenv('DISTCHECK_BUILD') == '1':
210        sys.exit(77)
211
212    unittest.main()
213