espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-31 16:17:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

[thread-cert] refactor case 5.3.2 using pktverify (#5803)

Browse Source
This commit is contained in:
Jing Ma
2020-11-16 17:17:48 -08:00
committed by GitHub
parent 515b2bb067
commit 1fc0a7be77
3 changed files with 217 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_3_02_RealmLocal.py
+195 -24
View File
@@ -31,33 +31,59 @@ import unittest
import config
import thread_cert
from pktverify.consts import MLE_CHILD_ID_REQUEST, MLE_CHILD_ID_RESPONSE, REALM_LOCAL_ALL_NODES_ADDRESS, REALM_LOCAL_ALL_ROUTERS_ADDRESS, REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS
from pktverify.packet_verifier import PacketVerifier
LEADER = 1
ROUTER1 = 2
DUT_ROUTER2 = 3
SED1 = 4
FRAGMENTED_DATA_LEN = 256
# Test Purpose and Description:
# -----------------------------
# The purpose of this test case is to validate the Realm-Local addresses
# that the DUT auto-configures.
#
# Test Topology:
# -------------
# Leader
# |
# Router_1 - Router_2(DUT)
# |
# SED
#
# DUT Types:
# ----------
# Router
class Cert_5_3_2_RealmLocal(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'name': 'LEADER',
'mode': 'rdn',
'panid': 0xface,
'allowlist': [ROUTER1]
},
ROUTER1: {
'name': 'ROUTER_1',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER, DUT_ROUTER2]
},
DUT_ROUTER2: {
'name': 'ROUTER_2',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [ROUTER1, SED1]
},
SED1: {
'name': 'SED',
'is_mtd': True,
'mode': 'n',
'panid': 0xface,
@@ -84,46 +110,191 @@ class Cert_5_3_2_RealmLocal(thread_cert.TestCase):
self.simulator.go(5)
self.assertEqual(self.nodes[SED1].get_state(), 'child')
self.collect_ipaddrs()
self.collect_rloc16s()
# 2 & 3
mleid = self.nodes[DUT_ROUTER2].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
self.assertTrue(self.nodes[LEADER].ping(mleid, size=256))
self.assertTrue(self.nodes[LEADER].ping(mleid, size=FRAGMENTED_DATA_LEN))
self.simulator.go(2)
self.assertTrue(self.nodes[LEADER].ping(mleid))
self.simulator.go(2)
# 4 & 5
self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2, size=256))
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2, size=FRAGMENTED_DATA_LEN))
self.simulator.go(5)
self.assertTrue(self.nodes[LEADER].ping('ff03::1', num_responses=2))
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.simulator.go(5)
# 6 & 7
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2, size=256))
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2, size=FRAGMENTED_DATA_LEN))
self.simulator.go(5)
self.assertTrue(self.nodes[LEADER].ping('ff03::2', num_responses=2))
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertFalse(sed_messages.contains_icmp_message())
self.simulator.go(5)
# 8
self.assertTrue(self.nodes[LEADER].ping(
config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS,
num_responses=3,
size=256,
size=FRAGMENTED_DATA_LEN,
))
self.simulator.go(2)
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertTrue(sed_messages.contains_icmp_message())
self.simulator.go(5)
self.assertTrue(self.nodes[LEADER].ping(
config.REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS,
num_responses=3,
))
self.simulator.go(2)
sed_messages = self.simulator.get_messages_sent_by(SED1)
self.assertTrue(sed_messages.contains_icmp_message())
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
LEADER_MLEID = pv.vars['LEADER_MLEID']
ROUTER_1 = pv.vars['ROUTER_1']
ROUTER_2 = pv.vars['ROUTER_2']
ROUTER_2_RLOC16 = pv.vars['ROUTER_2_RLOC16']
ROUTER_2_MLEID = pv.vars['ROUTER_2_MLEID']
SED = pv.vars['SED']
SED_RLOC16 = pv.vars['SED_RLOC16']
# Step 1: Build the topology as described
pv.verify_attached('ROUTER_1', 'LEADER')
pv.verify_attached('ROUTER_2', 'ROUTER_1')
pv.verify_attached('SED', 'ROUTER_2', 'MTD')
# Step 2: Leader sends a Fragmented ICMPv6 Echo Request to
# DUT's ML-EID
# The DUT MUST respond with an ICMPv6 Echo Reply
_pkt = pkts.filter_ping_request().\
filter_ipv6_src_dst(LEADER_MLEID, ROUTER_2_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
# Step 3: Leader sends a Unfragmented ICMPv6 Echo Request to
# DUT’s ML-EID
# The DUT MUST respond with an ICMPv6 Echo Reply
_pkt = pkts.filter_ping_request().\
filter_ipv6_src_dst(LEADER_MLEID, ROUTER_2_MLEID).\
must_next()
pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
must_next()
# Step 4: Leader sends a Fragmented ICMPv6 Echo Request to the
# Realm-Local All Nodes multicast address (FF03::1)
# The DUT MUST respond with an ICMPv6 Echo Reply
# The DUT MUST NOT forward the ICMPv6 Echo Request to SED
_pkt1 = pkts.filter_ping_request().\
filter_wpan_src64(LEADER).\
filter_ipv6_dst(REALM_LOCAL_ALL_NODES_ADDRESS).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
with pkts.save_index():
pkts.filter_ping_reply(identifier=_pkt1.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
pkts.filter_ping_request(identifier=_pkt1.icmpv6.echo.identifier).\
filter_wpan_src16_dst16(ROUTER_2_RLOC16, SED_RLOC16).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_not_next()
# Step 5: Leader sends an Unfragmented ICMPv6 Echo Request to the
# Realm-Local All Nodes multicast address (FF03::1)
# The DUT MUST respond with an ICMPv6 Echo Reply
# The DUT MUST NOT forward the ICMPv6 Echo Request to SED
_pkt2 = pkts.filter_ping_request().\
filter_wpan_src64(LEADER).\
filter_ipv6_dst(REALM_LOCAL_ALL_NODES_ADDRESS).\
filter(lambda p: p.icmpv6.echo.sequence_number !=
_pkt1.icmpv6.echo.sequence_number
).\
must_next()
with pkts.save_index():
pkts.filter_ping_reply(identifier=_pkt2.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
must_next()
pkts.filter_ping_request(identifier = _pkt2.icmpv6.echo.identifier).\
filter_wpan_src16_dst16(ROUTER_2_RLOC16, SED_RLOC16).\
must_not_next()
# Step 6: Leader sends a Fragmented ICMPv6 Echo Request to the
# Realm-Local All Routers multicast address (FF03::2)
# The DUT MUST respond with an ICMPv6 Echo Reply
# The DUT MUST NOT forward the ICMPv6 Echo Request to SED
_pkt1 = pkts.filter_ping_request().\
filter_wpan_src64(LEADER).\
filter_ipv6_dst(REALM_LOCAL_ALL_ROUTERS_ADDRESS).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
with pkts.save_index():
pkts.filter_ping_reply(identifier=_pkt1.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
pkts.filter_ping_request(identifier=_pkt1.icmpv6.echo.identifier).\
filter_wpan_src16_dst16(ROUTER_2_RLOC16, SED_RLOC16).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_not_next()
# Step 7: Leader sends an Unfragmented ICMPv6 Echo Request to the
# Realm-Local All Routers multicast address (FF03::2)
# The DUT MUST respond with an ICMPv6 Echo Reply
# The DUT MUST NOT forward the ICMPv6 Echo Request to SED
_pkt2 = pkts.filter_ping_request().\
filter_wpan_src64(LEADER).\
filter_ipv6_dst(REALM_LOCAL_ALL_ROUTERS_ADDRESS).\
filter(lambda p: p.icmpv6.echo.sequence_number !=
_pkt1.icmpv6.echo.sequence_number
).\
must_next()
with pkts.save_index():
pkts.filter_ping_reply(identifier=_pkt2.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
must_next()
pkts.filter_ping_request(identifier=_pkt2.icmpv6.echo.identifier).\
filter_wpan_src16_dst16(ROUTER_2_RLOC16, SED_RLOC16).\
must_not_next()
# Step 8: Leader sends a Fragmented ICMPv6 Echo Request to the
# Realm-Local All Thread Nodes multicast address
# The DUT MUST respond with an ICMPv6 Echo Reply
# The Realm-Local All Thread Nodes multicast address
# MUST be a realm-local Unicast Prefix-Based Multicast
# Address [RFC 3306], with:
# - flgs set to 3 (P = 1 and T = 1)
# - scop set to 3
# - plen set to the Mesh Local Prefix length
# - network prefix set to the Mesh Local Prefix
# - group ID set to 1
# The DUT MUST use IEEE 802.15.4 indirect transmissions
# to forward packet to SED
_pkt = pkts.filter_ping_request().\
filter_wpan_src64(LEADER).\
filter_ipv6_dst(REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
with pkts.save_index():
pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
filter_ipv6_src_dst(ROUTER_2_MLEID, LEADER_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
pkts.filter_ping_request(identifier = _pkt.icmpv6.echo.identifier).\
filter_wpan_src16_dst16(ROUTER_2_RLOC16, SED_RLOC16).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
filter_wpan_src64(SED).\
filter_ipv6_dst(LEADER_MLEID).\
filter(lambda p: p.icmpv6.data.len == FRAGMENTED_DATA_LEN).\
must_next()
if __name__ == '__main__':
tests/scripts/thread-cert/pktverify/consts.py
+1
View File
@@ -35,6 +35,7 @@ BACKBONE_IPV6_PREFIX = Bytes('91')
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_NODES_ADDRESS = Ipv6Addr('ff03::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')
tests/scripts/thread-cert/pktverify/packet_verifier.py
+21 -9
View File
@@ -161,29 +161,41 @@ class PacketVerifier(object):
logging.info("add extra var: %s = %s", k, v)
self._vars[k] = v
def verify_attached(self, name: str, pkts=None) -> VerifyResult:
def verify_attached(self, child: str, parent: str = None, child_type: str = 'FTD', pkts=None) -> VerifyResult:
"""
Verify that the device attaches to the Thread network.
:param name: The device name.
:param child: The child device name.
:param parent: The parent device name.
:param child_type: The child device type (FTD, MTD).
"""
result = VerifyResult()
assert self.is_thread_device(name), name
assert self.is_thread_device(child), child
assert child_type in ('FTD', 'MTD'), child_type
pkts = pkts or self.pkts
extaddr = self.vars[name]
child_extaddr = self.vars[child]
src_pkts = pkts.filter_wpan_src64(extaddr)
src_pkts = pkts.filter_wpan_src64(child_extaddr)
if parent:
assert self.is_thread_device(parent), parent
src_pkts = pkts.filter_wpan_src64(child_extaddr).\
filter_wpan_dst64(self.vars[parent])
src_pkts.filter_mle_cmd(MLE_CHILD_ID_REQUEST).must_next() # Child Id Request
result.record_last('child_id_request', pkts)
dst_pkts = pkts.filter_wpan_dst64(extaddr)
dst_pkts = pkts.filter_wpan_dst64(child_extaddr)
if parent:
dst_pkts = pkts.filter_wpan_src64(self.vars[parent]).\
filter_wpan_dst64(child_extaddr)
dst_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next() # Child Id Response
result.record_last('child_id_response', pkts)
with pkts.save_index():
src_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next() # MLE Advertisement
result.record_last('mle_advertisement', pkts)
logging.info(f"verify attached: d={name}, result={result}")
if child_type == 'FTD':
src_pkts = pkts.filter_wpan_src64(child_extaddr)
src_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next() # MLE Advertisement
result.record_last('mle_advertisement', pkts)
logging.info(f"verify attached: d={child}, result={result}")
return result