[test] add traffic analysis for Cert_5_2_3_LeaderReject2Hops (#2142)

Also add a method last_mle_message to find the newest mle; Add a method assertAssignedRouterQuantity to confirm if Leader contains the Route64 TLV with 32 assigned router IDs.

tests/scripts/thread-cert/Cert_5_2_03_LeaderReject2Hops.py

@@ -31,61 +31,103 @@ import time
 import unittest
 
 import node
+import mle
+import network_layer
+import config
 
-LEADER = 1
-ROUTER = 2
-DUT = 33
+DUT_LEADER = 1
+ROUTER_1 = 2
+ROUTER_31 = 32
+ROUTER_32 = 33
+SNIFFER = 34
 
 class Cert_5_2_3_LeaderReject2Hops(unittest.TestCase):
     def setUp(self):
         self.nodes = {}
 
-        self.nodes[LEADER] = node.Node(LEADER)
-        self.nodes[LEADER].set_panid(0xface)
-        self.nodes[LEADER].set_mode('rsdn')
-        self.nodes[LEADER].enable_whitelist()
-        self.nodes[LEADER].set_router_upgrade_threshold(32)
-        self.nodes[LEADER].set_router_downgrade_threshold(33)
+        self.nodes[DUT_LEADER] = node.Node(DUT_LEADER)
+        self.nodes[DUT_LEADER].set_panid(0xface)
+        self.nodes[DUT_LEADER].set_mode('rsdn')
+        self.nodes[DUT_LEADER].enable_whitelist()
+        self.nodes[DUT_LEADER].set_router_upgrade_threshold(32)
+        self.nodes[DUT_LEADER].set_router_downgrade_threshold(33)
 
-        for i in range(2,33):
+        for i in range(2, 33):
             self.nodes[i] = node.Node(i)
             self.nodes[i].set_panid(0xface)
             self.nodes[i].set_mode('rsdn')
-            self.nodes[i].add_whitelist(self.nodes[LEADER].get_addr64())
-            self.nodes[LEADER].add_whitelist(self.nodes[i].get_addr64())
+            self.nodes[i].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
+            self.nodes[DUT_LEADER].add_whitelist(self.nodes[i].get_addr64())
             self.nodes[i].enable_whitelist()
             self.nodes[i].set_router_upgrade_threshold(33)
             self.nodes[i].set_router_downgrade_threshold(33)
             self.nodes[i].set_router_selection_jitter(1)
 
-        self.nodes[DUT] = node.Node(DUT)
-        self.nodes[DUT].set_panid(0xface)
-        self.nodes[DUT].set_mode('rsdn')
-        self.nodes[DUT].add_whitelist(self.nodes[ROUTER].get_addr64())
-        self.nodes[ROUTER].add_whitelist(self.nodes[DUT].get_addr64())
-        self.nodes[DUT].enable_whitelist()
-        self.nodes[DUT].set_router_upgrade_threshold(33)
-        self.nodes[DUT].set_router_downgrade_threshold(33)
-        self.nodes[DUT].set_router_selection_jitter(1)
+        self.nodes[ROUTER_32] = node.Node(ROUTER_32)
+        self.nodes[ROUTER_32].set_panid(0xface)
+        self.nodes[ROUTER_32].set_mode('rsdn')
+        self.nodes[ROUTER_32].add_whitelist(self.nodes[ROUTER_1].get_addr64())
+        self.nodes[ROUTER_1].add_whitelist(self.nodes[ROUTER_32].get_addr64())
+        self.nodes[ROUTER_32].enable_whitelist()
+        self.nodes[ROUTER_32].set_router_upgrade_threshold(33)
+        self.nodes[ROUTER_32].set_router_downgrade_threshold(33)
+        self.nodes[ROUTER_32].set_router_selection_jitter(1)
+
+        self.sniffer = config.create_default_thread_sniffer(SNIFFER)
+        self.sniffer.start()
 
     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')
 
-        for i in range(2, 33):
+        for i in range(2, 32):
             self.nodes[i].start()
             time.sleep(5)
             self.assertEqual(self.nodes[i].get_state(), 'router')
 
-        self.nodes[DUT].start()
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+
+        # 2
+        self.nodes[ROUTER_31].start()
         time.sleep(5)
-        self.assertEqual(self.nodes[DUT].get_state(), 'child')
+        self.assertEqual(self.nodes[ROUTER_31].get_state(), 'router')
+
+        # 3 - DUT_LEADER
+        # This method flushes the message queue so calling this method again will return only the newly logged messages.
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('2.04')
+        msg.assertCoapMessageContainsTlv(network_layer.Status)
+        msg.assertCoapMessageContainsTlv(network_layer.RouterMask)
+        msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
+
+        status_tlv = msg.get_coap_message_tlv(network_layer.Status)
+        self.assertEqual(network_layer.StatusValues.SUCCESS, status_tlv.status)
+
+        # 4 - DUT_LEADER
+        msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT)
+        msg.assertAssignedRouterQuantity(32)
+
+        # 5 - Router_32
+        self.nodes[ROUTER_32].start()
+        time.sleep(5)
+
+        # 6 - DUT_LEADER
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('2.04')
+        msg.assertCoapMessageContainsTlv(network_layer.Status)
+
+        status_tlv = msg.get_coap_message_tlv(network_layer.Status)
+        self.assertEqual(network_layer.StatusValues.NO_ADDRESS_AVAILABLE, status_tlv.status)
 
 if __name__ == '__main__':
     unittest.main()

tests/scripts/thread-cert/message.py

@@ -156,6 +156,14 @@ class Message(object):
         assert(self.mle.command.type == command_type)
 
     def assertMleMessageContainsTlv(self, tlv_class_type):
+        """To confirm if Mle message contains the TLV type.
+
+        Args:
+            tlv_class_type: tlv's type.
+
+        Returns:
+            mle.Route64: If contains the TLV, return it.
+        """
         if self.type != MessageType.MLE:
             raise ValueError("Invalid message type. Expected MLE message.")
 
@@ -166,6 +174,22 @@ class Message(object):
                 break
 
         assert(contains_tlv == True)
+        return tlv
+
+    def assertAssignedRouterQuantity(self, router_quantity):
+        """Confirm if Leader contains the Route64 TLV with router_quantity assigned Router IDs.
+
+        Args:
+            router_quantity: the quantity of router.
+        """
+        tlv = self.assertMleMessageContainsTlv(mle.Route64)
+        router_id_mask = tlv.router_id_mask
+
+        count = 0
+        for i in range(1, 65):
+            count += (router_id_mask & 1)
+            router_id_mask = (router_id_mask >> 1)
+        assert(count == router_quantity)
 
     def assertMleMessageDoesNotContainTlv(self, tlv_class_type):
         if self.type != MessageType.MLE:
@@ -294,8 +318,37 @@ class MessagesSet(object):
 
         return message
 
+    def last_mle_message(self, command_type, assert_enabled = True):
+        """Get the last Mle Message with specified type from existing capture.
+
+        Args:
+            command_type: the specified mle type.
+            assert_enabled: interrupt or not when get the mle.
+
+        Returns:
+            message.Message: the last Mle Message with specified type.
+        """
+        message = None
+        size = len(self.messages)
+
+        for i in range(size - 1, -1, -1):
+            m = self.messages[i]
+
+            if m.type != MessageType.MLE:
+                continue
+
+            #for command_type in command_types:
+            if m.mle.command.type == command_type:
+                message = m
+                break
+
+        if assert_enabled:
+            assert message is not None, "Could not find MleMessage with type: {}".format(command_type)
+
+        return message
+
     def next_mle_message(self, command_type, assert_enabled=True):
-        message = self.next_mle_message_of_one_of_command_types(command_type,)
+        message = self.next_mle_message_of_one_of_command_types(command_type)
 
         if assert_enabled:
             assert message is not None, "Could not find MleMessage of the type: {}".format(command_type)