mirror of
https://github.com/espressif/openthread.git
synced 2026-08-31 22:39:54 +00:00
[pktverify] add filter_ipv6_src() to PacketFilter class (#12444)
This commit adds the filter_ipv6_src() method to the PacketFilter class in the pktverify framework. This method improves the symmetry of the filtering API, complementing the existing filter_ipv6_dst(). The new method is used to update several verification scripts in tests/nexus, replacing manual lambda filters for IPv6 source addresses with a cleaner and more readable API call. Updated scripts: - tests/nexus/verify_5_3_3.py - tests/nexus/verify_5_3_7.py - tests/nexus/verify_5_3_8.py - tests/nexus/verify_5_3_10.py
This commit is contained in:
@@ -147,8 +147,8 @@ def verify(pv):
|
||||
pkts.filter_wpan_src16(DUT_RLOC16).\
|
||||
filter_coap_request(consts.ADDR_NTF_URI).\
|
||||
filter(lambda p: p.coap.type == COAP_TYPE_CON).\
|
||||
filter(lambda p: p.ipv6.src == DUT_RLOC).\
|
||||
filter(lambda p: p.ipv6.dst == BR_RLOC).\
|
||||
filter_ipv6_src(DUT_RLOC).\
|
||||
filter_ipv6_dst(BR_RLOC).\
|
||||
filter(lambda p: {
|
||||
consts.NL_TARGET_EID_TLV,
|
||||
consts.NL_RLOC16_TLV,
|
||||
@@ -209,7 +209,7 @@ def verify(pv):
|
||||
|
||||
# DUT must NOT respond with Address Notification
|
||||
with pkts.save_index():
|
||||
pkts.filter(lambda p: p.ipv6.src == DUT_RLOC).\
|
||||
pkts.filter_ipv6_src(DUT_RLOC).\
|
||||
filter_coap_request(consts.ADDR_NTF_URI).\
|
||||
filter(lambda p: is_same_iid(p.coap.tlv.target_eid, MED_1_GUA)).\
|
||||
must_not_next()
|
||||
|
||||
@@ -133,7 +133,7 @@ def verify(pv):
|
||||
# DUT responds with Address Notification
|
||||
pkts.filter_coap_request(consts.ADDR_NTF_URI).\
|
||||
filter(lambda p: p.coap.type == COAP_TYPE_CON).\
|
||||
filter(lambda p: p.ipv6.src == pv.vars['DUT_RLOC']).\
|
||||
filter_ipv6_src(pv.vars['DUT_RLOC']).\
|
||||
filter_ipv6_dst(pv.vars['ROUTER_1_RLOC']).\
|
||||
filter(lambda p: {
|
||||
consts.NL_ML_EID_TLV,
|
||||
@@ -161,7 +161,7 @@ def verify(pv):
|
||||
|
||||
# Now check in the range up to this reply.
|
||||
pkts.range(start_of_step_4, end_of_step_4).filter_wpan_src64(DUT).\
|
||||
filter(lambda p: p.ipv6.src == pv.vars['DUT_RLOC']).\
|
||||
filter_ipv6_src(pv.vars['DUT_RLOC']).\
|
||||
filter_coap_request(consts.ADDR_QRY_URI).\
|
||||
filter(lambda p: p.coap.tlv.target_eid == pv.vars['ROUTER_3_MLEID']).\
|
||||
must_not_next()
|
||||
@@ -201,7 +201,7 @@ def verify(pv):
|
||||
|
||||
# DUT must NOT respond with Address Notification for MED_1 ML-EID
|
||||
with pkts.save_index():
|
||||
pkts.filter(lambda p: p.ipv6.src == pv.vars['DUT_RLOC']).\
|
||||
pkts.filter_ipv6_src(pv.vars['DUT_RLOC']).\
|
||||
filter_coap_request(consts.ADDR_NTF_URI).\
|
||||
filter(lambda p: p.coap.tlv.target_eid == pv.vars['MED_1_MLEID']).\
|
||||
must_not_next()
|
||||
|
||||
@@ -148,7 +148,8 @@ def verify(pv):
|
||||
filter_wpan_src64(LEADER).\
|
||||
filter_RLARMA().\
|
||||
filter_coap_request(consts.ADDR_ERR_URI).\
|
||||
filter(lambda p: p.coap.tlv.target_eid == '2001::1' and p.ipv6.src == pv.vars['LEADER_RLOC']).\
|
||||
filter(lambda p: p.coap.tlv.target_eid == '2001::1').\
|
||||
filter_ipv6_src(pv.vars['LEADER_RLOC']).\
|
||||
must_next()
|
||||
|
||||
|
||||
|
||||
@@ -160,7 +160,7 @@ def _verify_echo(pv, src_ext, dst_ext, target_ip):
|
||||
# Verify no Address Query from DUT (Leader) for the target IP in this range
|
||||
pkts.range(start_index, end_index).\
|
||||
filter_wpan_src64(pv.vars['LEADER']).\
|
||||
filter(lambda p: p.ipv6.src == DUT_RLOC).\
|
||||
filter_ipv6_src(DUT_RLOC).\
|
||||
filter_coap_request(consts.ADDR_QRY_URI).\
|
||||
filter(lambda p: Ipv6Addr(p.coap.tlv.target_eid) == target_ip_addr or\
|
||||
Ipv6Addr(p.coap.tlv.target_eid)[8:] == target_iid_bytes).\
|
||||
|
||||
@@ -538,6 +538,10 @@ class PacketFilter(object):
|
||||
assert isinstance(addr, (str, EthAddr))
|
||||
return self.filter(lambda p: p.eth.src == addr, **kwargs)
|
||||
|
||||
def filter_ipv6_src(self, addr, **kwargs):
|
||||
assert isinstance(addr, (str, Ipv6Addr))
|
||||
return self.filter(lambda p: p.ipv6.src == addr, **kwargs)
|
||||
|
||||
def filter_ipv6_dst(self, addr, **kwargs):
|
||||
assert isinstance(addr, (str, Ipv6Addr))
|
||||
return self.filter(lambda p: p.ipv6.dst == addr, **kwargs)
|
||||
|
||||
Reference in New Issue
Block a user