tests/remote: Add hwsim wrapper

This allow to run hwsim test cases.

duts go to apdev while refs go to dev

For now I tested:
./run-tests.py -d hwsim0 -r hwsim1 -h ap_open -h dfs
./run-tests.py -r hwsim0 -r hwsim1 -h ibss_open -v
./run-tests.py -r hwsim0 -r hwsim1 -r hwsim2 -d hwsim3 -d hwsim4 -h ap_vht80 -v
./run-tests.py -r hwsim0 -r hwsim1 -r hwsim2 -d hwsim3 -d hwsim4 -h all -k ap -k vht

Signed-off-by: Janusz Dziedzic <janusz.dziedzic@tieto.com>
This commit is contained in:
Janusz Dziedzic 2016-04-29 07:07:37 +02:00 committed by Jouni Malinen
parent ff9bb8a210
commit a185e9b10b
3 changed files with 190 additions and 5 deletions

View file

@ -42,7 +42,8 @@ setup_params = { "setup_hw" : "./tests/setup_hw.sh",
devices = [{"hostname": "localhost", "ifname": "wlan0", "port": "9868", "name": "hwsim0", "flags": "AP_VHT80 STA_VHT80"},
{"hostname": "localhost", "ifname": "wlan1", "port": "9878", "name": "hwsim1", "flags": "AP_VHT80 STA_VHT80"},
{"hostname": "localhost", "ifname": "wlan2", "port": "9888", "name": "hwsim2", "flags": "AP_VHT80 STA_VHT80"},
{"hostname": "localhost", "ifname": "wlan3", "port": "9898", "name": "hwsim3", "flags": "AP_VHT80 STA_VHT80"}]
{"hostname": "localhost", "ifname": "wlan3", "port": "9898", "name": "hwsim3", "flags": "AP_VHT80 STA_VHT80"},
{"hostname": "localhost", "ifname": "wlan4", "port": "9908", "name": "hwsim4", "flags": "AP_VHT80 STA_VHT80"}]
def get_setup_params(filename="cfg.py"):

View file

@ -0,0 +1,107 @@
# Hwsim wrapper
# Copyright (c) 2016, Tieto Corporation
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.
import remotehost
from wpasupplicant import WpaSupplicant
import hostapd
import config
import rutils
import monitor
import traceback
import logging
logger = logging.getLogger()
def run_hwsim_test(devices, setup_params, refs, duts, monitors, hwsim_test):
try:
ref_hosts = []
dut_hosts = []
dev = []
apdev = []
# get hosts
for ref in refs:
ref_host = rutils.get_host(devices, ref)
ref_hosts.append(ref_host)
for dut in duts:
dut_host = rutils.get_host(devices, dut)
dut_hosts.append(dut_host)
# setup log dir
local_log_dir = setup_params['local_log_dir']
# setup hw before test
rutils.setup_hw(ref_hosts, setup_params)
rutils.setup_hw(dut_hosts, setup_params)
# run monitors if requested/possible
for ref_host in ref_hosts:
monitor.add(ref_host, monitors)
monitor.run(ref_host, setup_params)
for dut_host in dut_hosts:
monitor.add(dut_host, monitors)
monitor.run(dut_host, setup_params)
# run hostapd/wpa_supplicant
for ref_host in ref_hosts:
rutils.run_wpasupplicant(ref_host, setup_params)
wpas = WpaSupplicant(hostname = ref_host.host, global_iface="udp", global_port = ref_host.port)
wpas.interface_add(ref_host.ifname)
dev.append(wpas)
for dut_host in dut_hosts:
rutils.run_hostapd(dut_host, setup_params)
dut_host.dev['bssid'] = rutils.get_mac_addr(dut_host)
apdev.append(dut_host.dev)
# run hwsim test/currently only 2 params tests
if hwsim_test.func_code.co_argcount == 1:
hwsim_test(dev)
elif hwsim_test.func_code.co_argcount == 2:
hwsim_test(dev, apdev)
else:
raise Exception("more than 2 arguments required")
# hostapd/wpa_supplicant cleanup
for wpas in dev:
wpas.interface_remove(wpas.host.ifname)
wpas.terminate()
dev = []
# remove monitors
for ref_host in ref_hosts:
monitor.remove(ref_host)
for dut_host in dut_hosts:
monitor.remove(dut_host)
for ref_host in ref_hosts:
ref_host.execute(["killall", "wpa_supplicant"])
ref_host.get_logs(local_log_dir)
for dut_host in dut_hosts:
dut_host.execute(["killall", "hostapd"])
dut_host.get_logs(local_log_dir)
return ""
except:
logger.info(traceback.format_exc())
for wpas in dev:
try:
wpas.interface_remove(wpas.host.ifname)
wpas.terminate()
except:
pass
for ref_host in ref_hosts:
monitor.remove(ref_host)
for dut_host in dut_hosts:
monitor.remove(dut_host)
for ref_host in ref_hosts:
ref_host.execute(["killall", "wpa_supplicant"])
ref_host.get_logs(local_log_dir)
for dut_host in dut_hosts:
dut_host.execute(["killall", "hostapd"])
dut_host.get_logs(local_log_dir)
raise

View file

@ -26,11 +26,13 @@ import config
from test_devices import show_devices
from test_devices import check_devices
from rutils import TestSkip
from utils import HwsimSkip
from hwsim_wrapper import run_hwsim_test
def usage():
print "USAGE: " + sys.argv[0] + " -t devices"
print "USAGE: " + sys.argv[0] + " -t check_devices"
print "USAGE: " + sys.argv[0] + " -d <dut_name> -t <all|sanity|tests_to_run> [-r <ref_name>] [-c <cfg_file.py>] [-m <all|monitor_name>] [-R][-T][-P][-v]"
print "USAGE: " + sys.argv[0] + " -d <dut_name> -t <all|sanity|tests_to_run> [-r <ref_name>] [-c <cfg_file.py>] [-m <all|monitor_name>] [-h hwsim_tests][-R][-T][-P][-v]"
print "USAGE: " + sys.argv[0]
def get_devices(devices, duts, refs, monitors):
@ -67,6 +69,8 @@ def main():
monitors = []
filter_keys = []
requested_tests = ["help"]
requested_hwsim_tests = []
hwsim_tests = []
cfg_file = "cfg.py"
log_dir = "./logs/"
verbose = False
@ -76,9 +80,9 @@ def main():
# parse input parameters
try:
opts, args = getopt.getopt(sys.argv[1:], "d:r:t:l:k:c:m:vRPT",
opts, args = getopt.getopt(sys.argv[1:], "d:r:t:l:k:c:m:h:vRPT",
["dut=", "ref=", "tests=", "log-dir=",
"cfg=", "key=", "monitor="])
"cfg=", "key=", "monitor=", "hwsim="])
except getopt.GetoptError as err:
print(err)
usage()
@ -107,6 +111,8 @@ def main():
monitors.append(argument)
elif option in ("-c", "--cfg"):
cfg_file = argument
elif option in ("-h", "--hwsim"):
requested_hwsim_tests = re.split('; | |, ', argument)
else:
assert False, "unhandled option"
@ -160,12 +166,50 @@ def main():
tests.append(val)
test_names = list(set([t.__name__.replace('test_', '', 1) for t in tests]))
# import test_*
files = os.listdir("../hwsim/")
for t in files:
m = re.match(r'(test_.*)\.py$', t)
if m:
mod = __import__(m.group(1))
test_modules.append(mod.__name__.replace('test_', '', 1))
for key,val in mod.__dict__.iteritems():
if key.startswith("test_"):
hwsim_tests.append(val)
# setup hwsim tests
hwsim_tests_to_run = []
if len(requested_hwsim_tests) > 0:
# apply filters
for filter_key in filter_keys:
filtered_tests = []
for hwsim_test in hwsim_tests:
if re.search(filter_key, hwsim_test.__name__):
filtered_tests.append(hwsim_test)
hwsim_tests = filtered_tests
# setup hwsim_test we should run
if requested_hwsim_tests[0] == "all":
hwsim_tests_to_run = hwsim_tests
else:
for test in requested_hwsim_tests:
t = None
for tt in hwsim_tests:
name = tt.__name__.replace('test_', '', 1)
if name == test and tt.func_code.co_argcount <= 2:
t = tt
break
if not t:
logger.warning("hwsim test case: " + test + " NOT-FOUND")
continue
hwsim_tests_to_run.append(t)
# sort the list
test_names.sort()
tests.sort()
# print help
if requested_tests[0] == "help":
if requested_tests[0] == "help" and len(requested_hwsim_tests) == 0:
usage()
print "\nAvailable Devices:"
for device in devices:
@ -173,6 +217,9 @@ def main():
print "\nAvailable tests:"
for test in test_names:
print "\t", test
print "\nAvailable hwsim tests:"
for hwsim_test in hwsim_tests:
print "\t", hwsim_test.__name__.replace('test_', '', 1)
return
# show/check devices
@ -192,6 +239,8 @@ def main():
tests_to_run = []
if requested_tests[0] == "all":
tests_to_run = tests
if requested_tests[0] == "help":
pass
elif requested_tests[0] == "sanity":
for test in tests:
if test.__name__.startswith("test_sanity_"):
@ -279,6 +328,34 @@ def main():
failed.append(test.__name__.replace('test_', '', 1))
test_no += 1
test_no = 1
for hwsim_test in hwsim_tests_to_run:
try:
start = datetime.now()
setup_params['tc_name'] = hwsim_test.__name__.replace('test_', '', 1)
logger.warning("START - " + setup_params['tc_name'] + " (" + str(test_no) + "/" + str(len(hwsim_tests_to_run)) + ")")
res = run_hwsim_test(devices, setup_params, refs, duts, monitors, hwsim_test)
end = datetime.now()
logger.warning("PASS (" + res + ") - " + str((end - start).total_seconds()) + "s")
except KeyboardInterrupt:
put_devices(devices, duts, refs, monitors)
raise
except HwsimSkip,e:
end = datetime.now()
logger.warning("SKIP (" + str(e) + ") - " + str((end - start).total_seconds()) + "s")
failed.append(hwsim_test.__name__.replace('test_', '', 1))
except Exception, e:
end = datetime.now()
logger.warning("FAILED (" + str(e) + ") - " + str((end - start).total_seconds()) + "s")
logger.info(traceback.format_exc())
failed.append(hwsim_test.__name__.replace('test_', '', 1))
except:
end = datetime.now()
logger.warning("FAILED - " + str((end - start).total_seconds()) + "s")
logger.info(traceback.format_exc())
failed.append(hwsim_test.__name__.replace('test_', '', 1))
test_no += 1
# unlock devices
put_devices(devices, duts, refs, monitors)