[PATCH v2 17/17] tests/remote: add hwsim wrapper
Janusz Dziedzic
janusz.dziedzic at tieto.com
Wed Apr 6 22:38:15 PDT 2016
This allow to run hwsim test cases.
duts goes to apdev while refs goes 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 at tieto.com>
---
tests/remote/config.py | 3 +-
tests/remote/hwsim_wrapper.py | 108 ++++++++++++++++++++++++++++++++++++++++++
tests/remote/run-tests.py | 85 +++++++++++++++++++++++++++++++--
tests/remote/rutils.py | 14 ++++++
4 files changed, 205 insertions(+), 5 deletions(-)
create mode 100644 tests/remote/hwsim_wrapper.py
diff --git a/tests/remote/config.py b/tests/remote/config.py
index 44efd2f..04ac4c5 100644
--- a/tests/remote/config.py
+++ b/tests/remote/config.py
@@ -43,7 +43,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"):
diff --git a/tests/remote/hwsim_wrapper.py b/tests/remote/hwsim_wrapper.py
new file mode 100644
index 0000000..5a39554
--- /dev/null
+++ b/tests/remote/hwsim_wrapper.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python
+#
+# 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
diff --git a/tests/remote/run-tests.py b/tests/remote/run-tests.py
index c3cb16a..c033c55 100755
--- a/tests/remote/run-tests.py
+++ b/tests/remote/run-tests.py
@@ -25,11 +25,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):
@@ -66,6 +68,8 @@ def main():
monitors = []
filter_keys = []
requested_tests = ["help"]
+ requested_hwsim_tests = []
+ hwsim_tests = []
cfg_file = "cfg.py"
log_dir = "./logs/"
verbose = False
@@ -75,8 +79,8 @@ def main():
# parse input parameters
try:
- opts, args = getopt.getopt(sys.argv[1:], "d:r:t:l:k:c:m:vRPT",
- ["dut=", "ref=", "tests=", "log-dir=", "cfg=", "key=", "monitor="])
+ opts, args = getopt.getopt(sys.argv[1:], "d:r:t:l:k:c:m:h:vRPT",
+ ["dut=", "ref=", "tests=", "log-dir=", "cfg=", "key=", "monitor=", "hwsim="])
except getopt.GetoptError as err:
print(err)
usage()
@@ -105,6 +109,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"
@@ -158,12 +164,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:
@@ -171,6 +215,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__
return
# show/check devices
@@ -190,6 +237,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_"):
@@ -277,6 +326,34 @@ def main():
failed.append(test.__name__.replace('test_', '', 1))
test_no = 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 = test_no + 1
+
# unlock devices
put_devices(devices, duts, refs, monitors)
diff --git a/tests/remote/rutils.py b/tests/remote/rutils.py
index e321fee..2cbabd6 100644
--- a/tests/remote/rutils.py
+++ b/tests/remote/rutils.py
@@ -256,6 +256,20 @@ def get_ipv4_addr(setup_params, number):
return ipv4
+def get_mac_addr(host, iface=None):
+ if iface == None:
+ iface = host.ifname
+ status, buf = host.execute("ifconfig " + iface)
+ if status != 0:
+ raise Exception("ifconfig " + iface)
+ words = buf.split()
+ found = 0
+ for word in words:
+ if found == 1:
+ return word
+ if word == "HWaddr":
+ found = 1
+ raise Exception("Could not find HWaddr")
# connectivity/ping helpers
def get_ping_packet_loss(ping_res):
--
1.9.1
More information about the Hostap
mailing list