1#!/usr/bin/env python3 2# 3# Copyright (c) 2016, 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 config 33import thread_cert 34from pktverify.consts import MLE_ADVERTISEMENT, MLE_PARENT_REQUEST, MLE_CHILD_ID_RESPONSE, SOURCE_ADDRESS_TLV, ROUTE64_TLV, LEADER_DATA_TLV 35from pktverify.packet_verifier import PacketVerifier 36 37LEADER = 1 38ROUTER1 = 2 39ROUTER2 = 3 40ROUTER3 = 4 41ROUTER4 = 5 42 43 44class Cert_5_5_4_SplitMergeRouters(thread_cert.TestCase): 45 TOPOLOGY = { 46 LEADER: { 47 'name': 'LEADER', 48 'mode': 'rdn', 49 'allowlist': [ROUTER1, ROUTER2] 50 }, 51 ROUTER1: { 52 'name': 'ROUTER_1', 53 'mode': 'rdn', 54 'allowlist': [LEADER, ROUTER3] 55 }, 56 ROUTER2: { 57 'name': 'ROUTER_2', 58 'mode': 'rdn', 59 'allowlist': [LEADER, ROUTER4] 60 }, 61 ROUTER3: { 62 'name': 'ROUTER_3', 63 'mode': 'rdn', 64 'allowlist': [ROUTER1] 65 }, 66 ROUTER4: { 67 'name': 'ROUTER_4', 68 'mode': 'rdn', 69 'allowlist': [ROUTER2] 70 }, 71 } 72 73 def _setUpLeader(self): 74 self.nodes[LEADER].add_allowlist(self.nodes[ROUTER1].get_addr64()) 75 self.nodes[LEADER].add_allowlist(self.nodes[ROUTER2].get_addr64()) 76 self.nodes[LEADER].enable_allowlist() 77 self.nodes[LEADER].set_router_selection_jitter(1) 78 79 def test(self): 80 self.nodes[LEADER].start() 81 self.simulator.go(config.LEADER_STARTUP_DELAY) 82 self.assertEqual(self.nodes[LEADER].get_state(), 'leader') 83 84 self.nodes[ROUTER1].start() 85 self.simulator.go(config.ROUTER_STARTUP_DELAY) 86 self.assertEqual(self.nodes[ROUTER1].get_state(), 'router') 87 88 self.nodes[ROUTER2].start() 89 self.simulator.go(config.ROUTER_STARTUP_DELAY) 90 self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') 91 92 self.nodes[ROUTER3].start() 93 self.simulator.go(config.ROUTER_STARTUP_DELAY) 94 self.assertEqual(self.nodes[ROUTER3].get_state(), 'router') 95 96 self.nodes[ROUTER4].start() 97 self.simulator.go(config.ROUTER_STARTUP_DELAY) 98 self.assertEqual(self.nodes[ROUTER4].get_state(), 'router') 99 100 self.nodes[LEADER].reset() 101 self._setUpLeader() 102 self.simulator.go(150) 103 104 self.nodes[LEADER].start() 105 self.simulator.go(50 + config.LEADER_RESET_DELAY) 106 107 self.assertEqual(self.nodes[LEADER].get_state(), 'router') 108 109 self.collect_ipaddrs() 110 addrs = self.nodes[ROUTER4].get_addrs() 111 for addr in addrs: 112 if addr[0:4] != 'fe80': 113 self.assertTrue(self.nodes[ROUTER3].ping(addr)) 114 115 def verify(self, pv): 116 pkts = pv.pkts 117 pv.summary.show() 118 119 LEADER = pv.vars['LEADER'] 120 ROUTER_1 = pv.vars['ROUTER_1'] 121 ROUTER_3 = pv.vars['ROUTER_3'] 122 ROUTER_4 = pv.vars['ROUTER_4'] 123 leader_pkts = pkts.filter_wpan_src64(LEADER) 124 router1_pkts = pkts.filter_wpan_src64(ROUTER_1) 125 126 # Step 2: The Leader MUST send properly formatted MLE Advertisements 127 router1_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next() 128 leader_pkts.range(router1_pkts.index).filter_mle_cmd(MLE_ADVERTISEMENT).must_next().must_verify( 129 lambda p: {SOURCE_ADDRESS_TLV, ROUTE64_TLV, LEADER_DATA_TLV} <= set(p.mle.tlv.type)) 130 131 router1_pkts.filter_mle_cmd(MLE_PARENT_REQUEST).must_next() 132 lreset_start = router1_pkts.index 133 134 router1_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next() 135 lreset_stop = router1_pkts.index 136 137 # Step 3: The Leader MUST stop sending MLE advertisements. 138 leader_pkts.range(lreset_start, lreset_stop).filter_mle_cmd(MLE_ADVERTISEMENT).must_not_next() 139 140 # Step 5: Router_4 MUST send an ICMPv6 Echo Reply to Router_3 141 router3_mleid = pv.vars['ROUTER_3_MLEID'] 142 router4_mleid = pv.vars['ROUTER_4_MLEID'] 143 pkts.filter_wpan_src64(ROUTER_3).filter( 144 lambda p: p.ipv6.src == router3_mleid and p.ipv6.dst == router4_mleid).filter_ping_request().must_next() 145 pkts.filter_wpan_src64(ROUTER_4).filter( 146 lambda p: p.ipv6.src == router4_mleid and p.ipv6.dst == router3_mleid).filter_ping_reply().must_next() 147 148 149if __name__ == '__main__': 150 unittest.main() 151