X-Git-Url: http://www.project-moonshot.org/gitweb/?p=mech_eap.git;a=blobdiff_plain;f=libeap%2Ftests%2Fhwsim%2Fwlantest.py;fp=libeap%2Ftests%2Fhwsim%2Fwlantest.py;h=10bb45fea282c6576142cfea7471f1609e7f171f;hp=5f6b4ac2e1108fbaa50aa4b7e20d80bd3eab823e;hb=d1dd9aae6741e74f20bfc35e1db598652680279d;hpb=bd3bd69af16ab99706ba70ed11a3e291e968e5c6 diff --git a/libeap/tests/hwsim/wlantest.py b/libeap/tests/hwsim/wlantest.py index 5f6b4ac..10bb45f 100644 --- a/libeap/tests/hwsim/wlantest.py +++ b/libeap/tests/hwsim/wlantest.py @@ -4,7 +4,9 @@ # This software may be distributed under the terms of the BSD license. # See README for more details. +import re import os +import posixpath import time import subprocess import logging @@ -13,44 +15,133 @@ import wpaspy logger = logging.getLogger() class Wlantest: + remote_host = None + setup_params = None + exe_thread = None + exe_res = [] + monitor_mod = None + setup_done = False + + @classmethod + def stop_remote_wlantest(cls): + if cls.exe_thread is None: + # Local flow - no need for remote operations + return + + cls.remote_host.execute(["killall", "-9", "wlantest"]) + cls.remote_host.wait_execute_complete(cls.exe_thread, 5) + cls.exe_thread = None + cls.exe_res = [] + + @classmethod + def reset_remote_wlantest(cls): + cls.stop_remote_wlantest() + cls.remote_host = None + cls.setup_params = None + cls.exe_thread = None + cls.exe_res = [] + cls.monitor_mod = None + cls.setup_done = False + + @classmethod + def start_remote_wlantest(cls): + if cls.remote_host is None: + # Local flow - no need for remote operations + return + if cls.exe_thread is not None: + raise Exception("Cannot start wlantest twice") + + log_dir = cls.setup_params['log_dir'] + ifaces = re.split('; | |, ', cls.remote_host.ifname) + ifname = ifaces[0] + exe = cls.setup_params["wlantest"] + tc_name = cls.setup_params["tc_name"] + base_log_name = tc_name + "_wlantest_" + \ + cls.remote_host.name + "_" + ifname + log_file = posixpath.join(log_dir, base_log_name + ".log") + pcap_file = posixpath.join(log_dir, base_log_name + ".pcapng") + cmd = "{} -i {} -n {} -c -dtN -L {}".format(exe, ifname, + pcap_file, log_file) + cls.remote_host.add_log(log_file) + cls.remote_host.add_log(pcap_file) + cls.exe_thread = cls.remote_host.execute_run(cmd.split(), cls.exe_res) + # Give wlantest a chance to start working + time.sleep(1) + + @classmethod + def register_remote_wlantest(cls, host, setup_params, monitor_mod): + if cls.remote_host is not None: + raise Exception("Cannot register remote wlantest twice") + cls.remote_host = host + cls.setup_params = setup_params + cls.monitor_mod = monitor_mod + status, buf = host.execute(["which", setup_params['wlantest']]) + if status != 0: + raise Exception(host.name + " - wlantest: " + buf) + status, buf = host.execute(["which", setup_params['wlantest_cli']]) + if status != 0: + raise Exception(host.name + " - wlantest_cli: " + buf) + + @classmethod + def chan_from_wpa(cls, wpa, is_p2p=False): + if cls.monitor_mod is None: + return + m = cls.monitor_mod + return m.setup(cls.remote_host, [m.get_monitor_params(wpa, is_p2p)]) + + @classmethod + def setup(cls, wpa, is_p2p=False): + cls.chan_from_wpa(wpa, is_p2p) + cls.start_remote_wlantest() + cls.setup_done = True + def __init__(self): + if not self.setup_done: + raise Exception("Cannot create Wlantest instance before setup()") if os.path.isfile('../../wlantest/wlantest_cli'): self.wlantest_cli = '../../wlantest/wlantest_cli' else: self.wlantest_cli = 'wlantest_cli' + def cli_cmd(self, params): + if self.remote_host is not None: + exe = self.setup_params["wlantest_cli"] + ret = self.remote_host.execute([exe] + params) + if ret[0] != 0: + raise Exception("wlantest_cli failed") + return ret[1] + else: + return subprocess.check_output([self.wlantest_cli] + params) + def flush(self): - res = subprocess.check_output([self.wlantest_cli, "flush"]) + res = self.cli_cmd(["flush"]) if "FAIL" in res: raise Exception("wlantest_cli flush failed") def relog(self): - res = subprocess.check_output([self.wlantest_cli, "relog"]) + res = self.cli_cmd(["relog"]) if "FAIL" in res: raise Exception("wlantest_cli relog failed") def add_passphrase(self, passphrase): - res = subprocess.check_output([self.wlantest_cli, "add_passphrase", - passphrase]) + res = self.cli_cmd(["add_passphrase", passphrase]) if "FAIL" in res: raise Exception("wlantest_cli add_passphrase failed") def add_wepkey(self, key): - res = subprocess.check_output([self.wlantest_cli, "add_wepkey", key]) + res = self.cli_cmd(["add_wepkey", key]) if "FAIL" in res: raise Exception("wlantest_cli add_key failed") def info_bss(self, field, bssid): - res = subprocess.check_output([self.wlantest_cli, "info_bss", - field, bssid]) + res = self.cli_cmd(["info_bss", field, bssid]) if "FAIL" in res: raise Exception("Could not get BSS info from wlantest for " + bssid) return res def get_bss_counter(self, field, bssid): try: - res = subprocess.check_output([self.wlantest_cli, "get_bss_counter", - field, bssid]); + res = self.cli_cmd(["get_bss_counter", field, bssid]) except Exception, e: return 0 if "FAIL" in res: @@ -58,36 +149,30 @@ class Wlantest: return int(res) def clear_bss_counters(self, bssid): - subprocess.call([self.wlantest_cli, "clear_bss_counters", bssid], - stdout=open('/dev/null', 'w')); + self.cli_cmd(["clear_bss_counters", bssid]) def info_sta(self, field, bssid, addr): - res = subprocess.check_output([self.wlantest_cli, "info_sta", - field, bssid, addr]) + res = self.cli_cmd(["info_sta", field, bssid, addr]) if "FAIL" in res: raise Exception("Could not get STA info from wlantest for " + addr) return res def get_sta_counter(self, field, bssid, addr): - res = subprocess.check_output([self.wlantest_cli, "get_sta_counter", - field, bssid, addr]); + res = self.cli_cmd(["get_sta_counter", field, bssid, addr]) if "FAIL" in res: raise Exception("wlantest_cli command failed") return int(res) def clear_sta_counters(self, bssid, addr): - res = subprocess.check_output([self.wlantest_cli, "clear_sta_counters", - bssid, addr]); + res = self.cli_cmd(["clear_sta_counters", bssid, addr]) if "FAIL" in res: raise Exception("wlantest_cli command failed") def tdls_clear(self, bssid, addr1, addr2): - res = subprocess.check_output([self.wlantest_cli, "clear_tdls_counters", - bssid, addr1, addr2]); + self.cli_cmd(["clear_tdls_counters", bssid, addr1, addr2]) def get_tdls_counter(self, field, bssid, addr1, addr2): - res = subprocess.check_output([self.wlantest_cli, "get_tdls_counter", - field, bssid, addr1, addr2]); + res = self.cli_cmd(["get_tdls_counter", field, bssid, addr1, addr2]) if "FAIL" in res: raise Exception("wlantest_cli command failed") return int(res) @@ -139,15 +224,13 @@ class Wlantest: raise Exception("Unexpected STA key_mgmt") def get_tx_tid(self, bssid, addr, tid): - res = subprocess.check_output([self.wlantest_cli, "get_tx_tid", - bssid, addr, str(tid)]); + res = self.cli_cmd(["get_tx_tid", bssid, addr, str(tid)]) if "FAIL" in res: raise Exception("wlantest_cli command failed") return int(res) def get_rx_tid(self, bssid, addr, tid): - res = subprocess.check_output([self.wlantest_cli, "get_rx_tid", - bssid, addr, str(tid)]); + res = self.cli_cmd(["get_rx_tid", bssid, addr, str(tid)]) if "FAIL" in res: raise Exception("wlantest_cli command failed") return int(res)