tests: More coverage for WMM-AC

Signed-off-by: Jouni Malinen <jouni@qca.qualcomm.com>

tests/hwsim/test_wpas_wmm_ac.py

@@ -6,6 +6,7 @@
 
 import logging
 logger = logging.getLogger()
+import struct
 
 import hwsim_utils
 import hostapd
@@ -41,6 +42,16 @@ def test_tspec(dev, apdev):
     if "VO: acm=1 uapsd=0" not in status:
         raise Exception("Unexpected VO info" + status)
 
+    # no tsid --> tsid out of range
+    if "FAIL" not in dev[0].request("WMM_AC_ADDTS downlink"):
+        raise Exception("Invalid WMM_AC_ADDTS accepted")
+    # no direction
+    if "FAIL" not in dev[0].request("WMM_AC_ADDTS tsid=5"):
+        raise Exception("Invalid WMM_AC_ADDTS accepted")
+    # param out of range
+    if "FAIL" not in dev[0].request("WMM_AC_ADDTS tsid=5 downlink"):
+        raise Exception("Invalid WMM_AC_ADDTS accepted")
+
     tsid = 5
 
     # make sure we fail when the ac is not configured for acm
@@ -69,8 +80,168 @@ def test_tspec(dev, apdev):
             raise
 
     # update the tspec for a different UP of the same ac
-    dev[0].add_ts(tsid, 7)
+    dev[0].add_ts(tsid, 7, extra="fixed_nominal_msdu")
     dev[0].del_ts(tsid)
     status = dev[0].request("WMM_AC_STATUS")
     if "TSID" in status:
         raise Exception("Unexpected TSID info")
+
+    # verify failure on uplink/bidi without driver support
+    tsid = 6
+    try:
+        dev[0].add_ts(tsid, 7, direction="uplink")
+        raise Exception("ADDTS succeeded although it should have failed")
+    except Exception, e:
+        if not str(e).startswith("ADDTS failed"):
+            raise
+    try:
+        dev[0].add_ts(tsid, 7, direction="bidi")
+        raise Exception("ADDTS succeeded although it should have failed")
+    except Exception, e:
+        if not str(e).startswith("ADDTS failed"):
+            raise
+
+    # attempt to delete non-existing tsid
+    try:
+        dev[0].del_ts(tsid)
+        raise Exception("DELTS succeeded although it should have failed")
+    except Exception, e:
+        if not str(e).startswith("DELTS failed"):
+            raise
+
+def test_tspec_protocol(dev, apdev):
+    """Protocol tests for addts/delts"""
+    # configure ap with VO and VI requiring admission-control
+    hapd = add_wmm_ap(apdev, ["VO", "VI"])
+    dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462")
+
+    dev[0].dump_monitor()
+    hapd.set("ext_mgmt_frame_handling", "1")
+
+    tsid = 6
+
+    # timeout on ADDTS response
+    dev[0].add_ts(tsid, 7, expect_failure=True)
+
+    hapd.dump_monitor()
+    req = "WMM_AC_ADDTS downlink tsid=6 up=7 nominal_msdu_size=1500 sba=9000 mean_data_rate=1500 min_phy_rate=6000000"
+    if "OK" not in dev[0].request(req):
+        raise Exception("WMM_AC_ADDTS failed")
+    # a new request while previous is still pending
+    if "FAIL" not in dev[0].request(req):
+        raise Exception("WMM_AC_ADDTS accepted while oen was still pending")
+    msg = hapd.mgmt_rx()
+    payload = msg['payload']
+    (categ, action, dialog, status) = struct.unpack('BBBB', payload[0:4])
+    if action != 0:
+        raise Exception("Unexpected Action code: %d" % action)
+
+    msg['da'] = msg['sa']
+    msg['sa'] = apdev[0]['bssid']
+
+    # unexpected dialog token
+    msg['payload'] = struct.pack('BBBB', 17, 1, (dialog + 1) & 0xff, 0) + payload[4:]
+    hapd.mgmt_tx(msg)
+
+    # valid response
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:]
+    hapd.mgmt_tx(msg)
+    ev = dev[0].wait_event(["TSPEC-ADDED"], timeout=10)
+    if ev is None:
+        raise Exception("Timeout on TSPEC-ADDED")
+    if "tsid=%d" % tsid not in ev:
+        raise Exception("Unexpected TSPEC-ADDED contents: " + ev)
+
+    # duplicated response
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:]
+    hapd.mgmt_tx(msg)
+
+    # too short ADDTS
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0)
+    hapd.mgmt_tx(msg)
+
+    # invalid IE
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:] + struct.pack('BB', 0xdd, 100)
+    hapd.mgmt_tx(msg)
+
+    # too short WMM element
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:] + '\xdd\x06\x00\x50\xf2\x02\x02\x01'
+    hapd.mgmt_tx(msg)
+
+    # DELTS
+    dev[0].dump_monitor()
+    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0) + payload[4:]
+    hapd.mgmt_tx(msg)
+    ev = dev[0].wait_event(['TSPEC-REMOVED'], timeout=6)
+    if ev is None:
+        raise Exception("Timeout on TSPEC-REMOVED event")
+    if "tsid=%d" % tsid not in ev:
+        raise Exception("Unexpected TSPEC-REMOVED contents: " + ev)
+    # DELTS duplicated
+    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0) + payload[4:]
+    hapd.mgmt_tx(msg)
+
+    # start a new request
+    hapd.dump_monitor()
+    if "OK" not in dev[0].request(req):
+        raise Exception("WMM_AC_ADDTS failed")
+    msg = hapd.mgmt_rx()
+    payload = msg['payload']
+    (categ, action, dialog, status) = struct.unpack('BBBB', payload[0:4])
+    if action != 0:
+        raise Exception("Unexpected Action code: %d" % action)
+
+    msg['da'] = msg['sa']
+    msg['sa'] = apdev[0]['bssid']
+
+    # modified parameters
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 1) + payload[4:12] + struct.pack('B', ord(payload[12]) & ~0x60) + payload[13:]
+    hapd.mgmt_tx(msg)
+
+    # reject request
+    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 1) + payload[4:]
+    hapd.mgmt_tx(msg)
+    ev = dev[0].wait_event(["TSPEC-REQ-FAILED"], timeout=10)
+    if ev is None:
+        raise Exception("Timeout on TSPEC-REQ-FAILED")
+    if "tsid=%d" % tsid not in ev:
+        raise Exception("Unexpected TSPEC-REQ-FAILED contents: " + ev)
+
+    hapd.set("ext_mgmt_frame_handling", "0")
+
+def test_tspec_not_enabled(dev, apdev):
+    """addts failing if AP does not support WMM"""
+    params = { "ssid": "wmm_no_ac",
+               "hw_mode": "g",
+               "channel": "11",
+               "wmm_enabled" : "0" }
+    hapd = hostapd.add_ap(apdev[0]['ifname'], params)
+    dev[0].connect("wmm_no_ac", key_mgmt="NONE", scan_freq="2462")
+    status = dev[0].request("WMM_AC_STATUS")
+    if "Not associated to a WMM AP, WMM AC is Disabled" not in status:
+        raise Exception("Unexpected WMM_AC_STATUS: " + status)
+
+    try:
+        dev[0].add_ts(5, 6)
+        raise Exception("ADDTS succeeded although it should have failed")
+    except Exception, e:
+        if not str(e).startswith("ADDTS failed"):
+            raise
+
+    # attempt to delete non-existing tsid
+    try:
+        dev[0].del_ts(5)
+        raise Exception("DELTS succeeded although it should have failed")
+    except Exception, e:
+        if not str(e).startswith("DELTS failed"):
+            raise
+
+    # unexpected Action frame when WMM is disabled
+    MGMT_SUBTYPE_ACTION = 13
+    msg = {}
+    msg['fc'] = MGMT_SUBTYPE_ACTION << 4
+    msg['da'] = dev[0].p2p_interface_addr()
+    msg['sa'] = apdev[0]['bssid']
+    msg['bssid'] = apdev[0]['bssid']
+    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0)
+    hapd.mgmt_tx(msg)

tests/hwsim/wpasupplicant.py

@@ -721,20 +721,31 @@ class WpaSupplicant:
             raise Exception("Failed to request TDLS teardown")
         return None
 
-    def add_ts(self, tsid, up):
+    def add_ts(self, tsid, up, direction="downlink", expect_failure=False,
+               extra=None):
         params = {
             "sba": 9000,
             "nominal_msdu_size": 1500,
             "min_phy_rate": 6000000,
             "mean_data_rate": 1500,
         }
-        cmd = "WMM_AC_ADDTS downlink tsid=%d up=%d" % (tsid, up)
+        cmd = "WMM_AC_ADDTS %s tsid=%d up=%d" % (direction, tsid, up)
         for (key, value) in params.iteritems():
             cmd += " %s=%d" % (key, value)
+        if extra:
+            cmd += " " + extra
 
         if self.request(cmd).strip() != "OK":
             raise Exception("ADDTS failed (tsid=%d up=%d)" % (tsid, up))
 
+        if expect_failure:
+            ev = self.wait_event(["TSPEC-REQ-FAILED"], timeout=2)
+            if ev is None:
+                raise Exception("ADDTS failed (time out while waiting failure)")
+            if "tsid=%d" % (tsid) not in ev:
+                raise Exception("ADDTS failed (invalid tsid in TSPEC-REQ-FAILED")
+            return
+
         ev = self.wait_event(["TSPEC-ADDED"], timeout=1)
         if ev is None:
             raise Exception("ADDTS failed (time out)")