[thread-cert] refactor case 5.5.4 using pktverify (#5331)

This commit is contained in:
Jing Ma
2020-08-17 22:37:30 -07:00
committed by GitHub
parent 3f38068a71
commit 962c06f8e7
@@ -30,6 +30,8 @@
import unittest
import thread_cert
from pktverify.consts import MLE_ADVERTISEMENT, MLE_PARENT_REQUEST, MLE_CHILD_ID_RESPONSE, SOURCE_ADDRESS_TLV, ROUTE64_TLV, LEADER_DATA_TLV
from pktverify.packet_verifier import PacketVerifier
LEADER = 1
ROUTER1 = 2
@@ -41,30 +43,35 @@ ROUTER4 = 5
class Cert_5_5_4_SplitMergeRouters(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'name': 'LEADER',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [ROUTER1, ROUTER2]
},
ROUTER1: {
'name': 'ROUTER_1',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [LEADER, ROUTER3]
},
ROUTER2: {
'name': 'ROUTER_2',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [LEADER, ROUTER4]
},
ROUTER3: {
'name': 'ROUTER_3',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
'whitelist': [ROUTER1]
},
ROUTER4: {
'name': 'ROUTER_4',
'mode': 'rsdn',
'panid': 0xface,
'router_selection_jitter': 1,
@@ -108,11 +115,45 @@ class Cert_5_5_4_SplitMergeRouters(thread_cert.TestCase):
self.assertEqual(self.nodes[LEADER].get_state(), 'router')
self.collect_ipaddrs()
addrs = self.nodes[ROUTER4].get_addrs()
for addr in addrs:
if addr[0:4] != 'fe80':
self.assertTrue(self.nodes[ROUTER3].ping(addr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
ROUTER_1 = pv.vars['ROUTER_1']
ROUTER_3 = pv.vars['ROUTER_3']
ROUTER_4 = pv.vars['ROUTER_4']
leader_pkts = pkts.filter_wpan_src64(LEADER)
router1_pkts = pkts.filter_wpan_src64(ROUTER_1)
# Step 2: The Leader MUST send properly formatted MLE Advertisements
router1_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next()
leader_pkts.range(router1_pkts.index).filter_mle_cmd(MLE_ADVERTISEMENT).must_next().must_verify(
lambda p: {SOURCE_ADDRESS_TLV, ROUTE64_TLV, LEADER_DATA_TLV} == set(p.mle.tlv.type))
router1_pkts.filter_mle_cmd(MLE_PARENT_REQUEST).must_next()
lreset_start = router1_pkts.index
router1_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next()
lreset_stop = router1_pkts.index
# Step 3: The Leader MUST stop sending MLE advertisements.
leader_pkts.range(lreset_start, lreset_stop).filter_mle_cmd(MLE_ADVERTISEMENT).must_not_next()
# Step 5: Router_4 MUST send an ICMPv6 Echo Reply to Router_3
router3_mleid = pv.vars['ROUTER_3_MLEID']
router4_mleid = pv.vars['ROUTER_4_MLEID']
pkts.filter_wpan_src64(ROUTER_3).filter(
lambda p: p.ipv6.src == router3_mleid and p.ipv6.dst == router4_mleid).filter_ping_request().must_next()
pkts.filter_wpan_src64(ROUTER_4).filter(
lambda p: p.ipv6.src == router4_mleid and p.ipv6.dst == router3_mleid).filter_ping_reply().must_next()
if __name__ == '__main__':
unittest.main()