From 4dc9f8edf0e3e5fcba2e559d51c824bd99d730a4 Mon Sep 17 00:00:00 2001 From: hjian2017 <30794429+hjian2017@users.noreply.github.com> Date: Fri, 10 Nov 2017 10:47:54 -0600 Subject: [PATCH] [tests] add traffic for Cert_5_3_06_RouterIdMask.py (#2295) --- .../thread-cert/Cert_5_3_06_RouterIdMask.py | 86 ++++++++++++++++--- tests/scripts/thread-cert/command.py | 27 ++++++ tests/scripts/thread-cert/config.py | 5 ++ tests/scripts/thread-cert/node.py | 3 + tests/scripts/thread-cert/node_cli.py | 4 + 5 files changed, 115 insertions(+), 10 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_5_3_06_RouterIdMask.py b/tests/scripts/thread-cert/Cert_5_3_06_RouterIdMask.py index fcf3c35c0..ec511b909 100755 --- a/tests/scripts/thread-cert/Cert_5_3_06_RouterIdMask.py +++ b/tests/scripts/thread-cert/Cert_5_3_06_RouterIdMask.py @@ -31,8 +31,11 @@ import time import unittest import node +import mle +import config +import command -LEADER = 1 +DUT_LEADER = 1 ROUTER1 = 2 ROUTER2 = 3 @@ -42,14 +45,14 @@ class Cert_5_3_6_RouterIdMask(unittest.TestCase): for i in range(1,4): self.nodes[i] = node.Node(i) - self.nodes[LEADER].set_panid(0xface) - self.nodes[LEADER].set_mode('rsdn') - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) - self.nodes[LEADER].enable_whitelist() + self.nodes[DUT_LEADER].set_panid(0xface) + self.nodes[DUT_LEADER].set_mode('rsdn') + self.nodes[DUT_LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[DUT_LEADER].enable_whitelist() self.nodes[ROUTER1].set_panid(0xface) self.nodes[ROUTER1].set_mode('rsdn') - self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_LEADER].get_addr64()) self.nodes[ROUTER1].add_whitelist(self.nodes[ROUTER2].get_addr64()) self.nodes[ROUTER1].enable_whitelist() self.nodes[ROUTER1].set_router_selection_jitter(1) @@ -58,20 +61,27 @@ class Cert_5_3_6_RouterIdMask(unittest.TestCase): self.nodes[ROUTER2].set_mode('rsdn') self._setUpRouter2() + self.sniffer = config.create_default_thread_sniffer() + self.sniffer.start() + def _setUpRouter2(self): self.nodes[ROUTER2].add_whitelist(self.nodes[ROUTER1].get_addr64()) self.nodes[ROUTER2].enable_whitelist() self.nodes[ROUTER2].set_router_selection_jitter(1) def tearDown(self): + self.sniffer.stop() + del self.sniffer + for node in list(self.nodes.values()): node.stop() del self.nodes def test(self): - self.nodes[LEADER].start() - self.nodes[LEADER].set_state('leader') - self.assertEqual(self.nodes[LEADER].get_state(), 'leader') + # 1 + self.nodes[DUT_LEADER].start() + self.nodes[DUT_LEADER].set_state('leader') + self.assertEqual(self.nodes[DUT_LEADER].get_state(), 'leader') self.nodes[ROUTER1].start() time.sleep(5) @@ -80,15 +90,71 @@ class Cert_5_3_6_RouterIdMask(unittest.TestCase): self.nodes[ROUTER2].start() time.sleep(5) self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') + router2_id = self.nodes[ROUTER2].get_router_id() + # Wait DUT_LEADER to establish routing to ROUTER2 via ROUTER1's MLE advertisement. + time.sleep(config.MAX_ADVERTISEMENT_INTERVAL) + + # 2 self.nodes[ROUTER2].reset() self._setUpRouter2() - time.sleep(300) + # 3 & 4 + # Flush the message queue to avoid possible impact on follow-up verification. + dut_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + + # Verify the cost from DUT_LEADER to ROUTER2 goes to infinity in 12 mins. + routing_cost = 1 + for i in range(0, 24): + time.sleep(30) + print("%ss" %((i + 1) * 30)) + + leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT, False) + if msg == None: + continue + + self.assertTrue(command.check_id_set(msg, router2_id)) + + routing_cost = command.get_routing_cost(msg, router2_id) + if routing_cost == 0: + break + self.assertTrue(routing_cost == 0) + + time.sleep(config.INFINITE_COST_TIMEOUT + config.MAX_ADVERTISEMENT_INTERVAL) + leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT) + self.assertFalse(command.check_id_set(msg, router2_id)) + + # 5 + # Flush the message queue to avoid possible impact on follow-up verification. + dut_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) self.nodes[ROUTER2].start() time.sleep(5) self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') + time.sleep(config.MAX_ADVERTISEMENT_INTERVAL) + leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT) + + # 6 + self.nodes[ROUTER1].stop() + self.nodes[ROUTER2].stop() + + router1_id = self.nodes[ROUTER1].get_router_id() + router2_id = self.nodes[ROUTER2].get_router_id() + + time.sleep(config.MAX_NEIGHBOR_AGE + config.MAX_ADVERTISEMENT_INTERVAL) + leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT) + self.assertEqual(command.get_routing_cost(msg, router1_id), 0) + + time.sleep(config.INFINITE_COST_TIMEOUT + config.MAX_ADVERTISEMENT_INTERVAL) + leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER) + msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT) + self.assertFalse(command.check_id_set(msg, router1_id)) + self.assertFalse(command.check_id_set(msg, router2_id)) + if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/command.py b/tests/scripts/thread-cert/command.py index b4d099b5e..8f994560e 100644 --- a/tests/scripts/thread-cert/command.py +++ b/tests/scripts/thread-cert/command.py @@ -79,3 +79,30 @@ def check_icmp_path(sniffer, path, nodes, icmp_type = ipv6.ICMP_ECHO_REQUEST): return True return False + +def check_id_set(command_msg, router_id): + """Check the command_msg's Route64 tlv to verify router_id is an active router. + """ + tlv = command_msg.assertMleMessageContainsTlv(mle.Route64) + return ((tlv.router_id_mask >> (63 - router_id)) & 1) + +def get_routing_cost(command_msg, router_id): + """Check the command_msg's Route64 tlv to get the routing cost to router. + """ + tlv = command_msg.assertMleMessageContainsTlv(mle.Route64) + + # Get router's mask pos + # Turn the number into binary string. Need to consider the preceding 0 omitted during conversion. + router_id_mask_str = bin(tlv.router_id_mask).replace('0b','') + prefix_len = 64 - len(router_id_mask_str) + routing_entry_pos = 0 + + for i in range(0, router_id - prefix_len): + if router_id_mask_str[i] == '1': + routing_entry_pos += 1 + + assert router_id_mask_str[router_id - prefix_len] == '1', "Error: The router isn't in the topology. \n" \ + + "route64 tlv is: %s. \nrouter_id is: %s. \nrouting_entry_pos is: %s. \nrouter_id_mask_str is: %s." \ + %(tlv, router_id, routing_entry_pos, router_id_mask_str) + + return tlv.link_quality_and_route_data[routing_entry_pos].route diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index 7a33068c9..aefbcd204 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -59,6 +59,11 @@ RSSI = {'LINK_QULITY_0': -100, 'LINK_QULITY_1': -95, 'LINK_QULITY_2': -85, 'LINK SNIFFER_ID = int(os.getenv('SNIFFER_ID', 34)) PANID = 0xface +MAX_NEIGHBOR_AGE = 100 +INFINITE_COST_TIMEOUT = 90 + +MAX_ADVERTISEMENT_INTERVAL = 32 + 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/node.py b/tests/scripts/thread-cert/node.py index d86fd204f..4e1be862c 100755 --- a/tests/scripts/thread-cert/node.py +++ b/tests/scripts/thread-cert/node.py @@ -104,6 +104,9 @@ class Node: def get_addr16(self): return self.interface.get_addr16() + def get_router_id(self): + return self.interface.get_router_id() + def get_addr64(self): return self.interface.get_addr64() diff --git a/tests/scripts/thread-cert/node_cli.py b/tests/scripts/thread-cert/node_cli.py index 69db6f394..da135fb2b 100644 --- a/tests/scripts/thread-cert/node_cli.py +++ b/tests/scripts/thread-cert/node_cli.py @@ -192,6 +192,10 @@ class otCli: self.pexpect.expect('Done') return addr16 + def get_router_id(self): + rloc16 = self.get_addr16() + return (rloc16 >> 10) + def get_addr64(self): self.send_command('extaddr') i = self.pexpect.expect('([0-9a-fA-F]{16})')