| # Test cases for wpa_supplicant WMM-AC operations |
| # Copyright (c) 2014, Intel Corporation |
| # |
| # This software may be distributed under the terms of the BSD license. |
| # See README for more details. |
| |
| import logging |
| logger = logging.getLogger() |
| import struct |
| |
| import hwsim_utils |
| import hostapd |
| |
| def add_wmm_ap(apdev, acm_list): |
| params = { "ssid": "wmm_ac", |
| "hw_mode": "g", |
| "channel": "11", |
| "wmm_enabled" : "1"} |
| |
| for ac in acm_list: |
| params["wmm_ac_%s_acm" % (ac.lower())] = "1" |
| |
| return hostapd.add_ap(apdev['ifname'], params) |
| |
| def test_tspec(dev, apdev): |
| """Basic addts/delts tests""" |
| # configure ap with VO and VI requiring admission-control |
| hapd = add_wmm_ap(apdev[0], ["VO", "VI"]) |
| dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462") |
| hwsim_utils.test_connectivity(dev[0], hapd) |
| status = dev[0].request("WMM_AC_STATUS") |
| if "WMM AC is Enabled" not in status: |
| raise Exception("WMM-AC not enabled") |
| if "TSID" in status: |
| raise Exception("Unexpected TSID info") |
| if "BK: acm=0 uapsd=0" not in status: |
| raise Exception("Unexpected BK info" + status) |
| if "BE: acm=0 uapsd=0" not in status: |
| raise Exception("Unexpected BE info" + status) |
| if "VI: acm=1 uapsd=0" not in status: |
| raise Exception("Unexpected VI info" + status) |
| 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 |
| try: |
| dev[0].add_ts(tsid, 3) |
| raise Exception("ADDTS succeeded although it should have failed") |
| except Exception, e: |
| if not str(e).startswith("ADDTS failed"): |
| raise |
| status = dev[0].request("WMM_AC_STATUS") |
| if "TSID" in status: |
| raise Exception("Unexpected TSID info") |
| |
| # add tspec for UP=6 |
| dev[0].add_ts(tsid, 6) |
| status = dev[0].request("WMM_AC_STATUS") |
| if "TSID" not in status: |
| raise Exception("Missing TSID info") |
| |
| # using the same tsid for a different ac is invalid |
| try: |
| dev[0].add_ts(tsid, 5) |
| raise Exception("ADDTS succeeded although it should have failed") |
| except Exception, e: |
| if not str(e).startswith("ADDTS failed"): |
| raise |
| |
| # update the tspec for a different UP of the same ac |
| 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[0], ["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) |
| |
| def test_tspec_ap_roam_open(dev, apdev): |
| """Roam between two open APs while having tspecs""" |
| hapd0 = add_wmm_ap(apdev[0], ["VO", "VI"]) |
| dev[0].connect("wmm_ac", key_mgmt="NONE") |
| hwsim_utils.test_connectivity(dev[0], hapd0) |
| dev[0].add_ts(5, 6) |
| |
| hapd1 = add_wmm_ap(apdev[1], ["VO", "VI"]) |
| dev[0].scan_for_bss(apdev[1]['bssid'], freq=2462) |
| dev[0].roam(apdev[1]['bssid']) |
| hwsim_utils.test_connectivity(dev[0], hapd1) |
| if dev[0].tspecs(): |
| raise Exception("TSPECs weren't deleted on roaming") |
| |
| dev[0].scan_for_bss(apdev[0]['bssid'], freq=2462) |
| dev[0].roam(apdev[0]['bssid']) |
| hwsim_utils.test_connectivity(dev[0], hapd0) |
| |
| def test_tspec_reassoc(dev, apdev): |
| """Reassociation to same BSS while having tspecs""" |
| hapd0 = add_wmm_ap(apdev[0], ["VO", "VI"]) |
| dev[0].connect("wmm_ac", key_mgmt="NONE") |
| hwsim_utils.test_connectivity(dev[0], hapd0) |
| dev[0].add_ts(5, 6) |
| last_tspecs = dev[0].tspecs() |
| |
| dev[0].request("REASSOCIATE") |
| dev[0].wait_connected() |
| |
| hwsim_utils.test_connectivity(dev[0], hapd0) |
| if dev[0].tspecs() != last_tspecs: |
| raise Exception("TSPECs weren't saved on reassociation") |
