espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-06-06 05:24:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

[thread-cert] refactor case 9.2.7 using pktverify (#5459)

Browse Source
This commit is contained in:
Jing Ma
2020-10-12 07:26:13 +08:00
committed by GitHub
parent d2d4df6260
commit f01f4d96a3
6 changed files with 137 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_9_2_07_DelayTimer.py
+115 -22
View File
@@ -30,6 +30,8 @@
import unittest
import thread_cert
from pktverify.consts import MLE_CHILD_ID_RESPONSE, MLE_DATA_RESPONSE, MGMT_PENDING_SET_URI, MGMT_ACTIVE_SET_URI, MGMT_DATASET_CHANGED_URI, COAP_CODE_ACK, ACTIVE_OPERATION_DATASET_TLV, ACTIVE_TIMESTAMP_TLV, PENDING_TIMESTAMP_TLV, NM_CHANNEL_TLV, NM_CHANNEL_MASK_TLV, NM_EXTENDED_PAN_ID_TLV, NM_NETWORK_MASTER_KEY_TLV, NM_NETWORK_MESH_LOCAL_PREFIX_TLV, NM_NETWORK_NAME_TLV, NM_PAN_ID_TLV, NM_PSKC_TLV, NM_SECURITY_POLICY_TLV, SOURCE_ADDRESS_TLV, LEADER_DATA_TLV, ACTIVE_TIMESTAMP_TLV, NETWORK_DATA_TLV, NM_BORDER_AGENT_LOCATOR_TLV, NM_COMMISSIONER_SESSION_ID_TLV, NM_DELAY_TIMER_TLV, PENDING_OPERATION_DATASET_TLV
from pktverify.packet_verifier import PacketVerifier
PANID_INIT = 0xface
@@ -41,7 +43,11 @@ LEADER_ACTIVE_TIMESTAMP = 10
ROUTER_ACTIVE_TIMESTAMP = 20
ROUTER_PENDING_TIMESTAMP = 30
ROUTER_PENDING_ACTIVE_TIMESTAMP = 25
ROUTER_DELAY_TIMER = 3600000
COMMISSIONER_PENDING_TIMESTAMP = 40
COMMISSIONER_PENDING_ACTIVE_TIMESTAMP = 80
COMMISSIONER_DELAY_TIMER = 60000
COMMISSIONER_PENDING_CHANNEL = 20
COMMISSIONER_PENDING_PANID = 0xafce
@@ -51,15 +57,14 @@ class Cert_9_2_7_DelayTimer(thread_cert.TestCase):
TOPOLOGY = {
COMMISSIONER: {
'active_dataset': {
'timestamp': LEADER_ACTIVE_TIMESTAMP
},
'name': 'COMMISSIONER',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER]
},
LEADER: {
'name': 'LEADER',
'mode': 'rdn',
'panid': 0xface,
'partition_id': 0xffffffff,
@@ -67,16 +72,10 @@ class Cert_9_2_7_DelayTimer(thread_cert.TestCase):
'allowlist': [COMMISSIONER]
},
ROUTER: {
'active_dataset': {
'timestamp': ROUTER_ACTIVE_TIMESTAMP
},
'name': 'ROUTER',
'mode': 'rdn',
'panid': 0xface,
'partition_id': 1,
'pending_dataset': {
'pendingtimestamp': ROUTER_PENDING_TIMESTAMP,
'activetimestamp': ROUTER_PENDING_ACTIVE_TIMESTAMP
},
'router_selection_jitter': 1
},
}
@@ -92,32 +91,41 @@ class Cert_9_2_7_DelayTimer(thread_cert.TestCase):
self.nodes[COMMISSIONER].commissioner_start()
self.simulator.go(3)
self.nodes[COMMISSIONER].send_mgmt_active_set(active_timestamp=LEADER_ACTIVE_TIMESTAMP,)
self.simulator.go(5)
self.nodes[ROUTER].start()
self.simulator.go(10)
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER].get_state(), 'leader')
self.nodes[LEADER].add_allowlist(self.nodes[ROUTER].get_addr64())
self.nodes[ROUTER].add_allowlist(self.nodes[LEADER].get_addr64())
self.simulator.go(30)
self.simulator.go(20)
self.assertEqual(self.nodes[COMMISSIONER].get_state(), 'router')
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
ipaddrs = self.nodes[ROUTER].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
break
self.assertTrue(self.nodes[LEADER].ping(ipaddr))
self.nodes[ROUTER].commissioner_start()
self.simulator.go(3)
self.nodes[ROUTER].send_mgmt_active_set(active_timestamp=ROUTER_ACTIVE_TIMESTAMP,)
self.simulator.go(30)
self.nodes[ROUTER].send_mgmt_pending_set(
pending_timestamp=ROUTER_PENDING_TIMESTAMP,
active_timestamp=ROUTER_PENDING_ACTIVE_TIMESTAMP,
delay_timer=ROUTER_DELAY_TIMER,
)
self.simulator.go(60)
self.nodes[COMMISSIONER].send_mgmt_pending_set(
pending_timestamp=40,
active_timestamp=80,
delay_timer=10000,
pending_timestamp=COMMISSIONER_PENDING_TIMESTAMP,
active_timestamp=COMMISSIONER_PENDING_ACTIVE_TIMESTAMP,
delay_timer=COMMISSIONER_DELAY_TIMER,
channel=COMMISSIONER_PENDING_CHANNEL,
panid=COMMISSIONER_PENDING_PANID,
)
self.simulator.go(40)
self.simulator.go(120)
self.assertEqual(self.nodes[LEADER].get_panid(), COMMISSIONER_PENDING_PANID)
self.assertEqual(self.nodes[COMMISSIONER].get_panid(), COMMISSIONER_PENDING_PANID)
self.assertEqual(self.nodes[ROUTER].get_panid(), COMMISSIONER_PENDING_PANID)
@@ -129,12 +137,97 @@ class Cert_9_2_7_DelayTimer(thread_cert.TestCase):
)
self.assertEqual(self.nodes[ROUTER].get_channel(), COMMISSIONER_PENDING_CHANNEL)
self.collect_rloc16s()
self.collect_rlocs()
ipaddrs = self.nodes[ROUTER].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
break
self.assertTrue(self.nodes[LEADER].ping(ipaddr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
LEADER_RLOC16 = pv.vars['LEADER_RLOC16']
COMMISSIONER = pv.vars['COMMISSIONER']
COMMISSIONER_RLOC = pv.vars['COMMISSIONER_RLOC']
ROUTER = pv.vars['ROUTER']
ROUTER_RLOC = pv.vars['ROUTER_RLOC']
ROUTER_RLOC16 = pv.vars['ROUTER_RLOC16']
_lpkts = pkts.filter_wpan_src64(LEADER)
# Step 1: Ensure the topology is formed correctly
_lpkts.filter_wpan_dst64(COMMISSIONER).filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next()
_lpkts_coap = _lpkts.copy()
# Step 4: Leader MUST send a unicast MLE Child ID Response to the Router
_lpkts.filter_wpan_dst64(ROUTER).filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next(
).must_verify(lambda p: {ACTIVE_OPERATION_DATASET_TLV, ACTIVE_TIMESTAMP_TLV} < set(p.mle.tlv.type) and {
NM_CHANNEL_TLV, NM_CHANNEL_MASK_TLV, NM_EXTENDED_PAN_ID_TLV, NM_NETWORK_MASTER_KEY_TLV,
NM_NETWORK_MESH_LOCAL_PREFIX_TLV, NM_NETWORK_NAME_TLV, NM_PAN_ID_TLV, NM_PSKC_TLV, NM_SECURITY_POLICY_TLV
} <= set(p.thread_meshcop.tlv.type) and p.mle.tlv.active_tstamp == LEADER_ACTIVE_TIMESTAMP)
# Step 6: Leader automatically sends a MGMT_ACTIVE_SET.rsp to the Router
_lpkts.filter_ipv6_dst(ROUTER_RLOC).filter_coap_ack(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.coap.code == COAP_CODE_ACK and p.thread_meshcop.tlv.state == 1)
# Step 7: Leader multicasts a MLE Data Response with the new information
_lpkts.filter_LLANMA().filter_mle_cmd(MLE_DATA_RESPONSE).must_next().must_verify(
lambda p: {SOURCE_ADDRESS_TLV, LEADER_DATA_TLV, ACTIVE_TIMESTAMP_TLV, NETWORK_DATA_TLV} <= set(
p.mle.tlv.type) and {NM_BORDER_AGENT_LOCATOR_TLV, NM_COMMISSIONER_SESSION_ID_TLV} <= set(
p.thread_meshcop.tlv.type) and p.thread_nwd.tlv.stable == [0] and p.mle.tlv.active_tstamp ==
ROUTER_ACTIVE_TIMESTAMP)
# Step 10: Leader MUST send a unicast MLE Data Response to the Router
_lpkts.filter_wpan_dst64(ROUTER).filter_mle_cmd(MLE_DATA_RESPONSE).must_next(
).must_verify(lambda p: {ACTIVE_OPERATION_DATASET_TLV, ACTIVE_TIMESTAMP_TLV} < set(p.mle.tlv.type) and {
NM_CHANNEL_TLV, NM_CHANNEL_MASK_TLV, NM_EXTENDED_PAN_ID_TLV, NM_NETWORK_MASTER_KEY_TLV,
NM_NETWORK_MESH_LOCAL_PREFIX_TLV, NM_NETWORK_NAME_TLV, NM_PAN_ID_TLV, NM_PSKC_TLV, NM_SECURITY_POLICY_TLV
} <= set(p.thread_meshcop.tlv.type) and p.mle.tlv.active_tstamp == ROUTER_ACTIVE_TIMESTAMP)
# Step 12: Leader sends a MGMT_PENDING_SET.rsp to the Router with Status = Accept
_lpkts_coap.filter_ipv6_dst(ROUTER_RLOC).filter_coap_ack(MGMT_PENDING_SET_URI).must_next().must_verify(
lambda p: p.coap.code == COAP_CODE_ACK and p.thread_meshcop.tlv.state == 1)
# Step 13: Leader sends a multicast MLE Data Response
_lpkts.filter_LLANMA().filter_mle_cmd(MLE_DATA_RESPONSE).must_next().must_verify(
lambda p: {
SOURCE_ADDRESS_TLV, LEADER_DATA_TLV, ACTIVE_TIMESTAMP_TLV, PENDING_TIMESTAMP_TLV, NETWORK_DATA_TLV
} <= set(p.mle.tlv.type) and p.thread_nwd.tlv.stable == [0] and p.mle.tlv.active_tstamp ==
ROUTER_ACTIVE_TIMESTAMP and p.mle.tlv.pending_tstamp == ROUTER_PENDING_TIMESTAMP)
# Step 14: The DUT MUST send MGMT_DATASET_CHANGED.ntf to the Router
_lpkts_coap.filter_wpan_dst16(ROUTER_RLOC16).filter_coap_request(MGMT_DATASET_CHANGED_URI).must_next()
# Step 16: Leader MUST send a unicast MLE Data Response to the Router
_lpkts.filter_wpan_dst64(ROUTER).filter_mle_cmd(MLE_DATA_RESPONSE).must_next().must_verify(
lambda p: {ACTIVE_TIMESTAMP_TLV, PENDING_TIMESTAMP_TLV} < set(p.mle.tlv.type) and p.mle.tlv.active_tstamp
== ROUTER_ACTIVE_TIMESTAMP and p.mle.tlv.pending_tstamp == ROUTER_PENDING_TIMESTAMP)
# Step 18: The DUT MUST send MGMT_PENDING_SET.rsp to the Commissioner
_lpkts_coap.filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(MGMT_PENDING_SET_URI).must_next().must_verify(
lambda p: p.coap.code == COAP_CODE_ACK and p.thread_meshcop.tlv.state == 1)
# Step 19: Leader MUST send a unicast MLE Data Response to the Router
_lpkts.filter_LLANMA().filter_mle_cmd(MLE_DATA_RESPONSE).must_next().must_verify(
lambda p: {
SOURCE_ADDRESS_TLV, LEADER_DATA_TLV, NETWORK_DATA_TLV, ACTIVE_TIMESTAMP_TLV, PENDING_TIMESTAMP_TLV
} <= set(p.mle.tlv.type) and p.thread_nwd.tlv.stable == [0] and p.mle.tlv.active_tstamp ==
ROUTER_ACTIVE_TIMESTAMP and p.mle.tlv.pending_tstamp == COMMISSIONER_PENDING_TIMESTAMP)
# Step 20: Leader MUST send a unicast MLE Data Response to the Router
_lpkts.filter_wpan_dst64(ROUTER).filter_mle_cmd(MLE_DATA_RESPONSE).must_next(
).must_verify(lambda p: {ACTIVE_TIMESTAMP_TLV, PENDING_TIMESTAMP_TLV, PENDING_OPERATION_DATASET_TLV} < set(
p.mle.tlv.type) and {NM_CHANNEL_TLV, NM_COMMISSIONER_SESSION_ID_TLV, NM_PAN_ID_TLV, NM_DELAY_TIMER_TLV} <=
set(p.thread_meshcop.tlv.type) and p.mle.tlv.active_tstamp == ROUTER_ACTIVE_TIMESTAMP and p.mle.
tlv.pending_tstamp == COMMISSIONER_PENDING_TIMESTAMP and p.thread_meshcop.tlv.pan_id ==
COMMISSIONER_PENDING_PANID and p.thread_meshcop.tlv.channel == COMMISSIONER_PENDING_CHANNEL)
# Step 21: Router MUST respond with an ICMPv6 Echo Reply
pkts.filter_wpan_src16_dst16(ROUTER_RLOC16, LEADER_RLOC16).filter_ping_reply().must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/node.py
+6 -1
View File
@@ -1392,7 +1392,7 @@ class NodeImpl:
self.send_command('dataset commit active')
self._expect('Done')
def set_pending_dataset(self, pendingtimestamp, activetimestamp, panid=None, channel=None):
def set_pending_dataset(self, pendingtimestamp, activetimestamp, panid=None, channel=None, delay=None):
self.send_command('dataset clear')
self._expect('Done')
@@ -1414,6 +1414,11 @@ class NodeImpl:
self.send_command(cmd)
self._expect('Done')
if delay is not None:
cmd = 'dataset delay %d' % delay
self.send_command(cmd)
self._expect('Done')
# Set the meshlocal prefix in config.py
self.send_command('dataset meshlocalprefix %s' % config.MESH_LOCAL_PREFIX.split('/')[0])
self._expect('Done')
tests/scripts/thread-cert/pktverify/consts.py
+4 -3
View File
@@ -86,12 +86,13 @@ SVR_DATA_URI = '/a/sd'
ND_DATA_URI = '/a/nd'
RLY_RX_URI = '/c/rx'
RLY_TX_URI = '/c/tx'
MGMT_ACTIVE_SET_URI = '/c/as'
MGMT_ACTIVE_GET_URI = '/c/ag'
MGMT_PENDING_SET_URI = '/c/ps'
MGMT_PANID_QUERY = '/c/pq'
MGMT_PANID_CONFLICT = '/c/pc'
MGMT_ED_REPORT = '/c/er'
MGMT_ACTIVE_GET_URI = '/c/ag'
MGMT_ACTIVE_SET_URI = '/c/as'
MGMT_PENDING_SET_URI = '/c/ps'
MGMT_DATASET_CHANGED_URI = '/c/dc'
# MLE TLVs
SOURCE_ADDRESS_TLV = 0
tests/scripts/thread-cert/pktverify/layer_fields.py
+1
View File
@@ -265,6 +265,7 @@ _LAYER_FIELDS = {
'mle.tlv.version': _auto,
'mle.tlv.source_addr': _auto,
'mle.tlv.active_tstamp': _auto,
'mle.tlv.pending_tstamp': _auto,
'mle.tlv.leader_data.partition_id': _auto,
'mle.tlv.leader_data.weighting': _auto,
'mle.tlv.leader_data.data_version': _auto,
tests/scripts/thread-cert/pktverify/packet_filter.py
+9
View File
@@ -415,6 +415,15 @@ class PacketFilter(object):
"""
return self.filter(lambda p: p.wpan.channel == channel, **kwargs)
def filter_wpan_src16(self, addr, **kwargs):
return self.filter(lambda p: p.wpan.src16 == addr, **kwargs)
def filter_wpan_dst16(self, addr, **kwargs):
return self.filter(lambda p: p.wpan.dst16 == addr, **kwargs)
def filter_wpan_src16_dst16(self, src_addr, dst_addr, **kwargs):
return self.filter(lambda p: p.wpan.src16 == src_addr and p.wpan.dst16 == dst_addr, **kwargs)
def filter_wpan_src64(self, addr, **kwargs):
assert isinstance(addr, (str, ExtAddr)), addr
return self.filter(lambda p: p.wpan.src64 == addr, **kwargs)
tests/scripts/thread-cert/thread_cert.py
+2 -1
View File
@@ -191,7 +191,8 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
self.nodes[i].set_pending_dataset(params['pending_dataset']['pendingtimestamp'],
params['pending_dataset']['activetimestamp'],
panid=params['pending_dataset'].get('panid'),
channel=params['pending_dataset'].get('channel'))
channel=params['pending_dataset'].get('channel'),
delay=params['pending_dataset'].get('delay'))
if 'key_switch_guardtime' in params:
self.nodes[i].set_key_switch_guardtime(params['key_switch_guardtime'])