[PATCH 4/4] tests: AP+STA on same radio

[Raphaël Mélotte]

From: "Arnout Vandecappelle (Essensium/Mind)" <arnout at mind.be>

Add a test for the support of an AP (hostapd) and a STA (wpa_supplicant)
running simultaneously on the same radio.

The test creates a new radio, with a managed and a station interface on
that radio. On the managed interface, hostapd is started in the usual
way. On the station interface, wpa_supplicant is started manually so we
can pass the -H argument. The station is connected to apdev[0]; dev[0]
is connected to the AP. Finally, a channel switch is started on
apdev[0]. This channel switch must propagate all the way down to dev[0].

Without the AP+STA functionality, the test fails in the following ways.

- Once AP2 is started on channel 11, it's not possible for wpas to
  connect to channel 1. For that, the AP has to be stopped first.
- When AP2 is started on channel 1, it's possible for wpas to connect.
  However, it cannot follow the CSA from AP1.

Signed-off-by: Arnout Vandecappelle (Essensium/Mind) <arnout at mind.be>
---
 tests/hwsim/test_ap_csa.py | 115 +++++++++++++++++++++++++++++++++++++
 1 file changed, 115 insertions(+)

diff --git a/tests/hwsim/test_ap_csa.py b/tests/hwsim/test_ap_csa.py
index 744d1e1f2..93bf6a892 100644
--- a/tests/hwsim/test_ap_csa.py
+++ b/tests/hwsim/test_ap_csa.py
@@ -9,9 +9,11 @@ import time
 import logging
 logger = logging.getLogger()
 
+from hwsim import HWSimRadio
 import hwsim_utils
 import hostapd
 from utils import *
+from wpasupplicant import WpaSupplicant
 
 def connect(dev, apdev, scan_freq="2412", **kwargs):
     params = {"ssid": "ap-csa",
@@ -187,3 +189,116 @@ def test_ap_csa_disable(dev, apdev):
     ap.enable()
     dev[0].wait_disconnected()
     dev[0].wait_connected()
+
+def write_hostapd_config(conffile, ifname, ssid):
+    with open(conffile, "w") as f:
+        f.write("driver=nl80211\n")
+        f.write("hw_mode=g\n")
+        f.write("channel=11\n")
+        f.write("ieee80211n=1\n")
+        f.write("ctrl_interface=/var/run/hostapd\n")
+        f.write("interface=" + ifname + "\n")
+        f.write("ssid=" + ssid + "\n")
+
+def test_ap_csa_1_switch_backhaul(dev, apdev, test_params):
+    """AP Channel Switch of backhaul with AP on same radio, one switch"""
+    csa_supported(dev[0])
+
+    params1 = {"ssid": "ap-csa-back",
+               "channel": "1"}
+    ap1 = hostapd.add_ap(apdev[0], params1)
+
+    with HWSimRadio() as (radio, ifname):
+        apifname2 = ifname + "_ap"
+        apdev2 = {"ifname": apifname2}
+
+        subprocess.call(['iw', 'dev', ifname, 'interface', 'add', apifname2,
+                         'type', 'managed'])
+
+        ap2 = None
+        try:
+            prefix = "ap_csa_1_switch_backhaul-ap"
+            conffile = os.path.join(test_params["logdir"], prefix + ".conf")
+            logfile = os.path.join(test_params["logdir"], prefix + ".log")
+            pidfile = os.path.join(test_params["logdir"], prefix + ".pid")
+            prg = os.path.join(test_params['logdir'], 'alt-hostapd/hostapd/hostapd')
+            if not os.path.exists(prg):
+                prg = '../../hostapd/hostapd'
+            write_hostapd_config(conffile, apifname2, "ap-csa-front")
+            arg = [prg, '-B', '-dddt', '-P', pidfile, '-f', logfile, conffile]
+            logger.info("Start hostapd: " + str(arg))
+            res = subprocess.check_call(arg)
+            if res != 0:
+                raise Exception("Could not start hostapd: %s" % str(res))
+            ap2 = hostapd.Hostapd(apifname2)
+            ap2.ping()
+
+            freq = int(ap2.get_status_field("freq"))
+            if freq != 2462:
+                raise Exception("Unexpected AP2 freq=%d before association" % freq)
+
+            prefix = "ap_csa_1_switch_backhaul-sta"
+            conffile = os.path.join(test_params["logdir"], prefix + ".conf")
+            logfile = os.path.join(test_params["logdir"], prefix + ".log")
+            pidfile2 = os.path.join(test_params["logdir"], prefix + ".pid")
+
+            with open(conffile, 'w') as f:
+                f.write("ctrl_interface=DIR=/var/run/wpa_supplicant\n")
+
+            prg = os.path.join(test_params['logdir'],
+                            'alt-wpa_supplicant/wpa_supplicant/wpa_supplicant')
+            if not os.path.exists(prg):
+                prg = '../../wpa_supplicant/wpa_supplicant'
+
+            arg = [prg, '-BddtK', '-P', pidfile2, '-f', logfile,
+                '-Dnl80211', '-c', conffile, '-i', ifname,
+                '-H', '/var/run/hostapd/' + apifname2]
+            logger.info("Start wpa_supplicant: " + str(arg))
+            subprocess.call(arg)
+            wpas = WpaSupplicant(ifname=ifname)
+            try:
+                csa_supported(wpas)
+
+                wpas.connect("ap-csa-back", key_mgmt="NONE", scan_freq="2412")
+                freq = int(wpas.get_driver_status_field("freq"))
+                if freq != 2412:
+                    raise Exception("Unexpected freq=%d after association" % freq)
+                hwsim_utils.test_connectivity(wpas, ap1)
+
+                freq = int(ap2.get_status_field("freq"))
+                if freq != 2412:
+                    raise Exception("Unexpected AP2 freq=%d after association" % freq)
+
+                dev[0].connect("ap-csa-front", key_mgmt="NONE", scan_freq="2412 2462")
+                freq = int(dev[0].get_status_field("freq"))
+                if freq != 2412:
+                    raise Exception("Unexpected freq=%d after association" % freq)
+                hwsim_utils.test_connectivity(dev[0], ap2)
+
+                switch_channel(ap1, 10, 2437)
+                wait_channel_switch(wpas, 2437)
+                wait_channel_switch(dev[0], 2437)
+
+                hwsim_utils.test_connectivity(wpas, ap1)
+                hwsim_utils.test_connectivity(dev[0], ap2)
+
+                for f in (ap1, wpas, ap2, dev[0]):
+                    freq = int(f.get_status_field("freq"))
+                    if freq != 2437:
+                        raise Exception("Unexpected driver freq=%d after channel switch" % freq)
+            finally:
+                wpas.close_monitor()
+                wpas.request("TERMINATE")
+        finally:
+            if ap2:
+                if "OK" not in ap2.request("TERMINATE"):
+                    raise Exception("Failed to terminate hostapd process")
+                ev = ap2.wait_event(["CTRL-EVENT-TERMINATING"], timeout=15)
+                if ev is None:
+                    raise Exception("CTRL-EVENT-TERMINATING not seen")
+            for i in range(30):
+                time.sleep(0.1)
+                if not os.path.exists(pidfile):
+                    break
+            if os.path.exists(pidfile):
+                raise Exception("PID file exits after process termination")
-- 
2.37.3