espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-08 19:57:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] add traffic to 4 test cases: from Cert_5_2_07 to Cert_5_3_03 (#2214)

Browse Source 
- Add a method in message.py to verify if receive a ICMPv6 message.
- Add two methods in command.py to verify link request and link accept.
This commit is contained in:
hjian2017
2017-10-18 09:25:22 -07:00
committed by Jonathan Hui
parent e2ca9caac3
commit 5bc2b8113b
8 changed files with 265 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_2_07_REEDSynchronization.py
+53 -54
View File
@@ -29,86 +29,85 @@
import time
import unittest
import ipv6
import node
import mle
import config
import command
LEADER = 1
ROUTER1 = 2
REED = 3
ROUTER2 = 4
ROUTER3 = 5
DUT_ROUTER1 = 2
DUT_REED = 17
class Cert_5_2_7_REEDSynchronization(unittest.TestCase):
def setUp(self):
self.nodes = {}
for i in range(1,6):
for i in range(1, 18):
self.nodes[i] = node.Node(i)
self.nodes[i].set_panid(0xface)
self.nodes[i].set_mode('rsdn')
self.nodes[i].set_router_selection_jitter(1)
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[REED].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[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[REED].get_addr64())
self.nodes[ROUTER1].add_whitelist(self.nodes[ROUTER3].get_addr64())
self.nodes[ROUTER1].enable_whitelist()
self.nodes[ROUTER1].set_router_selection_jitter(1)
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[ROUTER1].get_addr64())
self.nodes[REED].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[REED].add_whitelist(self.nodes[ROUTER3].get_addr64())
self.nodes[REED].enable_whitelist()
self.nodes[REED].set_router_upgrade_threshold(0)
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[ROUTER3].get_addr64())
self.nodes[ROUTER2].enable_whitelist()
self.nodes[ROUTER2].set_router_selection_jitter(1)
self.nodes[ROUTER3].set_panid(0xface)
self.nodes[ROUTER3].set_mode('rsdn')
self.nodes[ROUTER3].add_whitelist(self.nodes[REED].get_addr64())
self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[ROUTER3].enable_whitelist()
self.nodes[ROUTER3].set_router_selection_jitter(1)
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 DUT_ROUTER1.
self.nodes[LEADER].start()
self.nodes[LEADER].set_state('leader')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[REED].start()
for i in range(2, 17):
self.nodes[i].start()
time.sleep(5)
self.assertEqual(self.nodes[REED].get_state(), 'child')
self.nodes[ROUTER1].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
for i in range(2, 17):
self.assertEqual(self.nodes[i].get_state(), 'router')
self.nodes[ROUTER2].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
# 2. DUT_REED: Attach to network. Verify it didn't send an Address Solicit Request.
# Avoid DUT_REED attach to DUT_ROUTER1.
self.nodes[DUT_REED].add_whitelist(self.nodes[DUT_ROUTER1].get_addr64(), config.RSSI['LINK_QULITY_1'])
self.nodes[ROUTER3].start()
self.nodes[DUT_REED].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER3].get_state(), 'router')
self.assertEqual(self.nodes[DUT_REED].get_state(), 'child')
# The DUT_REED must not send a coap message here.
reed_messages = self.sniffer.get_messages_sent_by(DUT_REED)
msg = reed_messages.does_not_contain_coap_message()
assert msg is True, "Error: The DUT_REED sent an Address Solicit Request"
# 3. DUT_REED: Verify sent a Link Request to at least 3 neighboring Routers.
for i in range(0, 3):
msg = reed_messages.next_mle_message(mle.CommandType.LINK_REQUEST)
command.check_link_request(msg)
# 4. DUT_ROUTER1: Verify sent a Link Accept to DUT_REED.
time.sleep(30)
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER1)
flag_link_accept = False
while True:
msg = dut_messages.next_mle_message(mle.CommandType.LINK_ACCEPT, False)
if msg == None :
break
destination_link_local = self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL)
if ipv6.ip_address(destination_link_local) == msg.ipv6_packet.ipv6_header.destination_address:
flag_link_accept = True
break
assert flag_link_accept is True, "Error: DUT_ROUTER1 didn't send a Link Accept to DUT_REED"
command.check_link_accept(msg, self.nodes[DUT_REED])
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Cert_5_3_01_LinkLocal.py
+18 -15
View File
@@ -31,9 +31,11 @@ import time
import unittest
import node
import config
import command
LEADER = 1
ROUTER1 = 2
DUT_ROUTER1 = 2
class Cert_5_3_1_LinkLocal(unittest.TestCase):
def setUp(self):
@@ -43,14 +45,10 @@ class Cert_5_3_1_LinkLocal(unittest.TestCase):
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].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].enable_whitelist()
self.nodes[ROUTER1].set_router_selection_jitter(1)
self.nodes[DUT_ROUTER1].set_panid(0xface)
self.nodes[DUT_ROUTER1].set_mode('rsdn')
self.nodes[DUT_ROUTER1].set_router_selection_jitter(1)
def tearDown(self):
for node in list(self.nodes.values()):
@@ -58,25 +56,30 @@ class Cert_5_3_1_LinkLocal(unittest.TestCase):
del self.nodes
def test(self):
# 1
self.nodes[LEADER].start()
self.nodes[LEADER].set_state('leader')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[ROUTER1].start()
self.nodes[DUT_ROUTER1].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.assertEqual(self.nodes[DUT_ROUTER1].get_state(), 'router')
addrs = self.nodes[ROUTER1].get_addrs()
for addr in addrs:
if addr[0:4] == 'fe80':
self.assertTrue(self.nodes[LEADER].ping(addr, size=256))
self.assertTrue(self.nodes[LEADER].ping(addr))
# 2 & 3
link_local = self.nodes[DUT_ROUTER1].get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL)
self.assertTrue(self.nodes[LEADER].ping(link_local, size=256))
self.assertTrue(self.nodes[LEADER].ping(link_local))
# 4 & 5
self.assertTrue(self.nodes[LEADER].ping('ff02::1', size=256))
self.assertTrue(self.nodes[LEADER].ping('ff02::1'))
# 6 & 7
self.assertTrue(self.nodes[LEADER].ping('ff02::2', size=256))
self.assertTrue(self.nodes[LEADER].ping('ff02::2'))
# 8
self.assertTrue(self.nodes[LEADER].ping(config.LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS))
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py
+47 -19
View File
@@ -31,10 +31,12 @@ import time
import unittest
import node
import config
import command
LEADER = 1
ROUTER1 = 2
ROUTER2 = 3
DUT_ROUTER2 = 3
SED1 = 4
class Cert_5_3_2_RealmLocal(unittest.TestCase):
@@ -51,29 +53,36 @@ class Cert_5_3_2_RealmLocal(unittest.TestCase):
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[ROUTER2].get_addr64())
self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_ROUTER2].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[ROUTER1].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[DUT_ROUTER2].set_panid(0xface)
self.nodes[DUT_ROUTER2].set_mode('rsdn')
self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[SED1].get_addr64())
self.nodes[DUT_ROUTER2].enable_whitelist()
self.nodes[DUT_ROUTER2].set_router_selection_jitter(1)
self.nodes[SED1].set_panid(0xface)
self.nodes[SED1].set_mode('sn')
self.nodes[SED1].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[SED1].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64())
self.nodes[SED1].enable_whitelist()
self.nodes[SED1].set_timeout(3)
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
self.nodes[LEADER].start()
self.nodes[LEADER].set_state('leader')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
@@ -82,28 +91,47 @@ class Cert_5_3_2_RealmLocal(unittest.TestCase):
time.sleep(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.nodes[ROUTER2].start()
self.nodes[DUT_ROUTER2].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.assertEqual(self.nodes[DUT_ROUTER2].get_state(), 'router')
self.nodes[SED1].start()
time.sleep(5)
self.assertEqual(self.nodes[SED1].get_state(), 'child')
addrs = self.nodes[ROUTER2].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[LEADER].ping(addr, size=256))
self.assertTrue(self.nodes[LEADER].ping(addr))
# 2 & 3
mleid = self.nodes[DUT_ROUTER2].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
self.assertTrue(self.nodes[LEADER].ping(mleid, size=256))
self.assertTrue(self.nodes[LEADER].ping(mleid))
# 4 & 5
self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2, size=256))
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2))
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
# 6 & 7
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2, size=256))
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2))
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.assertTrue(self.nodes[LEADER].ping('ff33:0040:fdde:ad00:beef:0:0:1', num_responses=3, size=256))
self.assertTrue(self.nodes[LEADER].ping('ff33:0040:fdde:ad00:beef:0:0:1', num_responses=3))
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2))
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
# 8
self.assertTrue(self.nodes[LEADER].ping(config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS, num_responses=3, size=256))
time.sleep(2)
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertTrue(sed_messages.contains_icmp_message())
self.assertTrue(self.nodes[LEADER].ping(config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS, num_responses=3))
time.sleep(2)
sed_messages = self.sniffer.get_messages_sent_by(SED1)
self.assertTrue(sed_messages.contains_icmp_message())
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Cert_5_3_03_AddressQuery.py
+88 -73
View File
@@ -31,121 +31,136 @@ import time
import unittest
import node
import config
import command
BR = 1
LEADER = 2
ROUTER2 = 3
LEADER = 1
ROUTER1 = 2
DUT_ROUTER2 = 3
ROUTER3 = 4
ED2 = 5
MED1 = 5
MED1_TIMEOUT = 3
class Cert_5_3_3_AddressQuery(unittest.TestCase):
def setUp(self):
self.nodes = {}
for i in range(1,6):
self.nodes[i] = node.Node(i, (i == ED2))
self.nodes[i] = node.Node(i, (i == MED1))
self.nodes[LEADER].set_panid(0xface)
self.nodes[LEADER].set_panid()
self.nodes[LEADER].set_mode('rsdn')
self.nodes[LEADER].add_whitelist(self.nodes[BR].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER1].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64())
self.nodes[LEADER].add_whitelist(self.nodes[ROUTER3].get_addr64())
self.nodes[LEADER].enable_whitelist()
self.nodes[BR].set_panid(0xface)
self.nodes[BR].set_mode('rsdn')
self.nodes[BR].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[BR].enable_whitelist()
self.nodes[BR].set_router_selection_jitter(1)
self.nodes[ROUTER1].set_panid()
self.nodes[ROUTER1].set_mode('rsdn')
self.nodes[ROUTER1].add_whitelist(self.nodes[LEADER].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[ROUTER3].get_addr64())
self.nodes[ROUTER2].add_whitelist(self.nodes[ED2].get_addr64())
self.nodes[ROUTER2].enable_whitelist()
self.nodes[ROUTER2].set_router_selection_jitter(1)
self.nodes[DUT_ROUTER2].set_panid()
self.nodes[DUT_ROUTER2].set_mode('rsdn')
self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[ROUTER3].get_addr64())
self.nodes[DUT_ROUTER2].add_whitelist(self.nodes[MED1].get_addr64())
self.nodes[DUT_ROUTER2].enable_whitelist()
self.nodes[DUT_ROUTER2].set_router_selection_jitter(1)
self.nodes[ROUTER3].set_panid(0xface)
self.nodes[ROUTER3].set_panid()
self.nodes[ROUTER3].set_mode('rsdn')
self.nodes[ROUTER3].add_whitelist(self.nodes[LEADER].get_addr64())
self.nodes[ROUTER3].add_whitelist(self.nodes[ROUTER2].get_addr64())
self.nodes[ROUTER3].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64())
self.nodes[ROUTER3].enable_whitelist()
self.nodes[ROUTER3].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].set_timeout(3)
self.nodes[ED2].enable_whitelist()
self.nodes[MED1].set_panid()
self.nodes[MED1].set_mode('rsn')
self.nodes[MED1].add_whitelist(self.nodes[DUT_ROUTER2].get_addr64())
self.nodes[MED1].set_timeout(MED1_TIMEOUT)
self.nodes[MED1].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
self.nodes[LEADER].start()
self.nodes[LEADER].set_state('leader')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[BR].start()
time.sleep(5)
self.assertEqual(self.nodes[BR].get_state(), 'router')
self.nodes[BR].add_prefix('2001:2:0:3::/64', 'paros')
self.nodes[BR].add_prefix('2001:2:0:4::/64', 'paros')
self.nodes[BR].register_netdata()
self.nodes[ROUTER2].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.nodes[ROUTER1].start()
self.nodes[DUT_ROUTER2].start()
self.nodes[ROUTER3].start()
self.nodes[MED1].start()
time.sleep(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.assertEqual(self.nodes[DUT_ROUTER2].get_state(), 'router')
self.assertEqual(self.nodes[ROUTER3].get_state(), 'router')
self.assertEqual(self.nodes[MED1].get_state(), 'child')
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
self.nodes[ED2].start()
time.sleep(5)
self.assertEqual(self.nodes[ED2].get_state(), 'child')
# 2
router3_mleid = self.nodes[ROUTER3].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
self.assertTrue(self.nodes[MED1].ping(router3_mleid))
addrs = self.nodes[ROUTER3].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[ED2].ping(addr))
time.sleep(1)
# Verify DUT_ROUTER2 sent an Address Query Request to the Realm local address.
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
msg = dut_messages.next_coap_message('0.02', '/a/aq')
command.check_address_query(msg, self.nodes[DUT_ROUTER2], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS)
addrs = self.nodes[ED2].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[LEADER].ping(addr))
time.sleep(1)
# 3
med1_mleid = self.nodes[MED1].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
self.assertTrue(self.nodes[ROUTER1].ping(med1_mleid))
addrs = self.nodes[BR].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[ED2].ping(addr))
time.sleep(1)
# Verify DUT_ROUTER2 responded with an Address Notification.
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
msg = dut_messages.next_coap_message('0.02', '/a/an')
command.check_address_notification(msg, self.nodes[DUT_ROUTER2], self.nodes[ROUTER1])
addrs = self.nodes[ROUTER3].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[ED2].ping(addr))
time.sleep(1)
# 4
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
self.assertTrue(self.nodes[MED1].ping(router3_mleid))
addrs = self.nodes[ROUTER3].get_addrs()
# Verify DUT_ROUTER2 didn't send an Address Query Request.
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
msg = dut_messages.next_coap_message('0.02', '/a/aq', False)
assert msg is None, "The Address Query Request is not expected."
# 5
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
self.nodes[ROUTER3].stop()
time.sleep(140)
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertFalse(self.nodes[ED2].ping(addr))
# Wait for the Leader to expire its Router ID.
# MAX_NEIGHBOR_AGE + INFINITE_COST_TIMEOUT + ID_REUSE_DELAY + propagation time + transmission time ~ 550s.
time.sleep(550)
self.assertFalse(self.nodes[MED1].ping(router3_mleid))
addrs = self.nodes[ED2].get_addrs()
self.nodes[ED2].stop()
time.sleep(10)
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertFalse(self.nodes[BR].ping(addr))
# Verify DUT_ROUTER2 sent an Address Query Request to the Realm local address.
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
msg = dut_messages.next_coap_message('0.02', '/a/aq')
command.check_address_query(msg, self.nodes[DUT_ROUTER2], config.REALM_LOCAL_ALL_ROUTERS_ADDRESS)
# 6
self.nodes[MED1].stop()
time.sleep(MED1_TIMEOUT)
self.assertFalse(self.nodes[ROUTER1].ping(med1_mleid))
self.assertFalse(self.nodes[ROUTER1].ping(med1_mleid))
# Verify DUT_ROUTER2 didn't respond with an Address Notification.
dut_messages = self.sniffer.get_messages_sent_by(DUT_ROUTER2)
msg = dut_messages.next_coap_message('0.02', '/a/an', False)
assert msg is None, "The Address Notification is not expected."
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/command.py
+34 -3
View File
@@ -2,6 +2,17 @@ import ipv6
import network_layer
import config
import mle
def check_address_query(command_msg, source_node, destination_address):
"""Verify source_node sent a properly formatted Address Query Request message to the destination_address.
"""
command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid)
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."
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."
def check_address_notification(command_msg, source_node, destination_node):
"""Verify source_node sent a properly formatted Address Notification command message to destination_node.
@@ -12,10 +23,10 @@ def check_address_notification(command_msg, source_node, destination_node):
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."
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."
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.
@@ -25,4 +36,24 @@ def check_address_release(command_msg, destination_node):
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"
assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is not RLOC address"
def check_link_request(command_msg):
"""Verify a properly formatted Link Request command message.
"""
command_msg.assertMleMessageContainsTlv(mle.Challenge)
command_msg.assertMleMessageContainsTlv(mle.LeaderData)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.Version)
def check_link_accept(command_msg, destination_node):
"""Verify a properly formatted Link Accept command message.
"""
command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
command_msg.assertMleMessageContainsTlv(mle.SourceAddress)
command_msg.assertMleMessageContainsTlv(mle.Response)
command_msg.assertMleMessageContainsTlv(mle.Version)
command_msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter)
destination_link_local = destination_node.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL)
assert ipv6.ip_address(destination_link_local) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The destination is unexpected"
tests/scripts/thread-cert/config.py
+3
View File
@@ -46,6 +46,9 @@ 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..$'
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'
DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff])
tests/scripts/thread-cert/message.py
+8
View File
@@ -377,6 +377,14 @@ class MessagesSet(object):
return message
def contains_icmp_message(self):
for m in self.messages:
if m.type == MessageType.ICMP:
return True
return False
def contains_mle_message(self, command_type):
for m in self.messages:
if m.type != MessageType.MLE:
tests/scripts/thread-cert/node_cli.py
+14 -9
View File
@@ -427,14 +427,20 @@ class otCli:
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)
if address_type == config.ADDRESS_TYPE.LINK_LOCAL:
return self.__getLinkLocalAddress()
elif address_type == config.ADDRESS_TYPE.GLOBAL:
return self.__getGlobalAddress()
elif address_type == config.ADDRESS_TYPE.RLOC:
return self.__getRloc()
elif address_type == config.ADDRESS_TYPE.ALOC:
return self.__getAloc()
elif address_type == config.ADDRESS_TYPE.ML_EID:
return self.__getMleid()
else:
return None
return None
def get_context_reuse_delay(self):
self.send_command('contextreusedelay')
@@ -623,7 +629,6 @@ class otCli:
def send_mgmt_pending_set(self, pending_timestamp=None, active_timestamp=None, delay_timer=None, channel=None,
panid=None, master_key=None, mesh_local=None, network_name=None):
cmd = 'dataset mgmtsetcommand pending '
if pending_timestamp != None:
cmd += 'pendingtimestamp %d ' % pending_timestamp