tests: Add test cases for automatic channel selection

This extends the Hostapd class to support monitor interface events and
STATUS command similarly to the WpaSupplicant class so that internal
hostapd state can be verified in more detail.

Signed-hostap: Jouni Malinen <j@w1.fi>
This commit is contained in:
Jouni Malinen 2013-11-03 20:50:39 +02:00
parent 77990cd7a9
commit b47750beba
3 changed files with 132 additions and 0 deletions

View file

@ -46,6 +46,8 @@ class Hostapd:
def __init__(self, ifname): def __init__(self, ifname):
self.ifname = ifname self.ifname = ifname
self.ctrl = wpaspy.Ctrl(os.path.join(hapd_ctrl, ifname)) self.ctrl = wpaspy.Ctrl(os.path.join(hapd_ctrl, ifname))
self.mon = wpaspy.Ctrl(os.path.join(hapd_ctrl, ifname))
self.mon.attach()
def request(self, cmd): def request(self, cmd):
logger.debug(self.ifname + ": CTRL: " + cmd) logger.debug(self.ifname + ": CTRL: " + cmd)
@ -109,6 +111,39 @@ class Hostapd:
if not "OK" in self.ctrl.request("ENABLE"): if not "OK" in self.ctrl.request("ENABLE"):
raise Exception("Failed to disable hostapd interface " + self.ifname) raise Exception("Failed to disable hostapd interface " + self.ifname)
def dump_monitor(self):
while self.mon.pending():
ev = self.mon.recv()
logger.debug(self.ifname + ": " + ev)
def wait_event(self, events, timeout):
count = 0
while count < timeout * 10:
count = count + 1
time.sleep(0.1)
while self.mon.pending():
ev = self.mon.recv()
logger.debug(self.ifname + ": " + ev)
for event in events:
if event in ev:
return ev
return None
def get_status(self):
res = self.request("STATUS")
lines = res.splitlines()
vals = dict()
for l in lines:
[name,value] = l.split('=', 1)
vals[name] = value
return vals
def get_status_field(self, field):
vals = self.get_status()
if field in vals:
return vals[field]
return None
def add_ap(ifname, params): def add_ap(ifname, params):
logger.info("Starting AP " + ifname) logger.info("Starting AP " + ifname)
hapd_global = HostapdGlobal() hapd_global = HostapdGlobal()
@ -133,6 +168,7 @@ def add_ap(ifname, params):
else: else:
hapd.set(f, v) hapd.set(f, v)
hapd.enable() hapd.enable()
return hapd
def add_bss(phy, ifname, confname, ignore_error=False): def add_bss(phy, ifname, confname, ignore_error=False):
logger.info("Starting BSS phy=" + phy + " ifname=" + ifname) logger.info("Starting BSS phy=" + phy + " ifname=" + ifname)

View file

@ -0,0 +1,28 @@
driver=nl80211
hw_mode=g
channel=0
ieee80211n=1
interface=wlan3
ctrl_interface=/var/run/hostapd
ssid=bss-1
bss=wlan3-2
bssid=02:00:00:00:03:01
ctrl_interface=/var/run/hostapd
ssid=bss-2
wpa=2
wpa_key_mgmt=WPA-PSK
rsn_pairwise=CCMP
wpa_passphrase=12345678
bss=wlan3-3
bssid=02:00:00:00:03:02
ctrl_interface=/var/run/hostapd
ssid=bss-3
wpa=1
wpa_key_mgmt=WPA-PSK
rsn_pairwise=TKIP
wpa_passphrase=qwertyuiop

View file

@ -0,0 +1,68 @@
#!/usr/bin/python
#
# Test cases for automatic channel selection with hostapd
# 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()
import hostapd
def wait_acs(hapd):
ev = hapd.wait_event(["ACS-STARTED", "ACS-COMPLETED", "ACS-FAILED",
"AP-ENABLED"], timeout=5)
if not ev:
raise Exception("ACS start timed out")
if "ACS-STARTED" not in ev:
raise Exception("Unexpected ACS event")
state = hapd.get_status_field("state")
if state != "ACS":
raise Exception("Unexpected interface state")
ev = hapd.wait_event(["ACS-COMPLETED", "ACS-FAILED", "AP-ENABLED"],
timeout=20)
if not ev:
raise Exception("ACS timed out")
if "ACS-COMPLETED" not in ev:
raise Exception("Unexpected ACS event")
ev = hapd.wait_event(["AP-ENABLED"], timeout=5)
if not ev:
raise Exception("AP setup timed out")
state = hapd.get_status_field("state")
if state != "ENABLED":
raise Exception("Unexpected interface state")
def test_ap_acs(dev, apdev):
"""Automatic channel selection"""
params = hostapd.wpa2_params(ssid="test-acs", passphrase="12345678")
params['channel'] = '0'
hapd = hostapd.add_ap(apdev[0]['ifname'], params)
wait_acs(hapd)
freq = hapd.get_status_field("freq")
if int(freq) < 2400:
raise Exception("Unexpected frequency")
dev[0].connect("test-acs", psk="12345678", scan_freq=freq)
def test_ap_multi_bss_acs(dev, apdev):
"""hostapd start with a multi-BSS configuration file using ACS"""
ifname = apdev[0]['ifname']
hostapd.add_iface(ifname, 'multi-bss-acs.conf')
hapd = hostapd.Hostapd(ifname)
hapd.enable()
wait_acs(hapd)
freq = hapd.get_status_field("freq")
if int(freq) < 2400:
raise Exception("Unexpected frequency")
dev[0].connect("bss-1", key_mgmt="NONE", scan_freq=freq)
dev[1].connect("bss-2", psk="12345678", scan_freq=freq)
dev[2].connect("bss-3", psk="qwertyuiop", scan_freq=freq)