[PATCH 2/2] tests: eht: check for duplicate broadcast packets in multi-link
Johannes Berg
johannes at sipsolutions.net
Fri Sep 11 15:50:09 PDT 2026
From: Johannes Berg <johannes.berg at intel.com>
We shouldn't receive multiple identical multicast packets even
in a multi-link scenario, check for that.
Signed-off-by: Johannes Berg <johannes.berg at intel.com>
---
tests/hwsim/hwsim_utils.py | 54 ++++++++++++++++++++++++++++++++------
tests/hwsim/test_eht.py | 7 ++---
2 files changed, 50 insertions(+), 11 deletions(-)
diff --git a/tests/hwsim/hwsim_utils.py b/tests/hwsim/hwsim_utils.py
index 7b0afb496061..c8d918cb7376 100644
--- a/tests/hwsim/hwsim_utils.py
+++ b/tests/hwsim/hwsim_utils.py
@@ -59,17 +59,23 @@ def run_multicast_connectivity_test(dev1, dev2, tos=None,
ifname1=None, ifname2=None,
config=True, timeout=5,
send_len=None, multicast_to_unicast=False,
- broadcast_retry_c=1):
+ broadcast_retry_c=1,
+ check_dup=False):
addr1 = dev1.get_addr(dev1group)
addr2 = dev2.get_addr(dev2group)
if config:
config_data_test(dev1, dev2, dev1group, dev2group, ifname1, ifname2)
- cmd = "DATA_TEST_TX ff:ff:ff:ff:ff:ff {} {}".format(addr1, tos)
+ cmd_tmpl = "DATA_TEST_TX ff:ff:ff:ff:ff:ff {} {}".format(addr1, tos)
if send_len is not None:
- cmd += " len=" + str(send_len)
+ cmd_tmpl += " len=" + str(send_len)
+ rxed = set()
for i in range(broadcast_retry_c):
+ if check_dup:
+ cmd = cmd_tmpl + " id=" + str(1000 + i)
+ else:
+ cmd = cmd_tmpl
try:
if dev1group:
dev1.group_request(cmd)
@@ -96,18 +102,47 @@ def run_multicast_connectivity_test(dev1, dev2, tos=None,
else:
if " len=" in ev:
raise Exception("Unexpected dev1->dev2 broadcast data length")
+ if check_dup:
+ if not " id=" in ev:
+ raise Exception("Expected IP ID in RX event")
+ idstr = ev.split(' id=')[1]
+ if ' ' in idstr:
+ idstr = idstr.split(' ')[0]
+ _id = int(idstr)
+ if _id in rxed:
+ raise Exception("duplicate packet with ID %d received" % _id)
+ rxed.add(_id)
break
except Exception as e:
if i == broadcast_retry_c - 1:
raise
+ if check_dup:
+ if dev2group:
+ ev = dev2.wait_group_event(["DATA-TEST-RX"],
+ timeout=timeout)
+ else:
+ ev = dev2.wait_event(["DATA-TEST-RX"], timeout=timeout)
+ if not ev:
+ return
+ if not " id=" in ev:
+ raise Exception("Expected IP ID in RX event")
+ idstr = ev.split(' id=')[1]
+ if ' ' in idstr:
+ idstr = idstr.split(' ')[0]
+ _id = int(idstr)
+ if _id in rxed:
+ raise Exception("duplicate packet with ID %d received" % _id)
+
def run_connectivity_test(dev1, dev2, tos, dev1group=False, dev2group=False,
ifname1=None, ifname2=None, config=True, timeout=5,
multicast_to_unicast=False, broadcast=True,
- send_len=None):
+ send_len=None, check_bcast_dup=False):
addr1 = dev1.get_addr(dev1group)
addr2 = dev2.get_addr(dev2group)
+ assert not check_bcast_dup or broadcast
+
dev1.dump_monitor()
dev2.dump_monitor()
@@ -146,7 +181,8 @@ def run_connectivity_test(dev1, dev2, tos, dev1group=False, dev2group=False,
run_multicast_connectivity_test(dev1, dev2, tos,
dev1group, dev2group,
ifname1, ifname2, False, timeout,
- send_len, False, broadcast_retry_c)
+ send_len, False, broadcast_retry_c,
+ check_dup=check_bcast_dup)
cmd = "DATA_TEST_TX {} {} {}".format(addr1, addr2, tos)
if send_len is not None:
@@ -175,7 +211,8 @@ def run_connectivity_test(dev1, dev2, tos, dev1group=False, dev2group=False,
dev2group, dev1group,
ifname2, ifname1, False, timeout,
send_len, multicast_to_unicast,
- broadcast_retry_c)
+ broadcast_retry_c,
+ check_dup=check_bcast_dup)
finally:
if config:
@@ -192,7 +229,7 @@ def test_connectivity(dev1, dev2, dscp=None, tos=None, max_tries=1,
dev1group=False, dev2group=False,
ifname1=None, ifname2=None, config=True, timeout=5,
multicast_to_unicast=False, success_expected=True,
- broadcast=True, send_len=None):
+ broadcast=True, send_len=None, check_bcast_dup=False):
if dscp:
tos = dscp << 2
if not tos:
@@ -206,7 +243,8 @@ def test_connectivity(dev1, dev2, dscp=None, tos=None, max_tries=1,
ifname1, ifname2, config=config,
timeout=timeout,
multicast_to_unicast=multicast_to_unicast,
- broadcast=broadcast, send_len=send_len)
+ broadcast=broadcast, send_len=send_len,
+ check_bcast_dup=check_bcast_dup)
success = True
break
except Exception as e:
diff --git a/tests/hwsim/test_eht.py b/tests/hwsim/test_eht.py
index bcdd034bc26b..61cf55dd426e 100644
--- a/tests/hwsim/test_eht.py
+++ b/tests/hwsim/test_eht.py
@@ -108,9 +108,10 @@ def eht_verify_status(wpas, hapd, freq, bw, is_ht=False, is_vht=False,
time.sleep(0.1)
_eht_verify_links(wpas, valid_links, active_links)
-def traffic_test(wpas, hapd, success=True, ifname2=None):
+def traffic_test(wpas, hapd, success=True, ifname2=None, check_bcast_dup=False):
hwsim_utils.test_connectivity(wpas, hapd, success_expected=success,
- ifname2=ifname2)
+ ifname2=ifname2,
+ check_bcast_dup=check_bcast_dup)
def test_eht_open(dev, apdev):
"""EHT AP with open mode configuration"""
@@ -545,7 +546,7 @@ def _eht_mld_owe_two_links(dev, apdev, second_link_disabled=False,
traffic_test(wpas, hapd0)
if not second_link_disabled:
- traffic_test(wpas, hapd1)
+ traffic_test(wpas, hapd1, check_bcast_dup=True)
if only_one_link:
wpas.set("bssid_filter", "")
--
2.55.0
More information about the Hostap
mailing list