[thread-cert] refactor case 9.2.14 using pktverify (#5429)

This commit is contained in:
Jing Ma
2020-09-10 08:30:54 -07:00
committed by GitHub
parent 21b49e2e2b
commit 29ace9cf52
2 changed files with 50 additions and 2 deletions
@@ -30,6 +30,8 @@
import unittest
import thread_cert
from pktverify.consts import MLE_CHILD_ID_REQUEST, MGMT_PANID_QUERY, MGMT_PANID_CONFLICT, MGMT_ED_REPORT, NM_COMMISSIONER_SESSION_ID_TLV, NM_CHANNEL_MASK_TLV, NM_PAN_ID_TLV, REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS
from pktverify.packet_verifier import PacketVerifier
COMMISSIONER = 1
LEADER1 = 2
@@ -42,23 +44,27 @@ class Cert_9_2_14_PanIdQuery(thread_cert.TestCase):
TOPOLOGY = {
COMMISSIONER: {
'name': 'COMMISSIONER',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [LEADER1]
},
LEADER1: {
'name': 'LEADER_1',
'mode': 'rsdn',
'panid': 0xface,
'whitelist': [COMMISSIONER, ROUTER1]
},
ROUTER1: {
'name': 'ROUTER',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [LEADER1, LEADER2]
},
LEADER2: {
'name': 'LEADER_2',
'mode': 'rsdn',
'panid': 0xdead,
'whitelist': [ROUTER1]
@@ -84,6 +90,7 @@ class Cert_9_2_14_PanIdQuery(thread_cert.TestCase):
self.simulator.go(5)
self.assertEqual(self.nodes[LEADER2].get_state(), 'leader')
self.collect_rlocs()
ipaddrs = self.nodes[ROUTER1].get_addrs()
for ipaddr in ipaddrs:
if ipaddr[0:4] != 'fe80':
@@ -95,6 +102,44 @@ class Cert_9_2_14_PanIdQuery(thread_cert.TestCase):
self.assertTrue(self.nodes[COMMISSIONER].ping(ipaddr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
ROUTER = pv.vars['ROUTER']
COMMISSIONER = pv.vars['COMMISSIONER']
ROUTER_RLOC = pv.vars['ROUTER_RLOC']
COMMISSIONER_RLOC = pv.vars['COMMISSIONER_RLOC']
_rpkts = pkts.filter_wpan_src64(ROUTER)
_cpkts = pkts.filter_wpan_src64(COMMISSIONER)
# Step 1: Ensure the topology is formed correctly
_rpkts.filter_mle_cmd(MLE_CHILD_ID_REQUEST).must_next()
# Step 2: Commissioner MUST send a unicast MGMT_PANID_QUERY.qry unicast to Router_1
_cpkts.filter_ipv6_dst(ROUTER_RLOC).filter_coap_request(MGMT_PANID_QUERY).must_next().must_verify(
lambda p: {NM_COMMISSIONER_SESSION_ID_TLV, NM_CHANNEL_MASK_TLV, NM_PAN_ID_TLV} <= set(p.thread_meshcop.tlv.
type))
# Step 3: Router MUST send MGMT_ED_REPORT.ans to the Commissioner
_rpkts.range(
_cpkts.index).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_request(MGMT_PANID_CONFLICT).must_next(
).must_verify(lambda p: {NM_CHANNEL_MASK_TLV, NM_PAN_ID_TLV} <= set(p.thread_meshcop.tlv.type))
# Step 4: Commissioner MUST send a multicast MGMT_PANID_QUERY.qry
_cpkts.filter_ipv6_dst(REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS).filter_coap_request(
MGMT_PANID_QUERY).must_next().must_verify(
lambda p: {NM_COMMISSIONER_SESSION_ID_TLV, NM_CHANNEL_MASK_TLV, NM_PAN_ID_TLV} <= set(p.thread_meshcop.
tlv.type))
# Step 5: Router MUST send MGMT_PANID_CONFLICT.ans to the Commissioner
_rpkts.range(
_cpkts.index).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_request(MGMT_PANID_CONFLICT).must_next(
).must_verify(lambda p: {NM_CHANNEL_MASK_TLV, NM_PAN_ID_TLV} <= set(p.thread_meshcop.tlv.type))
# Step 6: Router MUST respond with an ICMPv6 Echo Reply
_rpkts.filter_ipv6_dst(COMMISSIONER_RLOC).filter_ping_reply().must_next()
if __name__ == '__main__':
unittest.main()
@@ -33,8 +33,8 @@ from pktverify.bytes import Bytes
DOMAIN_PREFIX = Bytes('fd00:7d03:7d03:7d03')
BACKBONE_IPV6_PREFIX = Bytes('2001:0db8:0001:0000')
LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff32:40:fdde:ad00:beef:0:0:1')
REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff33:40:fdde:ad00:beef:0:0:1')
LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff32:40:fd00:db8::1')
REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff33:40:fd00:db8::1')
REALM_LOCAL_ALL_ROUTERS_ADDRESS = Ipv6Addr('ff03::2')
LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff02::1')
LINK_LOCAL_ALL_ROUTERS_MULTICAST_ADDRESS = Ipv6Addr('ff02::2')
@@ -89,6 +89,9 @@ 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'
# MLE TLVs
SOURCE_ADDRESS_TLV = 0