[tests] dd traffic analysis for Cert_5_2_06_RouterDowngrade.py (#2192)

Also add a method in command.py to verify a properly formatted address release command message.

tests/scripts/thread-cert/Cert_5_2_06_RouterDowngrade.py

@@ -31,43 +31,74 @@ import time
 import unittest
 
 import node
+import config
+import command
+import mle
 
 LEADER = 1
-ROUTER1 = 2
+DUT_ROUTER1 = 2
+ROUTER2 = 3
+ROUTER24 = 24
 
 class Cert_5_2_06_RouterDowngrade(unittest.TestCase):
     def setUp(self):
         self.nodes = {}
-        for i in range(1, 26):
+        for i in range(1, 25):
             self.nodes[i] = node.Node(i)
-            self.nodes[i].set_panid(0xface)
+            self.nodes[i].set_panid()
             self.nodes[i].set_mode('rsdn')
             self.nodes[i].set_router_selection_jitter(1)
-            if i != ROUTER1:
+            if i != DUT_ROUTER1:
                 self.nodes[i].set_router_upgrade_threshold(32)
                 self.nodes[i].set_router_downgrade_threshold(32)
 
+        self.sniffer = config.create_default_thread_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):
+        # 1 Ensure topology is formed correctly without ROUTER24.
         self.nodes[LEADER].start()
         self.nodes[LEADER].set_state('leader')
         self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
 
-        for i in range(2, 25):
+        for i in range(2, 24):
             self.nodes[i].start()
             time.sleep(5)
             self.assertEqual(self.nodes[i].get_state(), 'router')
 
-        self.nodes[25].start()
-        time.sleep(5)
-        self.assertEqual(self.nodes[25].get_state(), 'router')
+        # This method flushes the message queue so calling this method again will return only the newly logged messages.
+        dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER1)
 
+        # 2 ROUTER24: Attach to network.
+        # All reference testbed devices have been configured with downgrade threshold as 32 except DUT_ROUTER1,
+        # so we don't need to ensure ROUTER24 has a better link quality on posix.
+        self.nodes[ROUTER24].start()
+        time.sleep(5)
+        self.assertEqual(self.nodes[ROUTER24].get_state(), 'router')
+
+        # 3 DUT_ROUTER1:
         time.sleep(10)
-        self.assertEqual(self.nodes[ROUTER1].get_state(), 'child')
+        # Verify it sent a Parent Request and Child ID Request.
+        dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER1)
+        dut_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
+        dut_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
+
+        # Verify it sent an Address Release Message to the Leader when it attached as a child.
+        self.assertEqual(self.nodes[DUT_ROUTER1].get_state(), 'child')
+        msg = dut_messages.next_coap_message('0.02')
+        command.check_address_release(msg, self.nodes[LEADER])
+
+        # 4 & 5
+        router1_rloc = self.nodes[DUT_ROUTER1].get_ip6_address(config.ADDRESS_TYPE.RLOC)
+        self.assertTrue(self.nodes[LEADER].ping(router1_rloc))
 
 if __name__ == '__main__':
     unittest.main()

tests/scripts/thread-cert/Makefile.am

@@ -121,6 +121,7 @@ EXTRA_DIST                                                         = \
     Cert_9_2_17_Orphan.py                                            \
     Cert_9_2_18_RollBackActiveTimestamp.py                           \
     coap.py                                                          \
+    command.py                                                       \
     common.py                                                        \
     config.py                                                        \
     ipv6.py                                                          \

tests/scripts/thread-cert/command.py

@@ -11,9 +11,18 @@ def check_address_notification(command_msg, source_node, destination_node):
     command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
     command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)
 
-    #pdb.set_trace()
     source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 Source address is not the RLOC of the originator."
 
     destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
     assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 Destination address is not the RLOC of the destination."
+
+def check_address_release(command_msg, destination_node):
+    """Verify the message is a properly formatted address release destined to the given node.
+    """
+    command_msg.assertCoapMessageRequestUriPath('/a/ar')
+    command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16)
+    command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress)
+
+    destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC)
+    assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The Destination is not RLOC address"

tests/scripts/thread-cert/config.py

@@ -51,6 +51,7 @@ DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
                                 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff])
 
 ADDRESS_TYPE = Enum('ADDRESS_TYPE', ('LINK_LOCAL', 'GLOBAL', 'RLOC', 'ALOC', 'ML_EID'))
+RSSI = {'LINK_QULITY_0': -100, 'LINK_QULITY_1': -95, 'LINK_QULITY_2': -85, 'LINK_QULITY_3': -65}
 
 SNIFFER_ID = int(os.getenv('SNIFFER_ID', 34))
 PANID = 0xface
@@ -203,6 +204,7 @@ def create_default_uri_path_based_payload_factories():
     return {
         "/a/as": network_layer_tlvs_factory,
         "/a/aq": network_layer_tlvs_factory,
+        "/a/ar": network_layer_tlvs_factory,
         "/a/an": network_layer_tlvs_factory
     }