espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-28 22:57:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] enhance Cert_7_1_03_BorderRouterAsLeader (#3399)

Browse Source
This commit is contained in:
wgtdkp
2018-12-29 11:25:39 -08:00
committed by Jonathan Hui
parent d34f279e5a
commit 3f6d50a728
4 changed files with 133 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_7_1_03_BorderRouterAsLeader.py
+64 -10
View File
@@ -27,18 +27,22 @@
# POSSIBILITY OF SUCH DAMAGE.
#
from binascii import hexlify
import time
import unittest
import config
import command
from command import CheckType
import mle
import node
LEADER = 1
ROUTER = 2
SED1 = 3
ED1 = 4
MED1 = 4
MTDS = [SED1, ED1]
MTDS = [SED1, MED1]
class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
def setUp(self):
@@ -52,7 +56,7 @@ class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
self.nodes[LEADER].set_mode('rsdn')
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[SED1].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ED1].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[MED1].get_addr64())
self.nodes[LEADER].enable_whitelist()
self.nodes[ROUTER].set_panid(0xface)
@@ -67,10 +71,10 @@ class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
self.nodes[SED1].enable_whitelist()
self.nodes[SED1].set_timeout(config.DEFAULT_CHILD_TIMEOUT)
self.nodes[ED1].set_panid(0xface)
self.nodes[ED1].set_mode('rsn')
self.nodes[ED1].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[ED1].enable_whitelist()
self.nodes[MED1].set_panid(0xface)
self.nodes[MED1].set_mode('rsn')
self.nodes[MED1].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[MED1].enable_whitelist()
def tearDown(self):
for node in list(self.nodes.values()):
@@ -79,6 +83,7 @@ class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
self.simulator.stop()
def test(self):
# 1 - All
self.nodes[LEADER].start()
self.simulator.go(5)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
@@ -91,15 +96,25 @@ class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
self.simulator.go(5)
self.assertEqual(self.nodes[SED1].get_state(), 'child')
self.nodes[ED1].start()
self.nodes[MED1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ED1].get_state(), 'child')
self.assertEqual(self.nodes[MED1].get_state(), 'child')
# 2 - N/A
# Clear collected messages
self.simulator.get_messages_sent_by(LEADER)
self.simulator.get_messages_sent_by(MED1)
self.simulator.get_messages_sent_by(SED1)
self.nodes[LEADER].add_prefix('2001:2:0:1::/64', 'paros')
self.nodes[LEADER].add_prefix('2001:2:0:2::/64', 'paro')
self.nodes[LEADER].register_netdata()
self.simulator.go(5)
leader_messages = self.simulator.get_messages_sent_by(LEADER)
med1_messages = self.simulator.get_messages_sent_by(MED1)
sed1_messages = self.simulator.get_messages_sent_by(SED1)
addrs = self.nodes[SED1].get_addrs()
self.assertTrue(any('2001:2:0:1' in addr[0:10] for addr in addrs))
self.assertFalse(any('2001:2:0:2' in addr[0:10] for addr in addrs))
@@ -107,12 +122,51 @@ class Cert_7_1_3_BorderRouterAsLeader(unittest.TestCase):
if addr[0:10] == '2001:2:0:1' or addr[0:10] == '2001:2:0:2':
self.assertTrue(self.nodes[LEADER].ping(addr))
addrs = self.nodes[ED1].get_addrs()
addrs = self.nodes[MED1].get_addrs()
self.assertTrue(any('2001:2:0:1' in addr[0:10] for addr in addrs))
self.assertTrue(any('2001:2:0:2' in addr[0:10] for addr in addrs))
for addr in addrs:
if addr[0:10] == '2001:2:0:1' or addr[0:10] == '2001:2:0:2':
self.assertTrue(self.nodes[LEADER].ping(addr))
# 3 - Leader
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE)
command.check_data_response(msg, network_data=CheckType.CONTAIN,
prefixes=[('2001:2:0:1::/64', 'paros'), ('2001:2:0:2::/64', 'paro')])
# 4 - N/A
# Get addresses registered by MED1
msg = med1_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
med1_addresses = msg.get_mle_message_tlv(mle.AddressRegistration).addresses
# 5 - Leader
# Make a copy of leader's messages to ensure that we don't miss messages to SED1
leader_messages_copy = leader_messages.clone()
msg = leader_messages_copy.next_mle_message(mle.CommandType.CHILD_UPDATE_RESPONSE, sent_to_node=self.nodes[MED1])
command.check_child_update_response_from_parent(msg, address_registration=CheckType.CONTAIN)
leader_addresses = msg.get_mle_message_tlv(mle.AddressRegistration).addresses
self.assertTrue(all(addr in leader_addresses for addr in med1_addresses))
# 6A & 6B - Leader
if config.LEADER_NOTIFY_SED_BY_CHILD_UPDATE_REQUEST:
msg = leader_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST, sent_to_node=self.nodes[SED1])
command.check_child_update_request_from_parent(msg,
leader_data=CheckType.CONTAIN, network_data=CheckType.CONTAIN, active_timestamp=CheckType.CONTAIN)
else:
msg = leader_messages.next_mle_message(mle.CommandType.DATA_RESPONSE, sent_to_node=self.nodes[SED1])
command.check_data_response(msg, network_data=CheckType.CONTAIN, active_timestamp=CheckType.CONTAIN)
# 7 - N/A
# Get addresses registered by SED1
msg = sed1_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_REQUEST)
sed1_addresses = msg.get_mle_message_tlv(mle.AddressRegistration).addresses
# 8 - Leader
msg = leader_messages.next_mle_message(mle.CommandType.CHILD_UPDATE_RESPONSE, sent_to_node=self.nodes[SED1])
command.check_child_update_response_from_parent(msg, address_registration=CheckType.CONTAIN)
leader_addresses = msg.get_mle_message_tlv(mle.AddressRegistration).addresses
self.assertTrue(all(addr in leader_addresses for addr in sed1_addresses))
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/command.py
+61
View File
@@ -30,6 +30,7 @@
import sys
import ipv6
from network_data import Prefix, BorderRouter, LowpanId
import network_layer
import config
import mle
@@ -345,8 +346,68 @@ def check_router_id_cached(node, router_id, cached = True):
assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches) is False
def contains_tlv(sub_tlvs, tlv_type):
"""Verify if a specific type of tlv is included in a sub-tlv list.
"""
return any(isinstance(sub_tlv, tlv_type) for sub_tlv in sub_tlvs)
def check_secure_mle_key_id_mode(command_msg, key_id_mode):
"""Verify if the mle command message sets the right key id mode.
"""
assert isinstance(command_msg.mle, mle.MleMessageSecured)
assert command_msg.mle.aux_sec_hdr.key_id_mode == key_id_mode
def check_data_response(command_msg, network_data=CheckType.OPTIONAL, prefixes=[],
active_timestamp=CheckType.OPTIONAL):
"""Verify a properly formatted Data Response command message.
"""
check_secure_mle_key_id_mode(command_msg, 0x02)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
if network_data == CheckType.CONTAIN and len(prefixes) > 0:
network_data_tlv = command_msg.get_mle_message_tlv(mle.NetworkData)
prefix_tlvs = [tlv for tlv in network_data_tlv.tlvs if isinstance(tlv, Prefix)]
assert len(prefix_tlvs) >= len(prefixes)
for prefix in prefix_tlvs:
assert contains_tlv(prefix.sub_tlvs, BorderRouter)
assert contains_tlv(prefix.sub_tlvs, LowpanId)
def check_child_update_request_from_parent(command_msg, leader_data=CheckType.OPTIONAL,
network_data=CheckType.OPTIONAL, challenge=CheckType.OPTIONAL,
tlv_request=CheckType.OPTIONAL, active_timestamp=CheckType.OPTIONAL):
"""Verify a properly formatted Child Update Request(from parent) command message.
"""
check_secure_mle_key_id_mode(command_msg, 0x02)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)
check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
check_mle_optional_tlv(command_msg, challenge, mle.Challenge)
check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest)
check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp)
def check_child_update_response_from_parent(command_msg, timeout=CheckType.OPTIONAL,
address_registration=CheckType.OPTIONAL, address16=CheckType.OPTIONAL,
leader_data=CheckType.OPTIONAL, network_data=CheckType.OPTIONAL, response=CheckType.OPTIONAL,
link_layer_frame_counter=CheckType.OPTIONAL, mle_frame_counter=CheckType.OPTIONAL):
"""Verify a properly formatted Child Update Response from parent
"""
check_secure_mle_key_id_mode(command_msg, 0x02)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.Mode)
check_mle_optional_tlv(command_msg, timeout, mle.Timeout)
check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration)
check_mle_optional_tlv(command_msg, address16, mle.Address16)
check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData)
check_mle_optional_tlv(command_msg, network_data, mle.NetworkData)
check_mle_optional_tlv(command_msg, response, mle.Response)
check_mle_optional_tlv(command_msg, link_layer_frame_counter, mle.LinkLayerFrameCounter)
check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter)
def get_sub_tlv(tlvs, tlv_type):
for sub_tlv in tlvs:
if isinstance(sub_tlv, tlv_type):
tests/scripts/thread-cert/config.py
+2
View File
@@ -74,6 +74,8 @@ DEFAULT_CHILD_TIMEOUT = 6
VIRTUAL_TIME = bool(os.getenv('VIRTUAL_TIME', False))
LEADER_NOTIFY_SED_BY_CHILD_UPDATE_REQUEST = True
def create_default_network_data_prefix_sub_tlvs_factories():
return {
network_data.TlvType.HAS_ROUTE: network_data.HasRouteFactory(
tests/scripts/thread-cert/message.py
+6 -1
View File
@@ -292,7 +292,7 @@ class Message(object):
if self.mac_header.dest_address == mac_address:
sent_to_node = True
assert sent_to_node == True
assert sent_to_node
def assertSentToDestinationAddress(self, ipv6_address):
if sys.version_info[0] == 2:
@@ -469,6 +469,11 @@ class MessagesSet(object):
return True
def clone(self):
"""Make a copy of current MessageSet.
"""
return MessagesSet(self.messages[:])
class MessageFactory: