[PATCH v3 8/8] tests: uhr: add two tests for DBE

[Johannes Berg]

From: Johannes Berg <johannes.berg at intel.com>

Add two simple tests for DBE with 80 MHz instead of the
EHT 40 MHz, one where it gets used by a UHR client and
one where it doesn't get used by a non-UHR client.

Signed-off-by: Johannes Berg <johannes.berg at intel.com>
---
 tests/hwsim/test_uhr.py | 116 ++++++++++++++++++++++++++++++++++++++--
 1 file changed, 113 insertions(+), 3 deletions(-)

diff --git a/tests/hwsim/test_uhr.py b/tests/hwsim/test_uhr.py
index 4917bcc6ff2d..2127a8b7924e 100644
--- a/tests/hwsim/test_uhr.py
+++ b/tests/hwsim/test_uhr.py
@@ -15,7 +15,7 @@ from utils import *
 from hwsim import HWSimRadio
 import hwsim_utils
 from wpasupplicant import WpaSupplicant
-from test_eht import eht_verify_status, traffic_test
+from test_eht import eht_verify_status, traffic_test, eht_verify_wifi_version
 
 def uhr_verify_wifi_version(dev):
     status = dev.get_status()
@@ -26,7 +26,7 @@ def uhr_verify_wifi_version(dev):
     if status['wifi_generation'] != "8":
         raise Exception("Unexpected wifi_generation value: " + status['wifi_generation'])
 
-def uhr_verify_status(wpas, hapd, is_ht=False, is_vht=False):
+def uhr_verify_status(wpas, hapd, is_ht=False, is_vht=False, sta_expect_uhr=True):
     status = hapd.get_status()
 
     logger.info("hostapd STATUS: " + str(status))
@@ -54,7 +54,10 @@ def uhr_verify_status(wpas, hapd, is_ht=False, is_vht=False):
     if "[EHT]" not in sta['flags']:
         raise Exception("Missing STA flag: EHT")
     if "[UHR]" not in sta['flags']:
-        raise Exception("Missing STA flag: UHR")
+        if sta_expect_uhr:
+            raise Exception("Missing STA flag: UHR")
+    elif not sta_expect_uhr:
+        raise Exception("Erroneous STA flag: UHR")
 
 def test_uhr_open(dev, apdev):
     """UHR AP with open mode configuration"""
@@ -166,3 +169,110 @@ def run_uhr_mld_sae_single_link(dev, apdev, anti_clogging_token=False):
 def test_uhr_mld_sae_single_link(dev, apdev):
     """UHR MLD AP with MLD client SAE H2E connection using single link"""
     run_uhr_mld_sae_single_link(dev, apdev)
+
+def uhr_5ghz_params(channel, chanwidth, ccfs1, ccfs2=0,
+                    he_ccfs1=None, he_oper_chanwidth=None):
+    if he_ccfs1 is None:
+        he_ccfs1 = ccfs1
+    if he_oper_chanwidth is None:
+        he_oper_chanwidth = chanwidth
+
+    params = {"country_code": "US",
+              "hw_mode": "a",
+              "channel": str(channel),
+              "ieee80211n": "1",
+              "ieee80211ac": "1",
+              "ieee80211ax": "1",
+              "ieee80211be": "1",
+              "ieee80211bn": "1",
+              "vht_oper_chwidth": str(he_oper_chanwidth),
+              "vht_oper_centr_freq_seg0_idx": str(he_ccfs1),
+              "vht_oper_centr_freq_seg1_idx": str(ccfs2),
+              "he_oper_chwidth": str(he_oper_chanwidth),
+              "he_oper_centr_freq_seg0_idx": str(he_ccfs1),
+              "he_oper_centr_freq_seg1_idx": str(ccfs2),
+              "eht_oper_centr_freq_seg0_idx": str(ccfs1),
+              "eht_oper_chwidth": str(chanwidth)}
+
+    if he_oper_chanwidth == 0:
+        if channel < he_ccfs1:
+                params["ht_capab"] = "[HT40+]"
+        elif channel > he_ccfs1:
+                params["ht_capab"] = "[HT40-]"
+    else:
+        params["ht_capab"] = "[HT40+]"
+        if he_oper_chanwidth == 2:
+            params["vht_capab"] = "[VHT160]"
+        elif he_oper_chanwidth == 3:
+            params["vht_capab"] = "[VHT160-80PLUS80]"
+
+    return params
+
+def _test_uhr_5ghz(dev, apdev, channel, chanwidth, ccfs1, ccfs2=0,
+                   callback=None, **kw):
+    hapd = None
+    try:
+        params = uhr_5ghz_params(channel, chanwidth, ccfs1, ccfs2)
+        params['ssid'] = 'uhr'
+        params.update(kw)
+
+        freq = 5000 + channel * 5
+
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("uhr", key_mgmt="NONE", scan_freq=str(freq))
+        hapd.wait_sta()
+
+        uhr_verify_status(dev[0], hapd, is_ht=True, is_vht=True)
+        uhr_verify_wifi_version(dev[0])
+        hwsim_utils.test_connectivity(dev[0], hapd)
+        if callback: callback()
+    finally:
+        dev[0].request("DISCONNECT")
+        dev[0].wait_disconnected()
+        if hapd:
+            hapd.wait_sta_disconnect()
+        set_world_reg(apdev[0], None, dev[0])
+
+def test_uhr_5ghz_dbe_80(dev, apdev):
+    """UHR with DBE enabled"""
+    def check_80mhz():
+        # give the DBE enablement a bit of time
+        time.sleep(1)
+        p = subprocess.run(['iw', 'dev', dev[0].ifname, 'info'],
+                           capture_output=True, check=True)
+        assert b"channel 36 (5180 MHz), width: 80 MHz," in p.stdout, \
+            "client not connected with 80 MHz"
+    _test_uhr_5ghz(dev, apdev, 36, 0, 38, 0,
+                   callback=check_80mhz,
+                   dbe_bandwidth="2")
+
+def test_uhr_5ghz_dbe_80_not_used(dev, apdev):
+    """UHR with DBE enabled but client can't use it"""
+    hapd = None
+    try:
+        params = uhr_5ghz_params(36, 0, 38)
+        params['ssid'] = 'uhr'
+        params['dbe_bandwidth'] = '2'
+
+        freq = 5000 + 36 * 5
+
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("uhr", key_mgmt="NONE", scan_freq=str(freq),
+                       disable_uhr='1')
+        hapd.wait_sta()
+
+        uhr_verify_status(dev[0], hapd, is_ht=True, is_vht=True,
+                          sta_expect_uhr=False)
+        eht_verify_wifi_version(dev[0])
+        hwsim_utils.test_connectivity(dev[0], hapd)
+        time.sleep(1)
+        p = subprocess.run(['iw', 'dev', dev[0].ifname, 'info'],
+                           capture_output=True, check=True)
+        assert b"channel 36 (5180 MHz), width: 40 MHz," in p.stdout, \
+            "client not connected with 40 MHz"
+    finally:
+        dev[0].request("DISCONNECT")
+        dev[0].wait_disconnected()
+        if hapd:
+            hapd.wait_sta_disconnect()
+        set_world_reg(apdev[0], None, dev[0])
-- 
2.53.0