espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-28 14:47:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] add traffic analysis for Cert_5_2_05_AddressQuery.py (#2173)

Browse Source 
Add a fuction in node.py node_cli.py to node's address.
Add a module command to verify commands.
Spinel-cli.py doesn't request address for dhcp prefix now, so put
Cert_5_2_05_AddressQuery.py in XFAIL on ncp.
This commit is contained in:
hjian2017
2017-09-13 01:44:56 -07:00
committed by Jonathan Hui
parent 79f8a5a82b
commit d6a3519036
6 changed files with 191 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_2_05_AddressQuery.py
+83 -68
View File
@@ -31,111 +31,126 @@ import time
import unittest
import node
import config
import command
ED1 = 1
BR1 = 2
LEADER = 3
ROUTER2 = 4
REED = 5
ED2 = 6
ED3 = 7
MTDS = [ED1, ED2, ED3]
LEADER = 1
ROUTER1 = 2
BR = 3
ED1 = 17
DUT_REED = 18
ROUTER_SELECTION_JITTER = 1
class Cert_5_2_5_AddressQuery(unittest.TestCase):
def setUp(self):
self.nodes = {}
for i in range(1,8):
self.nodes[i] = node.Node(i, (i in MTDS))
for i in range(1, 19):
self.nodes[i] = node.Node(i, (i == ED1))
self.nodes[LEADER].set_panid(0xface)
self.nodes[LEADER].set_panid()
self.nodes[LEADER].set_mode('rsdn')
self.nodes[LEADER].add_whitelist(self.nodes[BR1].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[REED].get_addr64())
self.nodes[LEADER].enable_whitelist()
self.nodes[BR1].set_panid(0xface)
self.nodes[BR1].set_mode('rsdn')
self.nodes[BR1].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[BR1].add_whitelist(self.nodes[ED1].get_addr64())
self.nodes[BR1].enable_whitelist()
self.nodes[BR1].set_router_selection_jitter(1)
for i in range(2, 17):
self.nodes[i].set_panid()
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].enable_whitelist()
self.nodes[i].set_router_selection_jitter(1)
self.nodes[ED1].set_panid(0xface)
self.nodes[ED1].set_panid()
self.nodes[ED1].set_mode('rsn')
self.nodes[ED1].add_whitelist(self.nodes[BR1].get_addr64())
self.nodes[ED1].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ED1].get_addr64())
self.nodes[ED1].enable_whitelist()
self.nodes[REED].set_panid(0xface)
self.nodes[REED].set_mode('rsdn')
self.nodes[REED].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[REED].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[REED].set_router_upgrade_threshold(0)
self.nodes[REED].enable_whitelist()
self.nodes[DUT_REED].set_panid()
self.nodes[DUT_REED].set_mode('rsdn')
self.nodes[DUT_REED].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_REED].get_addr64())
self.nodes[DUT_REED].enable_whitelist()
self.nodes[DUT_REED].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[REED].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[ED2].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[ED3].get_addr64())
self.nodes[ROUTER2].enable_whitelist()
self.nodes[ROUTER2].set_router_selection_jitter(1)
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[ED3].set_panid(0xface)
self.nodes[ED3].set_mode('rsn')
self.nodes[ED3].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[ED3].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):
# 1. LEADER: DHCPv6 Server for prefix 2001::/64.
self.nodes[LEADER].start()
self.nodes[LEADER].set_state('leader')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[LEADER].add_prefix('2001::/64', 'pdros')
self.nodes[LEADER].register_netdata()
self.nodes[BR1].start()
# 2. BR: SLAAC Server for prefix 2002::/64.
self.nodes[BR].start()
time.sleep(5)
self.assertEqual(self.nodes[BR1].get_state(), 'router')
self.assertEqual(self.nodes[BR].get_state(), 'router')
self.nodes[BR].add_prefix('2002::/64', 'paros')
self.nodes[BR].register_netdata()
self.nodes[BR1].add_prefix('2001:2:0:3::/64', 'paros')
self.nodes[BR1].add_prefix('2001:2:0:4::/64', 'paros')
self.nodes[BR1].register_netdata()
# 3. Bring up remaining devices except DUT_REED.
for i in range(2, 17):
if i == BR:
continue
self.nodes[i].start()
time.sleep(5)
self.assertEqual(self.nodes[i].get_state(), 'router')
self.nodes[ED1].start()
time.sleep(5)
self.assertEqual(self.nodes[ED1].get_state(), 'child')
self.nodes[REED].start()
# 4. Bring up DUT_REED.
self.nodes[DUT_REED].start()
time.sleep(5)
self.assertEqual(self.nodes[REED].get_state(), 'child')
time.sleep(ROUTER_SELECTION_JITTER)
self.nodes[ROUTER2].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
reed_messages = self.sniffer.get_messages_sent_by(DUT_REED)
self.nodes[ED2].start()
time.sleep(5)
self.assertEqual(self.nodes[ED2].get_state(), 'child')
# Verify DUT_REED doesn't try to become router.
msg = reed_messages.does_not_contain_coap_message()
assert msg is True, "Error: The REED sent an Address Solicit Request"
self.nodes[ED3].start()
time.sleep(5)
self.assertEqual(self.nodes[ED3].get_state(), 'child')
# 5. Enable a link between the DUT and BR to create a one-way link.
self.nodes[DUT_REED].add_whitelist(self.nodes[BR].get_addr64())
self.nodes[BR].add_whitelist(self.nodes[DUT_REED].get_addr64())
addrs = self.nodes[REED].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[ED2].ping(addr))
time.sleep(1)
# 6. Verify DUT_REED would send Address Notification when ping to its ML-EID.
mleid = self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
self.assertTrue(self.nodes[ED1].ping(mleid))
reed_messages = self.sniffer.get_messages_sent_by(DUT_REED)
msg = reed_messages.next_coap_message('0.02')
command.check_address_notification(msg, self.nodes[DUT_REED], self.nodes[LEADER])
# 7 & 8. Verify DUT_REED would send Address Notification when ping to its 2001::EID and 2002::EID.
flag2001 = 0
flag2002 = 0
for global_address in self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.GLOBAL):
if global_address[0:4] == '2001':
flag2001 += 1
elif global_address[0:4] == '2002':
flag2002 += 1
else:
raise "Error: Address is unexpected."
self.assertTrue(self.nodes[ED1].ping(global_address))
reed_messages = self.sniffer.get_messages_sent_by(DUT_REED)
msg = reed_messages.next_coap_message('0.02')
command.check_address_notification(msg, self.nodes[DUT_REED], self.nodes[LEADER])
assert flag2001 == 1 , "Error: Expecting address 2001::EID not appear."
assert flag2002 == 1 , "Error: Expecting address 2002::EID not appear."
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Makefile.am
+1 -1
View File
@@ -174,7 +174,6 @@ check_SCRIPTS = \
Cert_5_1_12_NewRouterNeighborSync.py \
Cert_5_1_13_RouterReset.py \
Cert_5_2_01_BecomeActiveRouter.py \
Cert_5_2_05_AddressQuery.py \
Cert_5_2_06_RouterDowngrade.py \
Cert_5_2_07_REEDSynchronization.py \
Cert_5_2_04_REEDUpgrade.py \
@@ -262,6 +261,7 @@ TESTS = \
$(NULL)
XFAIL_NCP_TESTS = \
Cert_5_2_05_AddressQuery.py \
Cert_5_3_09_AddressQuery.py \
Cert_8_1_01_Commissioning.py \
Cert_8_1_02_Commissioning.py \
tests/scripts/thread-cert/command.py
+19
View File
@@ -0,0 +1,19 @@
import ipv6
import network_layer
import config
def check_address_notification(command_msg, source_node, destination_node):
"""Verify source_node sent a properly formatted Address Notification command message to destination_node.
"""
command_msg.assertCoapMessageRequestUriPath('/a/an')
command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
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."
tests/scripts/thread-cert/config.py
+16 -2
View File
@@ -26,6 +26,8 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import os
from enum import Enum
import coap
import ipv6
@@ -37,9 +39,21 @@ import network_data
import network_layer
import sniffer
MESH_LOCAL_PREFIX = 'fdde:ad00:beef::/64'
MESH_LOCAL_PREFIX_REGEX_PATTERN = '^fdde:ad00:beef:(0){0,4}:'
ROUTING_LOCATOR = '64/:0:ff:fe00:/16'
ROUTING_LOCATOR_REGEX_PATTERN = '.*:(0)?:0{0,2}ff:fe00:\w{1,4}$'
LINK_LOCAL = 'fe80:/112'
LINK_LOCAL_REGEX_PATTERN = '^fe80:.*'
ALOC_FLAG_REGEX_PATTERN = '.*:fc..$'
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'))
SNIFFER_ID = int(os.getenv('SNIFFER_ID', 34))
PANID = 0xface
def create_default_network_data_prefix_sub_tlvs_factories():
return {
@@ -279,7 +293,7 @@ def create_default_lowpan_decompressor(context_manager):
def create_default_thread_context_manager():
context_manager = lowpan.ContextManager()
context_manager[0] = lowpan.Context("fd00:0db8::/64")
context_manager[0] = lowpan.Context(MESH_LOCAL_PREFIX)
return context_manager
@@ -300,5 +314,5 @@ def create_default_thread_message_factory(master_key=DEFAULT_MASTER_KEY):
return message.MessageFactory(lowpan_parser=lowpan_parser)
def create_default_thread_sniffer(nodeid):
def create_default_thread_sniffer(nodeid=SNIFFER_ID):
return sniffer.Sniffer(nodeid, create_default_thread_message_factory())
tests/scripts/thread-cert/node.py
+5 -1
View File
@@ -35,6 +35,7 @@ if sys.platform != 'win32':
else:
import node_api
import unittest
import config
class Node:
def __init__(self, nodeid, is_mtd=False):
@@ -142,7 +143,7 @@ class Node:
def get_panid(self):
return self.interface.get_panid()
def set_panid(self, panid):
def set_panid(self, panid = config.PANID):
self.interface.set_panid(panid)
def get_partition_id(self):
@@ -187,6 +188,9 @@ class Node:
def get_addrs(self):
return self.interface.get_addrs()
def get_ip6_address(self, address_type):
return self.interface.get_ip6_address(address_type)
def get_context_reuse_delay(self):
return self.interface.get_context_reuse_delay()
tests/scripts/thread-cert/node_cli.py
+67 -3
View File
@@ -31,6 +31,9 @@ import os
import sys
import time
import pexpect
import re
import config
class otCli:
def __init__(self, nodeid, is_mtd):
@@ -83,7 +86,7 @@ class otCli:
time.sleep(0.2)
self.pexpect.expect('spinel-cli >')
self.debug(int(os.getenv('DEBUG', '0')))
def __init_soc(self, nodeid):
""" Initialize a System-on-a-chip node connected via UART. """
import fdpexpect
@@ -98,7 +101,7 @@ class otCli:
self.pexpect.close(force=True)
def send_command(self, cmd):
print ("%d: %s" % (self.nodeid, cmd))
print("%d: %s" % (self.nodeid, cmd))
self.pexpect.sendline(cmd)
def get_commands(self):
@@ -136,7 +139,7 @@ class otCli:
def thread_stop(self):
self.send_command('thread stop')
self.pexpect.expect('Done')
def commissioner_start(self):
cmd = 'commissioner start'
self.send_command(cmd)
@@ -372,6 +375,67 @@ class otCli:
return addrs
def __getLinkLocalAddress(self):
for ip6Addr in self.get_addrs():
if re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I):
return ip6Addr
return None
def __getGlobalAddress(self):
global_address = []
for ip6Addr in self.get_addrs():
if (not re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I)) and \
(not re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I)) and \
(not re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)):
global_address.append(ip6Addr)
return global_address
def __getRloc(self):
for ip6Addr in self.get_addrs():
if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \
re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \
not(re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I)):
return ip6Addr
return None
def __getAloc(self):
aloc = []
for ip6Addr in self.get_addrs():
if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \
re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \
re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I):
aloc.append(ip6Addr)
return aloc
def __getMleid(self):
for ip6Addr in self.get_addrs():
if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \
not(re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)):
return ip6Addr
return None
def get_ip6_address(self, address_type):
"""Get specific type of IPv6 address configured on thread device.
Args:
address_type: the config.ADDRESS_TYPE type of IPv6 address.
Returns:
IPv6 address string.
"""
switcher = {
config.ADDRESS_TYPE.LINK_LOCAL: self.__getLinkLocalAddress(),
config.ADDRESS_TYPE.GLOBAL: self.__getGlobalAddress(),
config.ADDRESS_TYPE.RLOC: self.__getRloc(),
config.ADDRESS_TYPE.ALOC: self.__getAloc(),
config.ADDRESS_TYPE.ML_EID: self.__getMleid()
}
return switcher.get(address_type, None)
def get_context_reuse_delay(self):
self.send_command('contextreusedelay')
i = self.pexpect.expect('(\d+)\r\n')