tests: Use wlantest to validate TDLS behavior

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

tests/hwsim/test_p2p_autogo.py

@@ -7,6 +7,7 @@
 # See README for more details.
 
 import time
+import subprocess
 import logging
 logger = logging.getLogger(__name__)
 
@@ -43,18 +44,63 @@ def test_autogo_2cli(dev):
     dev[1].remove_group()
     dev[0].remove_group()
 
+def wlantest_tdls(field, bssid, addr1, addr2):
+    res = subprocess.check_output(["../../wlantest/wlantest_cli",
+                                   "get_tdls_counter", field, bssid, addr1,
+                                   addr2]);
+    if "FAIL" in res:
+        raise Exception("wlantest_cli command failed")
+    return int(res)
+
+def wlantest_tdls_clear(bssid, addr1, addr2):
+    subprocess.call(["../../wlantest/wlantest_cli",
+                     "clear_tdls_counters", bssid, addr1, addr2]);
+
 def test_autogo_tdls(dev):
-    autogo(dev[0])
-    connect_cli(dev[0], dev[1])
-    connect_cli(dev[0], dev[2])
+    go = dev[0]
+    logger.info("Start autonomous GO with fixed parameters " + go.ifname)
+    id = go.add_network()
+    go.set_network_quoted(id, "ssid", "DIRECT-tdls")
+    go.set_network_quoted(id, "psk", "12345678")
+    go.set_network(id, "mode", "3")
+    go.set_network(id, "disabled", "2")
+    res = go.p2p_start_go(persistent=id)
+    logger.debug("res: " + str(res))
+    subprocess.call(["../../wlantest/wlantest_cli", "flush"]);
+    subprocess.call(["../../wlantest/wlantest_cli", "add_passphrase",
+                     "12345678"]);
+    connect_cli(go, dev[1])
+    connect_cli(go, dev[2])
     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
+    bssid = dev[0].p2p_interface_addr()
+    addr1 = dev[1].p2p_interface_addr()
     addr2 = dev[2].p2p_interface_addr()
     dev[1].tdls_setup(addr2)
     time.sleep(1)
     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
+    conf = wlantest_tdls("setup_conf_ok", bssid, addr1, addr2);
+    if conf == 0:
+        raise Exception("No TDLS Setup Confirm (success) seen")
+    dl = wlantest_tdls("valid_direct_link", bssid, addr1, addr2);
+    if dl == 0:
+        raise Exception("No valid frames through direct link")
+    wlantest_tdls_clear(bssid, addr1, addr2);
     dev[1].tdls_teardown(addr2)
     time.sleep(1)
+    teardown = wlantest_tdls("teardown", bssid, addr1, addr2);
+    if conf == teardown:
+        raise Exception("No TDLS Setup Teardown seen")
+    wlantest_tdls_clear(bssid, addr1, addr2);
     hwsim_utils.test_connectivity_p2p(dev[1], dev[2])
+    ap_path = wlantest_tdls("valid_ap_path", bssid, addr1, addr2);
+    if ap_path == 0:
+        raise Exception("No valid frames via AP path")
+    direct_link = wlantest_tdls("valid_direct_link", bssid, addr1, addr2);
+    if direct_link > 0:
+        raise Exception("Unexpected frames through direct link")
+    idirect_link = wlantest_tdls("invalid_direct_link", bssid, addr1, addr2);
+    if idirect_link > 0:
+        raise Exception("Unexpected frames through direct link (invalid)")
     dev[2].remove_group()
     dev[1].remove_group()
     dev[0].remove_group()

tests/hwsim/wpasupplicant.py

@@ -45,6 +45,30 @@ class WpaSupplicant:
         self.request("REMOVE_CRED *")
         self.group_ifname = None
 
+    def add_network(self):
+        id = self.request("ADD_NETWORK")
+        if "FAIL" in id:
+            raise Exception("ADD_NETWORK failed")
+        return int(id)
+
+    def remove_network(self, id):
+        id = self.request("REMOVE_NETWORK " + str(id))
+        if "FAIL" in id:
+            raise Exception("REMOVE_NETWORK failed")
+        return None
+
+    def set_network(self, id, field, value):
+        res = self.request("SET_NETWORK " + str(id) + " " + field + " " + value)
+        if "FAIL" in res:
+            raise Exception("SET_NETWORK failed")
+        return None
+
+    def set_network_quoted(self, id, field, value):
+        res = self.request("SET_NETWORK " + str(id) + " " + field + ' "' + value + '"')
+        if "FAIL" in res:
+            raise Exception("SET_NETWORK failed")
+        return None
+
     def get_status(self, field):
         res = self.request("STATUS")
         lines = res.splitlines()
@@ -204,9 +228,15 @@ class WpaSupplicant:
             raise Exception("Group could not be removed")
         self.group_ifname = None
 
-    def p2p_start_go(self):
+    def p2p_start_go(self, persistent=None):
         self.dump_monitor()
         cmd = "P2P_GROUP_ADD"
+        if persistent is None:
+            pass
+        elif persistent is True:
+            cmd = cmd + " persistent"
+        else:
+            cmd = cmd + " persistent=" + str(persistent)
         if "OK" in self.request(cmd):
             ev = self.wait_event(["P2P-GROUP-STARTED"], timeout=5)
             if ev is None: