mirror of
https://github.com/espressif/openthread.git
synced 2026-07-28 14:47:46 +00:00
[test] Add traffic analysis for Cert_6_1_01_RouterAttach (#2119)
Also fix the method assertMleMessageContainsOptionalTlv() in message.py to output appropriate log.
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user