# 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
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:
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)
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)
projects / mech_eap.git / blobdiff