espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-20 09:29:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] add traffic analysis for Cert_5_3_07_DuplicateAddress.py (#2298)

Browse Source
This commit is contained in:
hjian2017
2017-11-11 16:24:38 -08:00
committed by Jonathan Hui
parent c4dc72cf2d
commit d1f577d068
3 changed files with 100 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_3_07_DuplicateAddress.py
+72 -48
View File
@@ -31,15 +31,18 @@ import time
import unittest
import node
import command
import config
import mle
LEADER = 1
DUT_LEADER = 1
ROUTER1 = 2
ROUTER2 = 3
ED1 = 4
ED2 = 5
ED3 = 6
MED1 = 4
SED1 = 5
MED3 = 6
MTDS = [ED1, ED2, ED3]
MTDS = [MED1, SED1, MED3]
class Cert_5_3_7_DuplicateAddress(unittest.TestCase):
def setUp(self):
@@ -47,80 +50,101 @@ class Cert_5_3_7_DuplicateAddress(unittest.TestCase):
for i in range(1,7):
self.nodes[i] = node.Node(i, (i in MTDS))
self.nodes[LEADER].set_panid(0xface)
self.nodes[LEADER].set_mode('rsdn')
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ED3].get_addr64())
self.nodes[LEADER].enable_whitelist()
self.nodes[DUT_LEADER].set_panid(0xface)
self.nodes[DUT_LEADER].set_mode('rsdn')
self.nodes[DUT_LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[DUT_LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[DUT_LEADER].add_whitelist(self.nodes[MED3].get_addr64())
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[ED1].get_addr64())
self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
self.nodes[ROUTER1].add_whitelist(self.nodes[MED1].get_addr64())
self.nodes[ROUTER1].enable_whitelist()
self.nodes[ROUTER1].set_router_selection_jitter(1)
self.nodes[ROUTER2].set_panid(0xface)
self.nodes[ROUTER2].set_mode('rsdn')
self.nodes[ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[ED2].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[SED1].get_addr64())
self.nodes[ROUTER2].enable_whitelist()
self.nodes[ROUTER2].set_router_selection_jitter(1)
self.nodes[ED1].set_panid(0xface)
self.nodes[ED1].set_mode('rsn')
self.nodes[ED1].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[ED1].enable_whitelist()
self.nodes[MED1].set_panid(0xface)
self.nodes[MED1].set_mode('rsn')
self.nodes[MED1].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[MED1].enable_whitelist()
self.nodes[ED2].set_panid(0xface)
self.nodes[ED2].set_mode('rsn')
self.nodes[ED2].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[ED2].enable_whitelist()
self.nodes[SED1].set_panid(0xface)
self.nodes[SED1].set_mode('s')
self.nodes[SED1].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[SED1].enable_whitelist()
self.nodes[ED3].set_panid(0xface)
self.nodes[ED3].set_mode('rsn')
self.nodes[ED3].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[ED3].enable_whitelist()
self.nodes[MED3].set_panid(0xface)
self.nodes[MED3].set_mode('rsn')
self.nodes[MED3].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
self.nodes[MED3].enable_whitelist()
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):
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(ROUTER1, MED3 + 1):
self.nodes[i].start()
self.nodes[ROUTER1].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.nodes[ROUTER2].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
for i in [ROUTER1, ROUTER2]:
self.assertEqual(self.nodes[i].get_state(), 'router')
self.nodes[ED1].start()
time.sleep(5)
self.assertEqual(self.nodes[ED1].get_state(), 'child')
for i in MTDS:
self.assertEqual(self.nodes[i].get_state(), 'child')
self.nodes[ED2].start()
time.sleep(5)
self.assertEqual(self.nodes[ED2].get_state(), 'child')
self.nodes[ED3].start()
time.sleep(5)
self.assertEqual(self.nodes[ED3].get_state(), 'child')
# 2
leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
msg = leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT)
command.check_mle_advertisement(msg)
# 3
self.nodes[ROUTER2].add_prefix('2001:2:0:1::/64', 'paros')
self.nodes[ROUTER2].register_netdata()
self.nodes[ED1].add_ipaddr('2001:2:0:1::1')
self.nodes[ED2].add_ipaddr('2001:2:0:1::1')
self.nodes[MED1].add_ipaddr('2001:2:0:1::1234')
self.nodes[SED1].add_ipaddr('2001:2:0:1::1234')
time.sleep(5)
self.assertTrue(self.nodes[ED3].ping('2001:2:0:1::1'))
# 4
# Flush the message queue to avoid possible impact on follow-up verification.
self.sniffer.get_messages_sent_by(DUT_LEADER)
self.nodes[MED3].ping('2001:2:0:1::1234')
# Verify DUT_LEADER sent an Address Query Request to the Realm local address.
dut_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
msg = dut_messages.next_coap_message('0.02', '/a/aq')
command.check_address_query(msg, self.nodes[DUT_LEADER], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS)
# 5 & 6
# Verify DUT_LEADER sent an Address Error Notification to the Realm local address.
time.sleep(5)
dut_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
msg = dut_messages.next_coap_message('0.02', '/a/ae')
command.check_address_error_notification(msg, self.nodes[DUT_LEADER], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/command.py
+26 -1
View File
@@ -12,7 +12,7 @@ def check_address_query(command_msg, source_node, destination_address):
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. The source node's rloc is: " \
+ str(ipv6.ip_address(source_rloc)) + ", but the source_rloc in command msg is: " \
+ str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
+ str(command_msg.ipv6_packet.ipv6_header.source_address)
assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 destination address is not expected."
@@ -31,6 +31,24 @@ def check_address_notification(command_msg, source_node, destination_node):
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_error_notification(command_msg, source_node, destination_address):
"""Verify source_node sent a properly formatted Address Error Notification command message to destination_address.
"""
command_msg.assertCoapMessageRequestUriPath('/a/ae')
command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
command_msg.assertCoapMessageContainsTlv(network_layer.MlEid)
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. The source node's rloc is: " \
+ str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " \
+ str(command_msg.ipv6_packet.ipv6_header.source_address)
assert ipv6.ip_address(destination_address.decode('utf-8')) == command_msg.ipv6_packet.ipv6_header.destination_address, \
"Error: The IPv6 destination address is not expected. The destination node's rloc is: " \
+ str(ipv6.ip_address(destination_address.decode('utf-8'))) + ", but the destination_address in command msg is: " \
+ str(command_msg.ipv6_packet.ipv6_header.destination_address)
def check_address_release(command_msg, destination_node):
"""Verify the message is a properly formatted address release destined to the given node.
"""
@@ -106,3 +124,10 @@ def get_routing_cost(command_msg, router_id):
%(tlv, router_id, routing_entry_pos, router_id_mask_str)
return tlv.link_quality_and_route_data[routing_entry_pos].route
def check_mle_advertisement(command_msg):
command_msg.assertSentWithHopLimit(255)
command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_NODES_ADDRESS)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
command_msg.assertMleMessageContainsTlv(mle.Route64)
tests/scripts/thread-cert/config.py
+2
View File
@@ -49,6 +49,7 @@ ALOC_FLAG_REGEX_PATTERN = '.*:fc..$'
LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = 'ff32:40:fdde:ad00:beef:0:0:1'
REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = 'ff33:40:fdde:ad00:beef:0:0:1'
REALM_LOCAL_ALL_ROUTERS_ADDRESS = 'ff03::2'
LINK_LOCAL_ALL_NODES_ADDRESS = 'ff02::1'
DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff])
@@ -213,6 +214,7 @@ def create_default_uri_path_based_payload_factories():
"/a/as": network_layer_tlvs_factory,
"/a/aq": network_layer_tlvs_factory,
"/a/ar": network_layer_tlvs_factory,
"/a/ae": network_layer_tlvs_factory,
"/a/an": network_layer_tlvs_factory
}