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

[thread-cert] refactor case 5.3.6 using pktverify (#5813)

Browse Source
This commit is contained in:
Jing Ma
2020-12-01 21:37:19 -08:00
committed by GitHub
parent 6956d5e0f9
commit 8630547d10
4 changed files with 149 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_3_06_RouterIdMask.py
+123 -45
View File
@@ -29,30 +29,53 @@
import unittest
import command
import config
import mle
import thread_cert
from pktverify.consts import MLE_ADVERTISEMENT, MLE_CHILD_ID_REQUEST, MLE_CHILD_ID_RESPONSE
from pktverify.packet_verifier import PacketVerifier
DUT_LEADER = 1
ROUTER1 = 2
ROUTER2 = 3
# Test Purpose and Description:
# -----------------------------
# The purpose of this test case is to verify that the router ID mask is managed
# correctly, as the connectivity to a router or group of routers is lost and / or
# a new router is added to network.
#
# Test Topology:
# -------------
# Leader(DUT)
# |
# Router_1
# |
# Router_2
#
# DUT Types:
# ----------
# Leader
class Cert_5_3_6_RouterIdMask(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
DUT_LEADER: {
'name': 'LEADER',
'mode': 'rdn',
'panid': 0xface,
'allowlist': [ROUTER1]
},
ROUTER1: {
'name': 'ROUTER_1',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [DUT_LEADER, ROUTER2]
},
ROUTER2: {
'name': 'ROUTER_2',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
@@ -78,7 +101,8 @@ class Cert_5_3_6_RouterIdMask(thread_cert.TestCase):
self.nodes[ROUTER2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
router2_id = self.nodes[ROUTER2].get_router_id()
self.collect_rloc16s()
# Wait DUT_LEADER to establish routing to ROUTER2 via ROUTER1's MLE
# advertisement.
@@ -89,64 +113,118 @@ class Cert_5_3_6_RouterIdMask(thread_cert.TestCase):
self._setUpRouter2()
# 3 & 4
# Flush the message queue to avoid possible impact on follow-up
# verification.
self.simulator.get_messages_sent_by(DUT_LEADER)
# Verify the cost from DUT_LEADER to ROUTER2 goes to infinity in 12
# mins.
routing_cost = 1
for i in range(0, 24):
self.simulator.go(30)
print("%ss" % ((i + 1) * 30))
leader_messages = self.simulator.get_messages_sent_by(DUT_LEADER)
msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT, False)
if msg is None:
continue
self.assertTrue(command.check_id_set(msg, router2_id))
routing_cost = command.get_routing_cost(msg, router2_id)
if routing_cost == 0:
break
self.assertTrue(routing_cost == 0)
self.simulator.go(config.INFINITE_COST_TIMEOUT + config.MAX_ADVERTISEMENT_INTERVAL)
leader_messages = self.simulator.get_messages_sent_by(DUT_LEADER)
msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT)
self.assertFalse(command.check_id_set(msg, router2_id))
self.simulator.go(720)
# 5
# Flush the message queue to avoid possible impact on follow-up
# verification.
self.simulator.get_messages_sent_by(DUT_LEADER)
self.nodes[ROUTER2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.simulator.go(config.MAX_ADVERTISEMENT_INTERVAL)
leader_messages = self.simulator.get_messages_sent_by(DUT_LEADER)
leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT)
# 6
self.nodes[ROUTER1].reset()
self.nodes[ROUTER2].reset()
router1_id = self.nodes[ROUTER1].get_router_id()
router2_id = self.nodes[ROUTER2].get_router_id()
self.simulator.go(720)
self.simulator.go(config.MAX_NEIGHBOR_AGE + config.MAX_ADVERTISEMENT_INTERVAL)
leader_messages = self.simulator.get_messages_sent_by(DUT_LEADER)
msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT)
self.assertEqual(command.get_routing_cost(msg, router1_id), 0)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
self.simulator.go(config.INFINITE_COST_TIMEOUT + config.MAX_ADVERTISEMENT_INTERVAL)
leader_messages = self.simulator.get_messages_sent_by(DUT_LEADER)
msg = leader_messages.last_mle_message(mle.CommandType.ADVERTISEMENT)
self.assertFalse(command.check_id_set(msg, router1_id))
self.assertFalse(command.check_id_set(msg, router2_id))
LEADER = pv.vars['LEADER']
ROUTER_1 = pv.vars['ROUTER_1']
ROUTER_2 = pv.vars['ROUTER_2']
leader_rid = pv.vars['LEADER_RLOC16'] >> 10
router_1_rid = pv.vars['ROUTER_1_RLOC16'] >> 10
router_2_rid = pv.vars['ROUTER_2_RLOC16'] >> 10
# Step 1: Ensure topology is formed correctly
pv.verify_attached('ROUTER_1', 'LEADER')
pv.verify_attached('ROUTER_2', 'ROUTER_1')
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p:
{1,2,1} == set(p.mle.tlv.route64.cost) and\
{leader_rid, router_1_rid, router_2_rid} ==
p.mle.tlv.route64.id_mask
).\
must_next()
# Step 4: The DUT’s routing cost to Router_2 MUST count to infinity
# The DUT MUST remove Router_2 ID from its ID set
# Verify route data has settled
_pkt = pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {1,0,1} == set(p.mle.tlv.route64.cost)).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p:
{1,1} == set(p.mle.tlv.route64.cost) and\
{leader_rid, router_1_rid} ==
p.mle.tlv.route64.id_mask
).\
must_next()
# Step 5: Re-attach Router_2 to Router_1.
# The DUT MUST reset the MLE Advertisement trickle timer and
# send an Advertisement
pv.verify_attached('ROUTER_2', 'ROUTER_1')
with pkts.save_index():
_pkt = pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: p.sniff_timestamp - _pkt.sniff_timestamp <= 3).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {1,0,1} == set(p.mle.tlv.route64.cost)).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {1,2,1} == set(p.mle.tlv.route64.cost) and\
p.sniff_timestamp - _pkt.sniff_timestamp <= 3 and\
{leader_rid, router_1_rid, router_2_rid} ==
p.mle.tlv.route64.id_mask
).\
must_next()
# Step 6: The DUT’s routing cost to Router_1 MUST go directly to
# infinity as there is no multi-hop cost for Router_1
# The DUT MUST remove Router_1 & Router_2 IDs from its ID set
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {0, 0, 1} == set(p.mle.tlv.route64.cost)).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {0, 1} == set(p.mle.tlv.route64.cost)).\
must_next()
pkts.filter_wpan_src64(LEADER).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p:
[1] == p.mle.tlv.route64.cost and\
{leader_rid} ==
p.mle.tlv.route64.id_mask
).\
must_next()
if __name__ == '__main__':
tests/scripts/thread-cert/Cert_5_3_06b_RouterIdMask.py
-80
View File
@@ -1,80 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2016, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import unittest
import thread_cert
LEADER = 1
ROUTER1 = 2
ROUTER2 = 3
class Cert_5_3_6_RouterIdMask(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'mode': 'rdn',
'panid': 0xface,
'allowlist': [ROUTER1]
},
ROUTER1: {
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER, ROUTER2]
},
ROUTER2: {
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [ROUTER1]
},
}
def test(self):
self.nodes[LEADER].start()
self.simulator.go(5)
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
self.nodes[ROUTER1].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER1].get_state(), 'router')
self.nodes[ROUTER2].start()
self.simulator.go(5)
self.assertEqual(self.nodes[ROUTER2].get_state(), 'router')
self.nodes[ROUTER1].stop()
self.nodes[ROUTER2].stop()
self.simulator.go(300)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/Makefile.am
-2
View File
@@ -56,7 +56,6 @@ EXTRA_DIST = \
Cert_5_3_04_AddressMapCache.py \
Cert_5_3_05_RoutingLinkQuality.py \
Cert_5_3_06_RouterIdMask.py \
Cert_5_3_06b_RouterIdMask.py \
Cert_5_3_07_DuplicateAddress.py \
Cert_5_3_08_ChildAddressSet.py \
Cert_5_3_09_AddressQuery.py \
@@ -232,7 +231,6 @@ check_SCRIPTS = \
Cert_5_3_04_AddressMapCache.py \
Cert_5_3_05_RoutingLinkQuality.py \
Cert_5_3_06_RouterIdMask.py \
Cert_5_3_06b_RouterIdMask.py \
Cert_5_3_07_DuplicateAddress.py \
Cert_5_3_08_ChildAddressSet.py \
Cert_5_3_09_AddressQuery.py \
tests/scripts/thread-cert/pktverify/layer_fields.py
+26 -1
View File
@@ -191,6 +191,31 @@ def _eth_addr(v: Union[LayerFieldsContainer, LayerField]) -> EthAddr:
return EthAddr(v.get_default_value())
def _routerid_set(v: Union[LayerFieldsContainer, LayerField]) -> set:
"""parse the layer field as a set of router ids
Notes: the router ID mask in wireshark is a
hexadecimal string separated by ':'
"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
try:
ridmask = str(v.get_default_value())
assert isinstance(ridmask, str), ridmask
ridmask_int = int(ridmask.replace(':', ''), base=16)
rid_set = set()
count = 0
while ridmask_int:
count += 1
if ridmask_int & 1:
rid_set.add(64 - count)
ridmask_int = ridmask_int >> 1
except ValueError:
pass
return rid_set
class _first(object):
"""parse the first layer field"""
@@ -275,7 +300,7 @@ _LAYER_FIELDS = {
'mle.tlv.route64.nbr_out': _list(_auto),
'mle.tlv.route64.nbr_in': _list(_auto),
'mle.tlv.route64.id_seq': _auto,
'mle.tlv.route64.id_mask': _auto,
'mle.tlv.route64.id_mask': _routerid_set,
'mle.tlv.route64.cost': _list(_auto),
'mle.tlv.response': _bytes,
'mle.tlv.mle_frm_cntr': _auto,