[tests] add traffic for Cert_5_3_06_RouterIdMask.py (#2295)

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[DUT_LEADER].set_panid(0xface)
-        self.nodes[LEADER].set_mode('rsdn')
+        self.nodes[DUT_LEADER].set_mode('rsdn')
-        self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64())
+        self.nodes[DUT_LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64())
-        self.nodes[LEADER].enable_whitelist()
+        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()
+        # 1
-        self.nodes[LEADER].set_state('leader')
+        self.nodes[DUT_LEADER].start()
-        self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
+        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()

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

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(

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()
 

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})')