From 3f6d50a72849abb52c8aef6208f98949de14cdcc Mon Sep 17 00:00:00 2001 From: wgtdkp Date: Sun, 30 Dec 2018 03:25:39 +0800 Subject: [PATCH] [tests] enhance Cert_7_1_03_BorderRouterAsLeader (#3399) --- .../Cert_7_1_03_BorderRouterAsLeader.py | 74 ++++++++++++++++--- tests/scripts/thread-cert/command.py | 61 +++++++++++++++ tests/scripts/thread-cert/config.py | 2 + tests/scripts/thread-cert/message.py | 7 +- 4 files changed, 133 insertions(+), 11 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_7_1_03_BorderRouterAsLeader.py b/tests/scripts/thread-cert/Cert_7_1_03_BorderRouterAsLeader.py index daf6622ef..b58ba3953 100755 --- a/tests/scripts/thread-cert/Cert_7_1_03_BorderRouterAsLeader.py +++ b/tests/scripts/thread-cert/Cert_7_1_03_BorderRouterAsLeader.py @@ -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() diff --git a/tests/scripts/thread-cert/command.py b/tests/scripts/thread-cert/command.py index 1d2bb9498..4e1c452ba 100644 --- a/tests/scripts/thread-cert/command.py +++ b/tests/scripts/thread-cert/command.py @@ -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): diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index 373c01b51..fecac5bac 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -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( diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 30ca6f152..1c7655497 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -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: