tests: Verify QoS Mapping results in Data frames

This verifies that IP packets with various DSCP values are mapped to the
correct TID both with default mapping and with custom QoS mappings.

Signed-hostap: Jouni Malinen <j@w1.fi>

diff --git a/tests/hwsim/hwsim_utils.py b/tests/hwsim/hwsim_utils.py
index 42856fe9ee4b4a5e6101042787817f9b2e011708..621143b051f49fc00b8e6b0cf57068831fa447ea 100644 (file)
--- a/tests/hwsim/hwsim_utils.py
+++ b/tests/hwsim/hwsim_utils.py
@@ -11,7 +11,7 @@ import subprocess
 import logging
 logger = logging.getLogger()
 
-def test_connectivity(ifname1, ifname2):
+def test_connectivity(ifname1, ifname2, dscp=None, tos=None):
     if os.path.isfile("../../mac80211_hwsim/tools/hwsim_test"):
         hwsim_test = "../../mac80211_hwsim/tools/hwsim_test"
     else:
@@ -20,6 +20,12 @@ def test_connectivity(ifname1, ifname2):
            hwsim_test,
            ifname1,
            ifname2]
+    if dscp:
+        cmd.append('-D')
+        cmd.append(str(dscp))
+    elif tos:
+        cmd.append('-t')
+        cmd.append(str(tos))
     try:
         s = subprocess.check_output(cmd)
         logger.debug(s)
@@ -28,17 +34,17 @@ def test_connectivity(ifname1, ifname2):
         logger.info(e.output)
         raise
 
-def test_connectivity_p2p(dev1, dev2):
+def test_connectivity_p2p(dev1, dev2, dscp=None, tos=None):
     ifname1 = dev1.group_ifname if dev1.group_ifname else dev1.ifname
     ifname2 = dev2.group_ifname if dev2.group_ifname else dev2.ifname
-    test_connectivity(ifname1, ifname2)
+    test_connectivity(ifname1, ifname2, dscp, tos)
 
-def test_connectivity_p2p_sta(dev1, dev2):
+def test_connectivity_p2p_sta(dev1, dev2, dscp=None, tos=None):
     ifname1 = dev1.group_ifname if dev1.group_ifname else dev1.ifname
     ifname2 = dev2.ifname
-    test_connectivity(ifname1, ifname2)
+    test_connectivity(ifname1, ifname2, dscp, tos)
 
-def test_connectivity_sta(dev1, dev2):
+def test_connectivity_sta(dev1, dev2, dscp=None, tos=None):
     ifname1 = dev1.ifname
     ifname2 = dev2.ifname
-    test_connectivity(ifname1, ifname2)
+    test_connectivity(ifname1, ifname2, dscp, tos)

diff --git a/tests/hwsim/test_ap_qosmap.py b/tests/hwsim/test_ap_qosmap.py
index 6d0ecf23bddca136dc65c4f907c833fdcc5b8214..c67f6ad10687d6971e282197e8acfb81532f3591 100644 (file)
--- a/tests/hwsim/test_ap_qosmap.py
+++ b/tests/hwsim/test_ap_qosmap.py
@@ -13,6 +13,21 @@ logger = logging.getLogger()
 
 import hwsim_utils
 import hostapd
+from wlantest import Wlantest
+
+def check_qos_map(ap, dev, dscp, tid):
+    bssid = ap['bssid']
+    sta = dev.p2p_interface_addr()
+    wt = Wlantest()
+    wt.clear_sta_counters(bssid, sta)
+    hwsim_utils.test_connectivity(dev.ifname, ap['ifname'], dscp=dscp)
+    [ tx, rx ] = wt.get_tid_counters(bssid, sta)
+    if tx[tid] == 0:
+        logger.info("Expected TX DSCP " + str(dscp) + " with TID " + str(tid) + " but counters: " + str(tx))
+        raise Exception("No STA->AP data frame using the expected TID")
+    if rx[tid] == 0:
+        logger.info("Expected RX DSCP " + str(dscp) + " with TID " + str(tid) + " but counters: " + str(rx))
+        raise Exception("No AP->STA data frame using the expected TID")
 
 def test_ap_qosmap(dev, apdev):
     """QoS mapping"""
@@ -21,11 +36,38 @@ def test_ap_qosmap(dev, apdev):
         return "skip"
     ssid = "test-qosmap"
     params = { "ssid": ssid }
-    params['qos_map_set'] = '53,2,22,6,8,15,0,7,255,255,16,31,32,39,255,255,40,47,255,255'
+    params['qos_map_set'] = '53,2,22,6,8,15,0,7,255,255,16,31,32,39,255,255,40,47,48,55'
     hostapd.add_ap(apdev[0]['ifname'], params)
     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
-    hwsim_utils.test_connectivity(dev[0].ifname, apdev[0]['ifname'])
+    check_qos_map(apdev[0], dev[0], 53, 2)
+    check_qos_map(apdev[0], dev[0], 22, 6)
+    check_qos_map(apdev[0], dev[0], 8, 0)
+    check_qos_map(apdev[0], dev[0], 15, 0)
+    check_qos_map(apdev[0], dev[0], 0, 1)
+    check_qos_map(apdev[0], dev[0], 7, 1)
+    check_qos_map(apdev[0], dev[0], 16, 3)
+    check_qos_map(apdev[0], dev[0], 31, 3)
+    check_qos_map(apdev[0], dev[0], 32, 4)
+    check_qos_map(apdev[0], dev[0], 39, 4)
+    check_qos_map(apdev[0], dev[0], 40, 6)
+    check_qos_map(apdev[0], dev[0], 47, 6)
+    check_qos_map(apdev[0], dev[0], 48, 7)
+    check_qos_map(apdev[0], dev[0], 55, 7)
     hapd = hostapd.Hostapd(apdev[0]['ifname'])
-    hapd.request("SET_QOS_MAP_SET 22,6,8,15,0,7,255,255,16,31,32,39,255,255,40,47,255,255")
+    hapd.request("SET_QOS_MAP_SET 22,6,8,15,0,7,255,255,16,31,32,39,255,255,40,47,48,55")
     hapd.request("SEND_QOS_MAP_CONF " + dev[0].get_status_field("address"))
-    hwsim_utils.test_connectivity(dev[0].ifname, apdev[0]['ifname'])
+    check_qos_map(apdev[0], dev[0], 53, 7)
+    check_qos_map(apdev[0], dev[0], 22, 6)
+    check_qos_map(apdev[0], dev[0], 48, 7)
+    check_qos_map(apdev[0], dev[0], 55, 7)
+    check_qos_map(apdev[0], dev[0], 56, 56 >> 3)
+    check_qos_map(apdev[0], dev[0], 63, 63 >> 3)
+
+def test_ap_qosmap_default(dev, apdev):
+    """QoS mapping with default values"""
+    ssid = "test-qosmap-default"
+    params = { "ssid": ssid }
+    hostapd.add_ap(apdev[0]['ifname'], params)
+    dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
+    for dscp in [ 0, 7, 8, 15, 16, 23, 24, 31, 32, 39, 40, 47, 48, 55, 56, 63]:
+        check_qos_map(apdev[0], dev[0], dscp, dscp >> 3)

diff --git a/tests/hwsim/wlantest.py b/tests/hwsim/wlantest.py
index c6dba12a7dc977256089deed5e9c52d0894adc3e..57853fb41919588d8a961f8ffb5216b6b4c9001d 100644 (file)
--- a/tests/hwsim/wlantest.py
+++ b/tests/hwsim/wlantest.py
@@ -63,6 +63,12 @@ class Wlantest:
             raise Exception("wlantest_cli command failed")
         return int(res)
 
+    def clear_sta_counters(self, bssid, addr):
+        res = subprocess.check_output([self.wlantest_cli, "clear_sta_counters",
+                                       bssid, addr]);
+        if "FAIL" in res:
+            raise Exception("wlantest_cli command failed")
+
     def tdls_clear(self, bssid, addr1, addr2):
         res = subprocess.check_output([self.wlantest_cli, "clear_tdls_counters",
                                        bssid, addr1, addr2]);
@@ -114,3 +120,25 @@ class Wlantest:
         res = self.info_sta("key_mgmt", bssid, addr)
         if key_mgmt not in res:
             raise Exception("Unexpected STA key_mgmt")
+
+    def get_tx_tid(self, bssid, addr, tid):
+        res = subprocess.check_output([self.wlantest_cli, "get_tx_tid",
+                                       bssid, addr, str(tid)]);
+        if "FAIL" in res:
+            raise Exception("wlantest_cli command failed")
+        return int(res)
+
+    def get_rx_tid(self, bssid, addr, tid):
+        res = subprocess.check_output([self.wlantest_cli, "get_rx_tid",
+                                       bssid, addr, str(tid)]);
+        if "FAIL" in res:
+            raise Exception("wlantest_cli command failed")
+        return int(res)
+
+    def get_tid_counters(self, bssid, addr):
+        tx = {}
+        rx = {}
+        for tid in range(0, 17):
+            tx[tid] = self.get_tx_tid(bssid, addr, tid)
+            rx[tid] = self.get_rx_tid(bssid, addr, tid)
+        return [ tx, rx ]