tests: Add autonomous GO testing

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

tests/hwsim/run-p2p-tests.py

@@ -15,6 +15,7 @@ import logging
 from wpasupplicant import WpaSupplicant
 
 import test_p2p_grpform
+import test_p2p_autogo
 
 def main():
     idx = 1
@@ -46,6 +47,7 @@ def main():
 
     tests = []
     test_p2p_grpform.add_tests(tests)
+    test_p2p_autogo.add_tests(tests)
 
     passed = []
     failed = []

tests/hwsim/test_p2p_autogo.py

@@ -0,0 +1,35 @@
+#!/usr/bin/python
+#
+# P2P autonomous GO test cases
+# Copyright (c) 2013, Jouni Malinen <j@w1.fi>
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+import logging
+logger = logging.getLogger(__name__)
+
+import hwsim_utils
+
+def autogo(go, client):
+    logger.info("Start autonomous GO " + go.ifname)
+    res = go.p2p_start_go()
+    logger.debug("res: " + str(res))
+
+    logger.info("Try to connect the client to the GO")
+    pin = client.wps_read_pin()
+    go.p2p_go_authorize_client(pin)
+    client.p2p_connect_group(go.p2p_dev_addr(), pin, timeout=60)
+    logger.info("Group formed")
+    hwsim_utils.test_connectivity_p2p(go, client)
+
+def test_autogo(dev):
+    autogo(go=dev[0], client=dev[1])
+    dev[0].remove_group()
+    try:
+        dev[1].remove_group()
+    except:
+        pass
+
+def add_tests(tests):
+    tests.append(test_autogo)

tests/hwsim/wpasupplicant.py

@@ -180,3 +180,37 @@ class WpaSupplicant:
             ifname = self.ifname
         if "OK" not in self.request("P2P_GROUP_REMOVE " + ifname):
             raise Exception("Group could not be removed")
+
+    def p2p_start_go(self):
+        self.dump_monitor()
+        cmd = "P2P_GROUP_ADD"
+        if "OK" in self.request(cmd):
+            ev = self.wait_event(["P2P-GROUP-STARTED"], timeout=5)
+            if ev is None:
+                raise Exception("GO start up timed out")
+            self.dump_monitor()
+            return self.group_form_result(ev)
+        raise Exception("P2P_GROUP_ADD failed")
+
+    def p2p_go_authorize_client(self, pin):
+        cmd = "WPS_PIN any " + pin
+        if "FAIL" in self.request(cmd):
+            raise Exception("Failed to authorize client connection on GO")
+        return None
+
+    def p2p_connect_group(self, go_addr, pin, timeout=0):
+        self.dump_monitor()
+        if not self.discover_peer(go_addr):
+            raise Exception("GO " + go_addr + " not found")
+        self.dump_monitor()
+        cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
+        if "OK" in self.request(cmd):
+            if timeout == 0:
+                self.dump_monitor()
+                return None
+            ev = self.wait_event(["P2P-GROUP-STARTED"], timeout)
+            if ev is None:
+                raise Exception("Joining the group timed out")
+            self.dump_monitor()
+            return self.group_form_result(ev)
+        raise Exception("P2P_CONNECT(join) failed")