[PATCH 8/9] tests: Add test for matching ML neighbor entries

Benjamin Berg benjamin at sipsolutions.net
Fri Jul 18 04:01:04 PDT 2025


From: Benjamin Berg <benjamin.berg at intel.com>

This adds a test that neighbor entries with ML element work and will
apply to the the entire MLD AP instead of just the BSS that was listed
explicitly.

Signed-off-by: Benjamin Berg <benjamin.berg at intel.com>
Reviewed-by: Andrei Otcheretianski <andrei.otcheretianski at intel.com>
---
 tests/hwsim/hostapd.py  | 18 ++++++++
 tests/hwsim/test_eht.py | 95 +++++++++++++++++++++++++++++++++++++++++
 wpaspy/wpaspy.py        |  4 ++
 3 files changed, 117 insertions(+)

diff --git a/tests/hwsim/hostapd.py b/tests/hwsim/hostapd.py
index 9cfe5a2b61..163de0b2a3 100644
--- a/tests/hwsim/hostapd.py
+++ b/tests/hwsim/hostapd.py
@@ -15,6 +15,7 @@ import wpaspy
 import remotehost
 import utils
 import subprocess
+import select
 from remotectrl import RemoteCtrl
 
 logger = logging.getLogger()
@@ -658,6 +659,23 @@ class Hostapd:
             return vals
         return None
 
+def mld_wait_event(hapds, events, timeout):
+    start = os.times()[4]
+    while True:
+        for hapd in hapds:
+            ev = hapd.wait_event(events, 0)
+            if ev:
+                return ev
+
+        now = os.times()[4]
+        remaining = start + timeout - now
+        if remaining <= 0:
+            break
+        socks = [hapd.mon.socket for hapd in hapds]
+        r, w, e = select.select(socks, [], [], remaining)
+        if not r:
+            break
+
 def add_ap(apdev, params, wait_enabled=True, no_enable=False, timeout=30,
            global_ctrl_override=None, driver=False, set_channel=True):
         if isinstance(apdev, dict):
diff --git a/tests/hwsim/test_eht.py b/tests/hwsim/test_eht.py
index aa11765a6a..e3ce7da149 100644
--- a/tests/hwsim/test_eht.py
+++ b/tests/hwsim/test_eht.py
@@ -1249,6 +1249,101 @@ def test_eht_mld_bss_trans_mgmt_link_removal_imminent(dev, apdev):
         if ev is not None:
             raise Exception("Unexpected action on STA: " + ev)
 
+def test_eht_mld_bss_trans_mgmt_ml_neighbor(dev, apdev):
+    """EHT MLD with two links. BSS transition management with ML neighbor"""
+
+    with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \
+        HWSimRadio(use_mlo=True) as (hapd1_radio, hapd1_iface), \
+        HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface):
+
+        # This test does a HT scan on an MLD. Check kernel scanning support
+        buf = subprocess.Popen(['iw', 'phy', f'phy{hapd1_radio}', 'info'],
+                               stdout=subprocess.PIPE).communicate()[0]
+        if b'Device supports AP scan.' not in buf:
+            raise HwsimSkip('Kernel does not support AP scanning')
+
+        try:
+            wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
+            wpas.interface_add(wpas_iface)
+
+            ssid = "mld_ap_owe"
+            params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2")
+
+            # Enable BSS transition management support
+            params['mbo'] = '1'
+
+            # Start first MLD AP
+            hapd0_0 = eht_mld_enable_ap(hapd0_iface, 0, params)
+            params['channel'] = '6'
+            hapd0_1 = eht_mld_enable_ap(hapd0_iface, 1, params)
+
+            # Start second MLD AP, with 40MHz bw to make it more interesting
+            params_5ghz = eht_5ghz_params(36, 0, 38, 0)
+            params.update(params_5ghz)
+            hapd1_0 = eht_mld_enable_ap(hapd1_iface, 0, params)
+
+            params_5ghz = eht_5ghz_params(149, 1, 155, 0)
+            params.update(params_5ghz)
+            hapd1_1 = eht_mld_enable_ap(hapd1_iface, 1, params)
+
+            # Give APs some time to settle
+            time.sleep(1)
+
+            # Connect to and make sure we are on hapd0
+            wpas.connect(ssid, key_mgmt="OWE", ieee80211w="2")
+            wpas.roam(hapd0_0.own_addr(), check_bssid=False)
+
+            eht_verify_status(wpas, hapd0_0, 2412, 20, is_ht=True, mld=True,
+                              valid_links=3, active_links=3)
+
+            # Neighbor for hapd_1_0 with preference 0x00 and for the entire MLD
+            neighbor = hapd1_0.own_addr() + ',0x0000,' + '116,36,9,030100'
+            # ML neighbor report, control, common info with MLD addr
+            neighbor += 'c909000007' + hapd1_0.own_mld_addr().replace(':', '')
+
+            hapd0_0.request("BSS_TM_REQ " + wpas.own_addr() + ' pref=1 abridged=0 disassoc_imminent=1 valid_int=255 neighbor=' + neighbor)
+
+            # The station may respond on either link
+            ev = hostapd.mld_wait_event((hapd0_0, hapd0_1), ['BSS-TM-RESP'], timeout=20)
+            if ev is None:
+                raise Exception("No BSS Transition Management Response")
+            # Should be accepted, but the client has no where to roam
+            if 'status_code=0' not in ev or 'target_bssid=00:00:00:00:00:00' not in ev:
+                raise Exception("Unexpected BSS TM response status: " + ev)
+
+            # Give it a bit of time, we shouldn't roam as the other AP is forbidden
+            time.sleep(5)
+
+            # Didn't connect to the other AP
+            assert wpas.get_status()['ap_mld_addr'] == hapd0_0.own_mld_addr(), \
+                'STA should still be connected to same AP'
+
+            eht_verify_status(wpas, hapd0_0, 2412, 20, is_ht=True, mld=True,
+                              valid_links=3, active_links=3)
+
+            # Similar, but now ask to roam to the better AP
+            neighbor = hapd1_0.own_addr() + ',0x0000,' + '116,36,9,0301ff'
+            neighbor += 'c909000007' + hapd1_0.own_mld_addr().replace(':', '')
+
+            hapd0_0.request("BSS_TM_REQ " + wpas.own_addr() + ' pref=1 abridged=1 disassoc_imminent=1 valid_int=255  neighbor=' + neighbor)
+
+            # The station may respond on either link
+            ev = hostapd.mld_wait_event((hapd0_0, hapd0_1), ['BSS-TM-RESP'], timeout=20)
+            if ev is None:
+                raise Exception("No BSS Transition Management Response")
+            # Client accepts and roams to one the second AP
+            if 'status_code=0' not in ev:
+                raise Exception("Unexpected BSS TM response status: " + ev)
+
+            wpas.wait_connected(timeout=2)
+
+            eht_verify_status(wpas, hapd1_0, 5, 40, is_ht=True, is_vht=True,
+                              mld=True, valid_links=3, active_links=3)
+
+        finally:
+            wpas.cmd_execute(['iw', 'reg', 'set', '00'])
+            wait_regdom_changes(wpas)
+
 def send_check(hapd, frame, no_tx_status=False):
         cmd = "MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame="
         hapd.request(cmd + frame)
diff --git a/wpaspy/wpaspy.py b/wpaspy/wpaspy.py
index 5b8140b7c9..46862edc95 100644
--- a/wpaspy/wpaspy.py
+++ b/wpaspy/wpaspy.py
@@ -63,6 +63,10 @@ class Ctrl:
                 raise
         self.started = True
 
+    @property
+    def socket(self):
+        return self.s
+
     def __del__(self):
         self.close()
 
-- 
2.50.0




More information about the Hostap mailing list