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

[thread-cert] refactor case 5.1.12 using pktverify (#5765)

Browse Source
This commit is contained in:
Jing Ma
2020-11-02 23:00:52 -08:00
committed by GitHub
parent 3da49e6f8f
commit e21864df16
2 changed files with 134 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_1_12_NewRouterNeighborSync.py
+129 -56
View File
@@ -31,26 +31,48 @@ import unittest
import mle
import thread_cert
from pktverify.consts import MLE_ADVERTISEMENT, MLE_PARENT_REQUEST, MLE_PARENT_RESPONSE, MLE_CHILD_UPDATE_RESPONSE, MLE_CHILD_ID_REQUEST, MLE_CHILD_ID_RESPONSE, MLE_LINK_REQUEST, MLE_LINK_ACCEPT, MLE_LINK_ACCEPT_AND_REQUEST, ADDR_SOL_URI, SOURCE_ADDRESS_TLV, MODE_TLV, TIMEOUT_TLV, CHALLENGE_TLV, RESPONSE_TLV, LINK_LAYER_FRAME_COUNTER_TLV, MLE_FRAME_COUNTER_TLV, ROUTE64_TLV, ADDRESS16_TLV, LEADER_DATA_TLV, NETWORK_DATA_TLV, TLV_REQUEST_TLV, SCAN_MASK_TLV, CONNECTIVITY_TLV, LINK_MARGIN_TLV, VERSION_TLV, ADDRESS_REGISTRATION_TLV, NL_MAC_EXTENDED_ADDRESS_TLV, NL_RLOC16_TLV, NL_STATUS_TLV, NL_ROUTER_MASK_TLV, COAP_CODE_ACK
from pktverify.packet_verifier import PacketVerifier
from pktverify.null_field import nullField
LEADER = 1
ROUTER1 = 2
ROUTER2 = 3
# Test Purpose and Description:
# -----------------------------
# The purpose of this test case is to validate that when the DUT sees a new router for the first time, it will synchronize using the New Router Neighbor Synchronization procedure.
#
# Test Topology:
# -------------
# Leader
# / \
# Router2 Router1[DUT]
#
# DUT Types:
# ----------
# Router
class Cert_5_1_12_NewRouterSync(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
LEADER: {
'name': 'LEADER',
'mode': 'rdn',
'panid': 0xface,
'allowlist': [ROUTER1, ROUTER2]
},
ROUTER1: {
'name': 'ROUTER_1',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
'allowlist': [LEADER]
},
ROUTER2: {
'name': 'ROUTER_2',
'mode': 'rdn',
'panid': 0xface,
'router_selection_jitter': 1,
@@ -58,43 +80,6 @@ class Cert_5_1_12_NewRouterSync(thread_cert.TestCase):
},
}
def verify_step_4(self, router1_messages, router2_messages, req_receiver, accept_receiver):
if router2_messages.contains_mle_message(mle.CommandType.LINK_REQUEST) and (
router1_messages.contains_mle_message(mle.CommandType.LINK_ACCEPT) or
router1_messages.contains_mle_message(mle.CommandType.LINK_ACCEPT_AND_REQUEST)):
msg = router2_messages.next_mle_message(mle.CommandType.LINK_REQUEST)
msg.assertSentToNode(self.nodes[req_receiver])
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.Challenge)
msg.assertMleMessageContainsTlv(mle.Version)
msg.assertMleMessageContainsTlv(mle.TlvRequest)
msg = router1_messages.next_mle_message_of_one_of_command_types(
mle.CommandType.LINK_ACCEPT_AND_REQUEST,
mle.CommandType.LINK_ACCEPT,
)
self.assertIsNotNone(msg)
msg.assertSentToNode(self.nodes[accept_receiver])
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter)
msg.assertMleMessageContainsOptionalTlv(mle.MleFrameCounter)
msg.assertMleMessageContainsTlv(mle.Version)
msg.assertMleMessageContainsTlv(mle.LinkMargin)
if msg.mle.command.type == mle.CommandType.LINK_ACCEPT_AND_REQUEST:
msg.assertMleMessageContainsTlv(mle.TlvRequest)
msg.assertMleMessageContainsTlv(mle.Challenge)
return True
else:
return False
def test(self):
self.nodes[LEADER].start()
self.simulator.go(5)
@@ -110,31 +95,119 @@ class Cert_5_1_12_NewRouterSync(thread_cert.TestCase):
self.simulator.go(10)
self.simulator.get_messages_sent_by(LEADER)
router1_messages = self.simulator.get_messages_sent_by(ROUTER1)
router2_messages = self.simulator.get_messages_sent_by(ROUTER2)
# 2 - Router1
msg = router1_messages.next_mle_message(mle.CommandType.ADVERTISEMENT)
msg.assertSentWithHopLimit(255)
msg.assertSentToDestinationAddress("ff02::1")
msg.assertMleMessageContainsTlv(mle.SourceAddress)
msg.assertMleMessageContainsTlv(mle.LeaderData)
msg.assertMleMessageContainsTlv(mle.Route64)
self.nodes[ROUTER1].add_allowlist(self.nodes[ROUTER2].get_addr64())
self.nodes[ROUTER2].add_allowlist(self.nodes[ROUTER1].get_addr64())
self.simulator.go(35)
self.simulator.get_messages_sent_by(LEADER)
router1_messages = self.simulator.get_messages_sent_by(ROUTER1)
router2_messages = self.simulator.get_messages_sent_by(ROUTER2)
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
# 4 - Router1, Router2
self.assertTrue(
self.verify_step_4(router1_messages, router2_messages, ROUTER1, ROUTER2) or
self.verify_step_4(router2_messages, router1_messages, ROUTER2, ROUTER1))
LEADER = pv.vars['LEADER']
ROUTER_1 = pv.vars['ROUTER_1']
ROUTER_2 = pv.vars['ROUTER_2']
# Step 1: Verify topology is formed correctly.
# Step 2: The DUT MUST send properly formatted MLE Advertisements with
# an IP Hop Limit of 255 to the Link-Local All Nodes multicast
# address (FF02::1).
# The following TLVs MUST be present in the MLE Advertisements:
# - Leader Data TLV
# - Route64 TLV
# - Source Address TLV
pv.verify_attached('ROUTER_1')
pkts.filter_wpan_src64(ROUTER_1).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {
LEADER_DATA_TLV,
ROUTE64_TLV,
SOURCE_ADDRESS_TLV
} == set(p.mle.tlv.type) and\
p.ipv6.hlim == 255).\
must_next()
pv.verify_attached('ROUTER_2')
pkts.filter_wpan_src64(ROUTER_2).\
filter_LLANMA().\
filter_mle_cmd(MLE_ADVERTISEMENT).\
filter(lambda p: {
LEADER_DATA_TLV,
ROUTE64_TLV,
SOURCE_ADDRESS_TLV
} == set(p.mle.tlv.type) and\
p.ipv6.hlim == 255).\
must_next()
# Step 4: The DUT and Router_2 exchange unicast Link Request and unicast
# Link Accept messages OR Link Accept and Request messages.
#
# The Link Request Message MUST be unicast and contain
# the following TLVs:
# - Challenge TLV
# - Leader Data TLV
# - Source Address TLV
# - Version TLV
# - TLV Request TLV: Link Margin
#
# Link Accept or Link Accept And Request Messages MUST be
# Unicast.
# The following TLVs MUST be present in the Link Accept or
# Link Accept And Request Messages :
# - Leader Data TLV
# - Link-layer Frame Counter TLV
# - Link Margin TLV
# - Response TLV
# - Source Address TLV
# - Version TLV
# - TLV Request TLV: Link Margin
# - Challenge TLV (optional)
# - MLE Frame Counter TLV (optional)
# The Challenge TLV and TLV Request TLV MUST be included
# if the response is an Accept and Request message.
lq_src = ROUTER_1
lq_dst = ROUTER_2
_pkt = pkts.filter_mle_cmd(MLE_LINK_REQUEST).\
filter(lambda p: {
CHALLENGE_TLV,
LEADER_DATA_TLV,
SOURCE_ADDRESS_TLV,
VERSION_TLV,
TLV_REQUEST_TLV,
LINK_MARGIN_TLV
} <= set(p.mle.tlv.type) and\
p.mle.tlv.link_margin is nullField and\
(p.wpan.src64 == ROUTER_1 or\
p.wpan.src64 == ROUTER_2)
).\
must_next()
if _pkt.wpan.src64 != ROUTER_1:
_pkt.must_verify(lambda p: p.wpan.dst64 == ROUTER_1)
lq_src = ROUTER_2
lq_dst = ROUTER_1
_pkt = pkts.filter_wpan_src64(lq_dst).\
filter_wpan_dst64(lq_src).\
filter_mle_cmd2(MLE_LINK_ACCEPT, MLE_LINK_ACCEPT_AND_REQUEST).\
filter(lambda p: {
LEADER_DATA_TLV,
LINK_LAYER_FRAME_COUNTER_TLV,
LINK_MARGIN_TLV,
RESPONSE_TLV,
SOURCE_ADDRESS_TLV,
VERSION_TLV,
TLV_REQUEST_TLV,
LINK_MARGIN_TLV
} <= set(p.mle.tlv.type) and\
p.mle.tlv.link_margin is not nullField
).\
must_next()
if _pkt.mle.cmd == MLE_LINK_ACCEPT_AND_REQUEST:
_pkt.must_verify(lambda p: {CHALLENGE_TLV, TLV_REQUEST_TLV} <= set(p.mle.tlv.type))
if __name__ == '__main__':
tests/scripts/thread-cert/pktverify/packet_filter.py
+5
View File
@@ -511,6 +511,11 @@ class PacketFilter(object):
assert isinstance(cmd, int), cmd
return self.filter(lambda p: p.mle.cmd == cmd, **kwargs)
def filter_mle_cmd2(self, cmd1, cmd2, **kwargs):
assert isinstance(cmd1, int), cmd1
assert isinstance(cmd2, int), cmd2
return self.filter(lambda p: p.mle.cmd == cmd1 or p.mle.cmd == cmd2, **kwargs)
def filter_mle_has_tlv(self, *tlv_types, **kwargs):
return self.filter(lambda p: set(tlv_types) <= set(p.mle.tlv.type), **kwargs)