mirror of
https://github.com/espressif/openthread.git
synced 2026-06-05 21:14:49 +00:00
Jonathan Hui
321 lines
13 KiB
Python
Executable File
321 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
#
|
|
# Copyright (c) 2020, The OpenThread Authors.
|
|
# All rights reserved.
|
|
#
|
|
# Redistribution and use in source and binary forms, with or without
|
|
# modification, are permitted provided that the following conditions are met:
|
|
# 1. Redistributions of source code must retain the above copyright
|
|
# notice, this list of conditions and the following disclaimer.
|
|
# 2. Redistributions in binary form must reproduce the above copyright
|
|
# notice, this list of conditions and the following disclaimer in the
|
|
# documentation and/or other materials provided with the distribution.
|
|
# 3. Neither the name of the copyright holder nor the
|
|
# names of its contributors may be used to endorse or promote products
|
|
# derived from this software without specific prior written permission.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
|
|
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
|
|
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
|
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
|
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
|
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
|
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
|
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
# POSSIBILITY OF SUCH DAMAGE.
|
|
#
|
|
|
|
import unittest
|
|
|
|
import command
|
|
import config
|
|
import ipaddress
|
|
import mle
|
|
import thread_cert
|
|
|
|
BBR_1 = 1 # Collapsed with Leader Role
|
|
ROUTER_1_1 = 2
|
|
ROUTER_1_2 = 3
|
|
MED_1_2 = 4
|
|
SED_1_2 = 5
|
|
|
|
WAIT_ATTACH = 5
|
|
WAIT_REDUNDANCE = 3
|
|
ROUTER_SELECTION_JITTER = 1
|
|
BBR_REGISTRATION_JITTER = 5
|
|
SED_POLL_PERIOD = 2000 # 2s
|
|
MED_TIMEOUT = 20 # 20s
|
|
|
|
DUA_IID_MANUAL1 = '4444333322221111'
|
|
DUA_IID_MANUAL2 = '5555444433332222'
|
|
|
|
TEST_PREFIX1 = '2001:0:0:1::/64'
|
|
TEST_PREFIX2 = '2001:0:0:2::/64'
|
|
TEST_PREFIX3 = '2001:0:0:3::/64'
|
|
"""
|
|
Topology
|
|
|
|
|
|
SED_1_2
|
|
|
|
|
|
|
|
ROUTER_1_1 MED_1_2
|
|
| |
|
|
| |
|
|
BBR_1 (LEADER) --- ROUTER_1_2
|
|
|
|
|
|
1) Bring up BBR_1, BBR_1 becomes Leader and Primary Backbone Router, with Domain
|
|
Prefix without `P_slaac`.
|
|
|
|
2) Bring up ROUTER_1_1, no DUA was added due to that `P_slaac` flag is not set.
|
|
|
|
3) Bring up ROUTER_1_2, verify that it has DUA generated.
|
|
|
|
4) Bring up MED_1_2 with DUA_IID_MANUAL1 set in advance, verify
|
|
a) DUA_IID_MANUAL1 is registered in Address Registration TLV via Child Update Request.
|
|
b) Remove DUA_IID_MANUAL1, a new DUA generated via SLAAC would be registered in Address
|
|
Registration TLV via Child Update Request.
|
|
c) Set DUA_IID_MANUAL2 which should override the generated one and be registered in Address
|
|
Registration TLV via Child Update Request.
|
|
d) Remove DUA_IID_MANUAL2, a new DUA generated via SLAAC, the same as in above b) would
|
|
be registered in Address Registration TLV via Child Update Request.
|
|
|
|
5) Change BBR_1 from config.DOMAIN_PREFIX to config.DOMAIN_PRFIX_ALTER. Verify that MED_1_2
|
|
generate a new Interface Identifier different from the one generated in 4d) due to the
|
|
Domain Prefix change.
|
|
|
|
6) Recover config.Domain_Prefix on BBR_1. Verify that MED_1_2 generates and registers the same
|
|
DUA as in step 4b).
|
|
|
|
7) Configure ROUTER_1_1 as Border Router with 3 SLAAC prefixes, verify MED_1_2 would register
|
|
its DUA in Address Registration TLV.
|
|
|
|
8) Bring up SED_1_2, verify it generates one DUA, and registers it to its parent, though the parent
|
|
is a Thread 1.1 device.
|
|
|
|
"""
|
|
|
|
|
|
class TestDomainUnicastAddress(thread_cert.TestCase):
|
|
TOPOLOGY = {
|
|
BBR_1: {
|
|
'version': '1.2',
|
|
'allowlist': [ROUTER_1_1, ROUTER_1_2],
|
|
'is_bbr': True
|
|
},
|
|
ROUTER_1_1: {
|
|
'version': '1.1',
|
|
'allowlist': [BBR_1, SED_1_2]
|
|
},
|
|
ROUTER_1_2: {
|
|
'version': '1.2',
|
|
'allowlist': [BBR_1, MED_1_2]
|
|
},
|
|
MED_1_2: {
|
|
'mode': 'rn',
|
|
'version': '1.2',
|
|
'allowlist': [ROUTER_1_2],
|
|
},
|
|
SED_1_2: {
|
|
'mode': 'n',
|
|
'version': '1.2',
|
|
'allowlist': [ROUTER_1_1],
|
|
},
|
|
}
|
|
"""All nodes are created with default configurations"""
|
|
|
|
def __get_iid(self, address):
|
|
''' Get the interface identifier of an IPv6 address.
|
|
|
|
Args:
|
|
address (string): An IPv6 address;
|
|
'''
|
|
return ''.join(ipaddress.ip_address(address).exploded.split(':')[4:])
|
|
|
|
def __check_dua_registration(self, node, iid, dp_cid):
|
|
''' Check whether or not the specified Domain Unicast Address is registered in Address
|
|
Registration TLV.
|
|
|
|
Args:
|
|
node (int) : The device id
|
|
iid (string): The Interface Identifier
|
|
dp_cid (int): The context id of the domain prefix.
|
|
'''
|
|
|
|
messages = self.simulator.get_messages_sent_by(node)
|
|
msg = messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
|
|
command.check_compressed_address_registration_tlv(msg, dp_cid, iid, cid_present_once=True)
|
|
|
|
def test(self):
|
|
# starting context id
|
|
context_id = 1
|
|
|
|
# 1) Bring up BBR_1, BBR_1 becomes Leader and Primary Backbone Router, with Domain
|
|
# Prefix without `P_slaac`.
|
|
self.nodes[BBR_1].set_router_selection_jitter(ROUTER_SELECTION_JITTER)
|
|
self.nodes[BBR_1].set_bbr_registration_jitter(BBR_REGISTRATION_JITTER)
|
|
self.nodes[BBR_1].set_backbone_router(seqno=1)
|
|
self.nodes[BBR_1].start()
|
|
WAIT_TIME = WAIT_ATTACH + ROUTER_SELECTION_JITTER
|
|
self.simulator.go(WAIT_TIME * 2)
|
|
self.assertEqual(self.nodes[BBR_1].get_state(), 'leader')
|
|
self.nodes[BBR_1].enable_backbone_router()
|
|
WAIT_TIME = BBR_REGISTRATION_JITTER + WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
self.assertEqual(self.nodes[BBR_1].get_backbone_router_state(), 'Primary')
|
|
|
|
self.nodes[BBR_1].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
self.simulator.set_lowpan_context(context_id, config.DOMAIN_PREFIX)
|
|
domain_prefix_cid = context_id
|
|
|
|
# 2) Bring up ROUTER_1_1, no DUA was added due to that `P_slaac` flag is not set.
|
|
self.nodes[ROUTER_1_1].set_router_selection_jitter(ROUTER_SELECTION_JITTER)
|
|
WAIT_TIME = WAIT_ATTACH + ROUTER_SELECTION_JITTER
|
|
self.nodes[ROUTER_1_1].start()
|
|
self.simulator.go(WAIT_TIME)
|
|
self.assertEqual(self.nodes[ROUTER_1_1].get_state(), 'router')
|
|
dua = self.nodes[ROUTER_1_1].get_addr(config.DOMAIN_PREFIX)
|
|
assert not dua, 'Error: Unexpected DUA ({})'.format(dua)
|
|
|
|
# 3) Bring up ROUTER_1_2, verify that it has DUA generated.
|
|
self.nodes[ROUTER_1_2].set_router_selection_jitter(ROUTER_SELECTION_JITTER)
|
|
self.nodes[ROUTER_1_2].start()
|
|
WAIT_TIME = WAIT_ATTACH + ROUTER_SELECTION_JITTER
|
|
self.simulator.go(WAIT_TIME)
|
|
self.assertEqual(self.nodes[ROUTER_1_2].get_state(), 'router')
|
|
dua = self.nodes[ROUTER_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
assert dua, 'Error: Expected DUA not found'
|
|
self.assertTrue(self.nodes[BBR_1].ping(dua))
|
|
|
|
# 4) Bring up MED_1_2 with DUA_IID_MANUAL1 set in advance
|
|
self.nodes[MED_1_2].set_dua_iid(DUA_IID_MANUAL1)
|
|
self.nodes[MED_1_2].set_timeout(MED_TIMEOUT)
|
|
self.nodes[MED_1_2].start()
|
|
WAIT_TIME = WAIT_ATTACH
|
|
self.simulator.go(WAIT_TIME)
|
|
self.assertEqual(self.nodes[MED_1_2].get_state(), 'child')
|
|
|
|
# 4a) DUA_IID_MANUAL1 is registered in Address Registration TLV via Child Update Request.
|
|
self.__check_dua_registration(MED_1_2, DUA_IID_MANUAL1, domain_prefix_cid)
|
|
|
|
# 4b) Remove DUA_IID_MANUAL1, a new DUA generated via SLAAC would be registered in Address
|
|
# Registration TLV via Child Update Request.
|
|
|
|
# Flush relative message queues.
|
|
messages = self.simulator.get_messages_sent_by(MED_1_2)
|
|
|
|
self.nodes[MED_1_2].clear_dua_iid()
|
|
WAIT_TIME = MED_TIMEOUT + WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
med_1_2_dua = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
assert med_1_2_dua, 'Error: Expected DUA not found'
|
|
|
|
med_1_2_dua_iid = self.__get_iid(med_1_2_dua)
|
|
self.__check_dua_registration(MED_1_2, med_1_2_dua_iid, domain_prefix_cid)
|
|
|
|
# 4c) Set DUA_IID_MANUAL2 which should override the generated one and be registered in Address
|
|
# Registration TLV via Child Update Request.
|
|
|
|
# Flush relative message queues.
|
|
messages = self.simulator.get_messages_sent_by(MED_1_2)
|
|
self.nodes[MED_1_2].set_dua_iid(DUA_IID_MANUAL2)
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
dua = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
|
|
self.__check_dua_registration(MED_1_2, DUA_IID_MANUAL2, domain_prefix_cid)
|
|
|
|
# 4d) Remove DUA_IID_MANUAL2, a new DUA generated via SLAAC, the same as in above b) would
|
|
# be registered in Address Registration TLV via Child Update Request.
|
|
|
|
# Flush relative message queues.
|
|
messages = self.simulator.get_messages_sent_by(MED_1_2)
|
|
self.nodes[MED_1_2].clear_dua_iid()
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
dua = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
self.assertEqual(ipaddress.ip_address(dua), ipaddress.ip_address(med_1_2_dua))
|
|
self.assertEqual(ipaddress.ip_address(med_1_2_dua), ipaddress.ip_address(dua))
|
|
|
|
self.__check_dua_registration(MED_1_2, med_1_2_dua_iid, domain_prefix_cid)
|
|
|
|
# 5) Change BBR_1 from config.DOMAIN_PREFIX to config.DOMAIN_PRFIX_ALTER. Verify that MED_1_2
|
|
# generates a new Interface Identifier different from the one generated in 4d) due to the
|
|
# Domain Prefix change.
|
|
context_id += 1
|
|
self.simulator.set_lowpan_context(context_id, config.DOMAIN_PREFIX_ALTER)
|
|
self.nodes[BBR_1].set_domain_prefix(config.DOMAIN_PREFIX_ALTER)
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
med_1_2_dua2 = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX_ALTER)
|
|
med_1_2_dua2_iid = self.__get_iid(med_1_2_dua2)
|
|
self.__check_dua_registration(MED_1_2, med_1_2_dua2_iid, context_id)
|
|
|
|
#6) Recover config.Domain_Prefix on BBR_1. Verify that MED_1_2 generates and registers the same
|
|
# DUA as in step 4b).
|
|
self.nodes[BBR_1].set_domain_prefix(config.DOMAIN_PREFIX)
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
dua = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
assert dua, 'Error: Expected DUA not found'
|
|
self.assertEqual(ipaddress.ip_address(med_1_2_dua), ipaddress.ip_address(dua))
|
|
|
|
self.__check_dua_registration(MED_1_2, med_1_2_dua_iid, domain_prefix_cid)
|
|
|
|
#7) Configure ROUTER_1_1 as Border Router with 3 SLAAC prefixes, verify MED_1_2 would register
|
|
# its DUA in Address Registration TLV.
|
|
|
|
# Flush relative message queues.
|
|
messages = self.simulator.get_messages_sent_by(MED_1_2)
|
|
|
|
context_id += 1
|
|
self.simulator.set_lowpan_context(context_id, TEST_PREFIX1)
|
|
self.nodes[ROUTER_1_1].add_prefix(TEST_PREFIX1)
|
|
|
|
context_id += 1
|
|
self.simulator.set_lowpan_context(context_id, TEST_PREFIX2)
|
|
self.nodes[ROUTER_1_1].add_prefix(TEST_PREFIX2)
|
|
context_id += 1
|
|
self.simulator.set_lowpan_context(context_id, TEST_PREFIX3)
|
|
self.nodes[ROUTER_1_1].add_prefix(TEST_PREFIX3)
|
|
self.nodes[ROUTER_1_1].register_netdata()
|
|
|
|
WAIT_TIME = WAIT_REDUNDANCE
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
WAIT_TIME = MED_TIMEOUT
|
|
self.simulator.go(WAIT_TIME)
|
|
dua = self.nodes[MED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
assert dua, 'Error: Expected DUA not found'
|
|
self.assertEqual(ipaddress.ip_address(med_1_2_dua), ipaddress.ip_address(dua))
|
|
|
|
self.__check_dua_registration(MED_1_2, med_1_2_dua_iid, domain_prefix_cid)
|
|
|
|
#8) Bring up SED_1_2, verify that it generates one DUA, and registers it to its parent, though the parent
|
|
# is a Thread 1.1 device.
|
|
self.nodes[SED_1_2].set_pollperiod(SED_POLL_PERIOD)
|
|
self.nodes[SED_1_2].start()
|
|
WAIT_TIME = WAIT_ATTACH
|
|
self.simulator.go(WAIT_TIME)
|
|
|
|
dua = self.nodes[SED_1_2].get_addr(config.DOMAIN_PREFIX)
|
|
assert dua, 'Error: Expected DUA not found'
|
|
|
|
dua_iid = self.__get_iid(dua)
|
|
self.__check_dua_registration(SED_1_2, dua_iid, domain_prefix_cid)
|
|
|
|
self.assertTrue(self.nodes[BBR_1].ping(dua))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|