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

[nexus] support channel-based pcap filtering in verification (#12586)

Browse Source 
This commit enables channel-based filtering in Nexus simulation tests
by exporting node channel information and enhancing the pktverify
library to extract and use it.

Simulation Core:
- Export each node's PAN channel to the test JSON metadata in
  nexus_core.cpp.

Pktverify Library:
- Update packet.py to extract channel information from 'wpan-tap' or
  'wpan_tap' layers and normalize it into 'wpan.channel'.
- Update consts.py and layer_fields.py to support the new TAP layers
  and their channel fields.
- Enhance NullField to be callable and iterable, and update Bytes
  comparison to handle NullField, preventing script crashes when
  accessing missing layers or fields.

Verification Utilities:
- Automatically load channel information from test JSON into
  verification variables (e.g., {NODE_NAME}_CHANNEL).

Test Updates:
- Stabilize 1.1.9.2.14 by using channel-based filtering instead of
  relying solely on RLOC16s.
- Update 1.1.9.2.17 and 1.1.9.2.8 to use the normalized
  'wpan.channel' field instead of direct 'wpan_tap' access.
This commit is contained in:
Jonathan Hui
2026-02-28 21:15:00 -06:00
committed by GitHub
parent c9a153bdd8
commit 709217aa8a
10 changed files with 68 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/nexus/platform/nexus_core.cpp
+8
View File
@@ -191,6 +191,14 @@ void Core::SaveTestInfo(const char *aFilename, Node *aLeaderNode)
}
fprintf(file, " },\n");
fprintf(file, " \"channels\": {\n");
for (Node &node : mNodes)
{
fprintf(file, " \"%u\": %u%s\n", node.GetInstance().GetId(), node.Get<Mac::Mac>().GetPanChannel(),
(&node == tail) ? "" : ",");
}
fprintf(file, " },\n");
fprintf(file, " \"ipaddrs\": {\n");
for (Node &node : mNodes)
{
tests/nexus/verify_1_1_9_2_14.py
+14 -7
View File
@@ -65,6 +65,7 @@ def verify(pv):
COMMISSIONER_RLOC = pv.vars['COMMISSIONER_RLOC']
COMMISSIONER_MLEID = pv.vars['COMMISSIONER_MLEID']
ROUTER_1_MLEID = pv.vars['ROUTER_1_MLEID']
PRIMARY_CHANNEL = pv.vars['ROUTER_1_CHANNEL']
# Step 1: All
# - Description: Topology Ensure topology is formed correctly.
@@ -83,7 +84,8 @@ def verify(pv):
# - Channel Mask TLV
# - PAN ID TLV
print("Step 2: Commissioner")
pkts.filter_wpan_src64(COMMISSIONER).\
pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ipv6_src(COMMISSIONER_RLOC).\
filter_ipv6_dst(ROUTER_1_RLOC).\
filter_coap_request(consts.MGMT_PANID_QUERY).\
filter(lambda p: {
@@ -95,13 +97,14 @@ def verify(pv):
# Step 3: Router_1
# - Description: Automatically sends a MGMT_PANID_CONFLICT.ans reponse to the Commissioner.
# - Pass Criteria: For DUT = Router: The DUT MUST send MGMT_ED_REPORT.ans to the Commissioner:
# - Pass Criteria: For DUT = Router: The DUT MUST send MGMT_PANID_CONFLICT.ans to the Commissioner:
# - CoAP Request URI: coap://[Commissioner]:MM/c/pc
# - CoAP Payload:
# - Channel Mask TLV
# - PAN ID TLV
print("Step 3: Router_1")
pkts.filter_wpan_src64(ROUTER_1).\
pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ipv6_src(ROUTER_1_RLOC).\
filter_ipv6_dst(COMMISSIONER_RLOC).\
filter_coap_request(consts.MGMT_PANID_CONFLICT).\
filter(lambda p: {
@@ -122,7 +125,8 @@ def verify(pv):
# - Channel Mask TLV
# - PAN ID TLV
print("Step 4: Commissioner")
pkts.filter_wpan_src64(COMMISSIONER).\
pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ipv6_src(COMMISSIONER_RLOC).\
filter_coap_request(consts.MGMT_PANID_QUERY).\
filter(lambda p: str(p.ipv6.dst).startswith('ff33:0040:')).\
filter(lambda p: {
@@ -140,7 +144,8 @@ def verify(pv):
# - Channel Mask TLV
# - PAN ID TLV
print("Step 5: Router_1")
pkts.filter_wpan_src64(ROUTER_1).\
pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ipv6_src(ROUTER_1_RLOC).\
filter_ipv6_dst(COMMISSIONER_RLOC).\
filter_coap_request(consts.MGMT_PANID_CONFLICT).\
filter(lambda p: {
@@ -154,11 +159,13 @@ def verify(pv):
# - Description: Verify connectivity by sending an ICMPv6 Echo Request to the DUT mesh local address.
# - Pass Criteria: The DUT MUST respond with an ICMPv6 Echo Reply.
print("Step 6: Commissioner")
_pkt = pkts.filter_ping_request().\
_pkt = pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ping_request().\
filter_ipv6_src(COMMISSIONER_MLEID).\
filter_ipv6_dst(ROUTER_1_MLEID).\
must_next()
pkts.filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
pkts.filter_wpan_channel(PRIMARY_CHANNEL).\
filter_ping_reply(identifier=_pkt.icmpv6.echo.identifier).\
filter_ipv6_src(ROUTER_1_MLEID).\
filter_ipv6_dst(COMMISSIONER_MLEID).\
must_next()
tests/nexus/verify_1_1_9_2_17.py
+6 -6
View File
@@ -74,12 +74,12 @@ def verify(pv):
pkts.filter_wpan_src64(LEADER_1).\
filter_LLANMA().\
filter_mle_cmd(consts.MLE_ADVERTISEMENT).\
filter(lambda p: p.wpan_tap.ch_num == PRIMARY_CHANNEL).\
filter(lambda p: p.wpan.channel == PRIMARY_CHANNEL).\
must_next()
pkts.filter_wpan_src64(LEADER_2).\
filter_LLANMA().\
filter_mle_cmd(consts.MLE_ADVERTISEMENT).\
filter(lambda p: p.wpan_tap.ch_num == SECONDARY_CHANNEL).\
filter(lambda p: p.wpan.channel == SECONDARY_CHANNEL).\
must_next()
# Step 2: Leader_1
@@ -100,7 +100,7 @@ def verify(pv):
print("Step 4: The DUT MUST send a MLE Parent Request")
pkts.filter_wpan_src64(DUT).\
filter_mle_cmd(consts.MLE_PARENT_REQUEST).\
filter(lambda p: p.wpan_tap.ch_num == PRIMARY_CHANNEL).\
filter(lambda p: p.wpan.channel == PRIMARY_CHANNEL).\
must_next()
# Step 5: DUT
@@ -120,7 +120,7 @@ def verify(pv):
print("Step 6: The DUT MUST send a MLE Announce Message")
pkts.filter_wpan_src64(DUT).\
filter_mle_cmd(consts.MLE_ANNOUNCE).\
filter(lambda p: p.wpan_tap.ch_num == SECONDARY_CHANNEL).\
filter(lambda p: p.wpan.channel == SECONDARY_CHANNEL).\
filter(lambda p: {
consts.CHANNEL_TLV,
consts.ACTIVE_TIMESTAMP_TLV,
@@ -138,7 +138,7 @@ def verify(pv):
print("Step 7: Leader_2 sends a MLE Announce on the Primary channel")
pkts.filter_wpan_src64(LEADER_2).\
filter_mle_cmd(consts.MLE_ANNOUNCE).\
filter(lambda p: p.wpan_tap.ch_num == PRIMARY_CHANNEL).\
filter(lambda p: p.wpan.channel == PRIMARY_CHANNEL).\
must_next()
# Step 8: DUT
@@ -149,7 +149,7 @@ def verify(pv):
print("Step 8: The DUT MUST send a Parent Request on the Secondary channel")
pkts.filter_wpan_src64(DUT).\
filter_mle_cmd(consts.MLE_PARENT_REQUEST).\
filter(lambda p: p.wpan_tap.ch_num == SECONDARY_CHANNEL).\
filter(lambda p: p.wpan.channel == SECONDARY_CHANNEL).\
filter(lambda p: p.wpan.dst_pan == LEADER_2_PAN_ID).\
must_next()
tests/nexus/verify_1_1_9_2_8.py
+2 -2
View File
@@ -193,13 +193,13 @@ def verify(pv):
# We start searching from the same point for each DUT to allow intermingled packets.
pkts_dut = pkts.copy()
pkts_dut.filter_wpan_src64(dut). \
filter(lambda p: p.wpan_tap.ch_num == 11 and p.wpan.dst_pan == 0xFACE). \
filter(lambda p: p.wpan.channel == 11 and p.wpan.dst_pan == 0xFACE). \
filter(lambda p: p.mle.cmd == consts.MLE_PARENT_REQUEST). \
must_next()
# Then, it MUST attach using the new params.
pkts_dut.filter_wpan_src64(dut). \
filter(lambda p: p.wpan_tap.ch_num == 12 and p.wpan.dst_pan == 0xAFCE). \
filter(lambda p: p.wpan.channel == 12 and p.wpan.dst_pan == 0xAFCE). \
filter(lambda p: p.mle.cmd == consts.MLE_PARENT_REQUEST). \
must_next()
tests/nexus/verify_utils.py
+5
View File
@@ -366,6 +366,11 @@ def run_main(verify_func):
name = pv.test_info.get_node_name(int(node_id))
pv.add_vars(**{f'{name}_RLOC16': int(rloc16, 16)})
# Add channel variables
for node_id, channel in data.get('channels', {}).items():
name = pv.test_info.get_node_name(int(node_id))
pv.add_vars(**{f'{name}_CHANNEL': int(channel)})
verify_func(pv)
print("Verification PASSED")
except Exception as e:
tests/scripts/thread-cert/pktverify/bytes.py
+3 -1
View File
@@ -119,7 +119,9 @@ class Bytes(bytearray):
"""
Check if bytes is equal to other.
"""
if other is None:
from pktverify.null_field import nullField
if other is None or other is nullField:
return False
elif not isinstance(other, Bytes):
other = self.__class__(other)
tests/scripts/thread-cert/pktverify/consts.py
+2
View File
@@ -300,6 +300,8 @@ REAL_LAYER_NAMES = {
'coap',
'dtls',
'wpan',
'wpan-tap',
'wpan_tap',
'eth',
'tcp',
'udp',
tests/scripts/thread-cert/pktverify/layer_fields.py
+2
View File
@@ -280,6 +280,8 @@ _LAYER_FIELDS = {
'wpan.dst16': _auto,
'wpan.src64': _ext_addr,
'wpan.dst64': _ext_addr,
'wpan-tap.channel': _auto,
'wpan_tap.channel': _auto,
'wpan.fcs': _raw_hex_rev,
'wpan.fcs_ok': _auto,
'wpan.frame_length': _dec,
tests/scripts/thread-cert/pktverify/null_field.py
+12
View File
@@ -58,6 +58,18 @@ class NullField(object):
"""
return self
def __call__(self, *args, **kwargs):
"""
Calling NullField always returns NullField itself.
"""
return self
def __iter__(self):
"""
Iterating over NullField returns an empty iterator.
"""
return iter([])
def __setattr__(self, key, value):
pass
tests/scripts/thread-cert/pktverify/packet.py
+14
View File
@@ -69,6 +69,20 @@ class Packet(object):
logging.debug("stripping eth: src=%s, dst=%s", eth_src, eth_dst)
channel = eth_src[5]
self.wpan._add_field('wpan.channel', hex(channel))
elif layer.layer_name in ('wpan-tap', 'wpan_tap'):
channel_val = getattr(layer, 'ch_num', None)
if channel_val is None:
channel_val = getattr(layer, 'channel', None)
if channel_val is None and hasattr(layer, '_all_fields'):
channel_val = layer._all_fields.get(f'{layer.layer_name}.ch_num') or \
layer._all_fields.get(f'{layer.layer_name}.channel')
if channel_val is not None:
try:
channel = int(str(channel_val))
self.wpan._add_field('wpan.channel', hex(channel))
except (ValueError, TypeError):
pass
return