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

[thread-cert] refactor case 9.2.12 and 9.2.13 using pktverify (#5442)

Browse Source
This commit is contained in:
Jing Ma
2020-10-14 12:23:55 -07:00
committed by GitHub
parent 946a90520c
commit c2903713a1
4 changed files with 135 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_9_2_12_Announce.py
+52 -19
View File
@@ -30,12 +30,13 @@
import unittest
import thread_cert
from pktverify.consts import MLE_CHILD_ID_REQUEST, MLE_PARENT_REQUEST, MLE_CHILD_ID_RESPONSE, MLE_ANNOUNCE, CHANNEL_TLV, PAN_ID_TLV, ACTIVE_TIMESTAMP_TLV, LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS
from pktverify.packet_verifier import PacketVerifier
LEADER1 = 1
ROUTER1 = 2
LEADER2 = 3
ROUTER2 = 4
MED = 5
MED = 4
DATASET1_TIMESTAMP = 20
DATASET1_CHANNEL = 11
@@ -51,6 +52,7 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
TOPOLOGY = {
LEADER1: {
'name': 'LEADER_1',
'active_dataset': {
'timestamp': DATASET1_TIMESTAMP,
'panid': DATASET1_PANID,
@@ -60,6 +62,7 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
'allowlist': [ROUTER1]
},
ROUTER1: {
'name': 'ROUTER_1',
'active_dataset': {
'timestamp': DATASET1_TIMESTAMP,
'panid': DATASET1_PANID,
@@ -70,6 +73,7 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
'allowlist': [LEADER1, LEADER2]
},
LEADER2: {
'name': 'LEADER_2',
'active_dataset': {
'timestamp': DATASET2_TIMESTAMP,
'panid': DATASET2_PANID,
@@ -77,24 +81,15 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
},
'mode': 'rdn',
'router_selection_jitter': 1,
'allowlist': [ROUTER1, ROUTER2]
},
ROUTER2: {
'active_dataset': {
'timestamp': DATASET2_TIMESTAMP,
'panid': DATASET2_PANID,
'channel': DATASET2_CHANNEL
},
'mode': 'rdn',
'router_selection_jitter': 1,
'allowlist': [LEADER2, MED]
'allowlist': [MED, ROUTER1]
},
MED: {
'name': 'MED',
'channel': DATASET2_CHANNEL,
'is_mtd': True,
'mode': 'rn',
'panid': DATASET2_PANID,
'allowlist': [ROUTER2]
'allowlist': [LEADER2]
},
}
@@ -113,10 +108,6 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
self.nodes[LEADER2].set_state('leader')
self.assertEqual(self.nodes[LEADER2].get_state(), 'leader')
self.nodes[ROUTER2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.nodes[MED].start()
self.simulator.go(5)
self.assertEqual(self.nodes[MED].get_state(), 'child')
@@ -129,14 +120,56 @@ class Cert_9_2_12_Announce(thread_cert.TestCase):
self.nodes[LEADER1].announce_begin(0x1000, 1, 1000, ipaddr)
self.simulator.go(30)
self.assertEqual(self.nodes[LEADER2].get_state(), 'router')
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.assertEqual(self.nodes[MED].get_state(), 'child')
self.collect_rlocs()
ipaddrs = self.nodes[MED].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
self.assertTrue(self.nodes[LEADER1].ping(ipaddr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER_1 = pv.vars['LEADER_1']
ROUTER_1 = pv.vars['ROUTER_1']
LEADER_2 = pv.vars['LEADER_2']
LEADER_1_RLOC = pv.vars['LEADER_1_RLOC']
MED = pv.vars['MED']
MED_RLOC = pv.vars['MED_RLOC']
# Step 1: Ensure the topology is formed correctly
pkts.filter_wpan_src64(LEADER_1).filter_wpan_dst64(ROUTER_1).filter_mle_cmd(
MLE_CHILD_ID_RESPONSE).must_next().must_verify(lambda p: p.wpan.dst_pan == DATASET1_PANID)
pkts.copy().filter_wpan_src64(LEADER_2).filter_wpan_dst64(MED).filter_mle_cmd(
MLE_CHILD_ID_RESPONSE).must_next().must_verify(lambda p: p.wpan.dst_pan == DATASET2_PANID)
# Step 4: Leader_2 MUST send a MLE Child ID Request on its new channel to Router_1
# LEADER_2 MUST send a MLE Announce Message
# The Destination PAN ID (0xFFFF) in the IEEE 802.15.4 MAC and MUST be secured using Key ID Mode 2.
pkts.filter_wpan_src64(LEADER_2).filter_wpan_dst64(ROUTER_1).filter_mle_cmd(
MLE_CHILD_ID_REQUEST).must_next().must_verify(lambda p: p.wpan.dst_pan == DATASET1_PANID)
pkts.filter_wpan_src64(LEADER_2).filter_ipv6_dst(LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS).filter_mle_cmd(
MLE_ANNOUNCE).must_next().must_verify(
lambda p: {CHANNEL_TLV, PAN_ID_TLV, ACTIVE_TIMESTAMP_TLV} == set(p.mle.tlv.type) and p.wpan.dst_pan ==
0xffff and p.wpan.aux_sec.key_id_mode == 0x2 and p.wpan.aux_sec.key_source == 0x00000000ffffffff)
# Step 5: MED MUST send a MLE Child ID Request on its new channel
# MED MUST send a MLE Announce Message
# The Destination PAN ID (0xFFFF) in the IEEE 802.15.4 MAC and MUST be secured using Key ID Mode 2.
pkts.filter_wpan_src64(MED).filter_wpan_dst64(LEADER_2).filter_mle_cmd(
MLE_CHILD_ID_REQUEST).must_next().must_verify(lambda p: p.wpan.dst_pan == DATASET1_PANID)
pkts.filter_wpan_src64(MED).filter_ipv6_dst(LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS).filter_mle_cmd(
MLE_ANNOUNCE).must_next().must_verify(
lambda p: {CHANNEL_TLV, PAN_ID_TLV, ACTIVE_TIMESTAMP_TLV} == set(p.mle.tlv.type) and p.wpan.dst_pan ==
0xffff and p.wpan.aux_sec.key_id_mode == 0x2 and p.wpan.aux_sec.key_source == 0x00000000ffffffff)
# Step 6: MED MUST respond with an ICMPv6 Echo Reply
pkts.filter_ping_reply().filter_ipv6_src_dst(MED_RLOC, LEADER_1_RLOC).must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Cert_9_2_13_EnergyScan.py
+80 -22
View File
@@ -27,40 +27,45 @@
# POSSIBILITY OF SUCH DAMAGE.
#
import copy
import unittest
import config
import thread_cert
from pktverify.consts import MLE_CHILD_ID_RESPONSE, MGMT_ED_SCAN, MGMT_ED_REPORT, NM_CHANNEL_MASK_TLV, NM_ENERGY_LIST_TLV
from pktverify.packet_verifier import PacketVerifier
COMMISSIONER = 1
LEADER = 2
ROUTER1 = 3
ED1 = 4
ED = 4
class Cert_9_2_13_EnergyScan(thread_cert.TestCase):
class Cert_9_2_13_EnergyScan_Base(thread_cert.TestCase):
SUPPORT_NCP = False
TOPOLOGY = {
COMMISSIONER: {
'name': 'COMMISSIONER',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER]
},
LEADER: {
'name': 'LEADER',
'mode': 'rdn',
'panid': 0xface,
'allowlist': [COMMISSIONER, ROUTER1]
},
ROUTER1: {
'name': 'ROUTER',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER, ED1]
'allowlist': [LEADER, ED]
},
ED1: {
'is_mtd': True,
'mode': 'r',
ED: {
'panid': 0xface,
'allowlist': [ROUTER1]
},
@@ -75,36 +80,89 @@ class Cert_9_2_13_EnergyScan(thread_cert.TestCase):
self.simulator.go(5)
self.assertEqual(self.nodes[COMMISSIONER].get_state(), 'router')
self.nodes[COMMISSIONER].commissioner_start()
self.simulator.go(3)
self.simulator.go(5)
self.nodes[ROUTER1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.nodes[ED1].start()
self.nodes[ED].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ED1].get_state(), 'child')
self.assertEqual(self.nodes[ED].get_state(), 'child')
self.collect_rlocs()
if self.TOPOLOGY[ED]['name'] == 'DUT':
ipaddrs = self.nodes[ED].get_addrs()
else:
ipaddrs = self.nodes[ROUTER1].get_addrs()
ipaddrs = self.nodes[ROUTER1].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
break
self.assertTrue(self.nodes[COMMISSIONER].ping(ipaddr))
self.nodes[COMMISSIONER].energy_scan(0x50000, 0x02, 0x20, 0x3E8, ipaddr)
ipaddrs = self.nodes[ED1].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
break
self.assertTrue(self.nodes[COMMISSIONER].ping(ipaddr))
self.nodes[COMMISSIONER].energy_scan(0x50000, 0x02, 0x20, 0x3E8, ipaddr)
self.nodes[COMMISSIONER].energy_scan(0x50000, 0x02, 0x20, 0x3E8, 'ff33:0040:fd00:db8:0:0:0:1')
self.nodes[COMMISSIONER].energy_scan(0x50000, 0x02, 0x20, 0xc8, ipaddr)
self.simulator.go(3)
self.nodes[COMMISSIONER].energy_scan(0x50000, 0x02, 0x20, 0xc8, 'ff33:0040:fd00:db8:0:0:0:1')
self.simulator.go(3)
self.assertTrue(self.nodes[COMMISSIONER].ping(ipaddr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
DUT = pv.vars['DUT']
DUT_RLOC = pv.vars['DUT_RLOC']
COMMISSIONER_RLOC = pv.vars['COMMISSIONER_RLOC']
_pkts = pkts.filter_wpan_src64(DUT)
# Step 3: The DUT MUST send MGMT_ED_REPORT.ans to the Commissioner and report energy measurements
_pkts.filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_request(MGMT_ED_REPORT).must_next().must_verify(
lambda p: {NM_CHANNEL_MASK_TLV, NM_ENERGY_LIST_TLV} == set(p.thread_meshcop.tlv.type) and p.thread_meshcop.
tlv.chan_mask_mask == '0000a000' and len(p.thread_meshcop.tlv.energy_list) == 2)
# Step 5: The DUT MUST send MGMT_ED_REPORT.ans to the Commissioner and report energy measurements
_pkts.filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_request(MGMT_ED_REPORT).must_next().must_verify(
lambda p: {NM_CHANNEL_MASK_TLV, NM_ENERGY_LIST_TLV} == set(p.thread_meshcop.tlv.type) and p.thread_meshcop.
tlv.chan_mask_mask == '0000a000' and len(p.thread_meshcop.tlv.energy_list) == 2)
# Step 6: The DUT MUST respond with ICMPv6 Echo Reply
_pkts.filter_ping_reply().filter_ipv6_src_dst(DUT_RLOC, COMMISSIONER_RLOC).must_next()
class Cert_9_2_13_EnergyScan_FED(Cert_9_2_13_EnergyScan_Base):
TOPOLOGY = copy.deepcopy(Cert_9_2_13_EnergyScan_Base.TOPOLOGY)
TOPOLOGY[ROUTER1]['name'] = 'ROUTER'
TOPOLOGY[ED]['name'] = 'DUT'
TOPOLOGY[ED]['router_upgrade_threshold'] = 0
class Cert_9_2_13_EnergyScan_MED(Cert_9_2_13_EnergyScan_Base):
TOPOLOGY = copy.deepcopy(Cert_9_2_13_EnergyScan_Base.TOPOLOGY)
TOPOLOGY[ROUTER1]['name'] = 'ROUTER'
TOPOLOGY[ED]['name'] = 'DUT'
TOPOLOGY[ED]['mode'] = 'rn'
TOPOLOGY[ED]['is_mtd'] = True
class Cert_9_2_13_EnergyScan_ROUTER(Cert_9_2_13_EnergyScan_Base):
TOPOLOGY = copy.deepcopy(Cert_9_2_13_EnergyScan_Base.TOPOLOGY)
TOPOLOGY[ROUTER1]['name'] = 'DUT'
TOPOLOGY[ED]['name'] = 'ED'
TOPOLOGY[ED]['mode'] = 'rn'
TOPOLOGY[ED]['is_mtd'] = True
class Cert_9_2_13_EnergyScan_SED(Cert_9_2_13_EnergyScan_Base):
TOPOLOGY = copy.deepcopy(Cert_9_2_13_EnergyScan_Base.TOPOLOGY)
TOPOLOGY[ROUTER1]['name'] = 'ROUTER'
TOPOLOGY[ED]['name'] = 'DUT'
TOPOLOGY[ED]['mode'] = '-'
TOPOLOGY[ED]['is_mtd'] = True
TOPOLOGY[ED]['timeout'] = config.DEFAULT_CHILD_TIMEOUT
del (Cert_9_2_13_EnergyScan_Base)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/pktverify/consts.py
+2
View File
@@ -88,6 +88,7 @@ RLY_RX_URI = '/c/rx'
RLY_TX_URI = '/c/tx'
MGMT_PANID_QUERY = '/c/pq'
MGMT_PANID_CONFLICT = '/c/pc'
MGMT_ED_SCAN = '/c/es'
MGMT_ED_REPORT = '/c/er'
MGMT_ACTIVE_GET_URI = '/c/ag'
MGMT_ACTIVE_SET_URI = '/c/as'
@@ -162,6 +163,7 @@ NM_COMMISSIONER_UDP_PORT_TLV = 15
NM_JOINER_UDP_PORT_TLV = 18
NM_DELAY_TIMER_TLV = 52
NM_CHANNEL_MASK_TLV = 53
NM_ENERGY_LIST_TLV = 57
NM_DISCOVERY_RESPONSE_TLV = 129
# DTLS
tests/scripts/thread-cert/pktverify/layer_fields.py
+1
View File
@@ -531,6 +531,7 @@ _LAYER_FIELDS = {
'thread_meshcop.tlv.chan_mask_page': _auto,
'thread_meshcop.tlv.chan_mask_len': _auto,
'thread_meshcop.tlv.chan_mask_mask': _bytes,
'thread_meshcop.tlv.energy_list': _list(_auto),
'thread_meshcop.tlv.pan_id': _auto,
'thread_meshcop.tlv.xpan_id': _bytes,
'thread_meshcop.tlv.ml_prefix': _bytes,