From 97a2fc4cc3415f7388d57eae4a45d2407cb8f2d3 Mon Sep 17 00:00:00 2001
From: hjian2017 <30794429+hjian2017@users.noreply.github.com>
Date: Wed, 23 Aug 2017 09:35:10 +0800
Subject: [PATCH] [test] Add traffic analysis for Cert_6_1_01_RouterAttach
 (#2119)

Also fix the method assertMleMessageContainsOptionalTlv() in message.py to
output appropriate log.
---
 .../thread-cert/Cert_6_1_01_RouterAttach.py   | 58 ++++++++++++++++++-
 tests/scripts/thread-cert/message.py          |  5 +-
 2 files changed, 61 insertions(+), 2 deletions(-)

diff --git a/tests/scripts/thread-cert/Cert_6_1_01_RouterAttach.py b/tests/scripts/thread-cert/Cert_6_1_01_RouterAttach.py
index 4be313fff..4222a3a5c 100755
--- a/tests/scripts/thread-cert/Cert_6_1_01_RouterAttach.py
+++ b/tests/scripts/thread-cert/Cert_6_1_01_RouterAttach.py
@@ -30,10 +30,13 @@
 import time
 import unittest
 
+import config
+import mle
 import node
 
 LEADER = 1
 ED = 2
+SNIFFER = 3
 
 class Cert_6_1_1_RouterAttach(unittest.TestCase):
     def setUp(self):
@@ -51,19 +54,72 @@ class Cert_6_1_1_RouterAttach(unittest.TestCase):
         self.nodes[ED].add_whitelist(self.nodes[LEADER].get_addr64())
         self.nodes[ED].enable_whitelist()
 
+        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')
+        time.sleep(5)
         self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
 
         self.nodes[ED].start()
         time.sleep(5)
         self.assertEqual(self.nodes[ED].get_state(), 'child')
 
+        leader_messages = self.sniffer.get_messages_sent_by(LEADER)
+        ed_messages = self.sniffer.get_messages_sent_by(ED)
+
+        # 1 - leader
+        msg = leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT)
+
+        # 2 - ed
+        msg = ed_messages.next_mle_message(mle.CommandType.PARENT_REQUEST)
+        self.assertEqual(0x02, msg.mle.aux_sec_hdr.key_id_mode)
+        msg.assertSentWithHopLimit(255)
+        msg.assertSentToDestinationAddress("ff02::2")
+        msg.assertMleMessageContainsTlv(mle.Mode)
+        msg.assertMleMessageContainsTlv(mle.Challenge)
+        msg.assertMleMessageContainsTlv(mle.ScanMask)
+        msg.assertMleMessageContainsTlv(mle.Version)
+
+        scan_mask_tlv = msg.get_mle_message_tlv(mle.ScanMask)
+        self.assertEqual(1, scan_mask_tlv.router)
+        self.assertEqual(0, scan_mask_tlv.end_device)
+
+        # 3 - leader
+        msg = leader_messages.next_mle_message(mle.CommandType.PARENT_RESPONSE)
+        msg.assertSentToNode(self.nodes[ED])
+
+        # 4 - ed
+        msg = ed_messages.next_mle_message(mle.CommandType.CHILD_ID_REQUEST)
+        self.assertEqual(0x02, msg.mle.aux_sec_hdr.key_id_mode)
+        msg.assertSentToNode(self.nodes[LEADER])
+        msg.assertMleMessageContainsTlv(mle.AddressRegistration)
+        msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
+        msg.assertMleMessageContainsTlv(mle.Mode)
+        msg.assertMleMessageContainsTlv(mle.Response)
+        msg.assertMleMessageContainsTlv(mle.Timeout)
+        msg.assertMleMessageContainsTlv(mle.TlvRequest)
+        msg.assertMleMessageContainsTlv(mle.Version)
+        msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter)
+
+        # 5 - leader
+        msg = leader_messages.next_mle_message(mle.CommandType.CHILD_ID_RESPONSE)
+        msg.assertSentToNode(self.nodes[ED])
+
+        # 6 - leader
+        ed_addrs = self.nodes[ED].get_addrs()
+        for addr in ed_addrs:
+            if addr[0:4] != 'fe80':
+                self.assertTrue(self.nodes[LEADER].ping(addr))
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py
index 020a02508..c178dd1f1 100644
--- a/tests/scripts/thread-cert/message.py
+++ b/tests/scripts/thread-cert/message.py
@@ -189,7 +189,10 @@ class Message(object):
                 contains_tlv = True
                 break
 
-        print("MleMessage doesn't contain optional TLV: {}".format(tlv_class_type))
+        if contains_tlv == True:
+            print("MleMessage contains optional TLV: {}".format(tlv_class_type))
+        else:
+            print("MleMessage doesn't contain optional TLV: {}".format(tlv_class_type))
 
     def get_coap_message_tlv(self, tlv_class_type):
         if self.type != MessageType.COAP: