autotest: Refactor iw related code into a delegate
This allows us to easily reuse our interfaces to iw, such as
the parsing of iw list. The iw list parsing has been improved
to additionally parse supported 802.11n MCS indices.
TEST=wifi_matfunc continues to pass with these changes
BUG=chromium:308326
Change-Id: I92da1174d981bf0ba5f97227b68157ce9c2a6ecb
Reviewed-on: https://chromium-review.googlesource.com/173642
Reviewed-by: Christopher Wiley <wiley@chromium.org>
Tested-by: Christopher Wiley <wiley@chromium.org>
Commit-Queue: David James <davidjames@chromium.org>
diff --git a/server/cros/chaos_lib/chaos_runner.py b/server/cros/chaos_lib/chaos_runner.py
index 0c88a96..1779629 100644
--- a/server/cros/chaos_lib/chaos_runner.py
+++ b/server/cros/chaos_lib/chaos_runner.py
@@ -141,7 +141,7 @@
tag=ap.ssid)
continue
iw_scanner = iw_runner.IwRunner(self._host,
- iw_command=self._wifi_client.command_iw)
+ command_iw=self._wifi_client.command_iw)
networks = iw_scanner.wait_for_scan_result(
self._wifi_client._wifi_if, ssid=ap.ssid,
# We have some APs that need a while to come on-line
diff --git a/server/cros/network/iw_runner.py b/server/cros/network/iw_runner.py
index bba84bb..073fa37 100644
--- a/server/cros/network/iw_runner.py
+++ b/server/cros/network/iw_runner.py
@@ -3,6 +3,7 @@
# found in the LICENSE file.
import collections
+import re
import time
HT20 = 'HT20'
@@ -16,17 +17,138 @@
'above': HT40_ABOVE,
'below': HT40_BELOW}
+IwBand = collections.namedtuple('Band', ['num', 'frequencies', 'mcs_indices'])
IwBss = collections.namedtuple('IwBss', ['bss', 'frequency', 'ssid', 'ht'])
+IwPhy = collections.namedtuple('Phy', ['name', 'bands'])
DEFAULT_COMMAND_IW = 'iw'
class IwRunner(object):
- """Class that parses iw <device>."""
+ """Defines an interface to the 'iw' command."""
- def __init__(self, host, iw_command=DEFAULT_COMMAND_IW):
+ def __init__(self, host, command_iw=DEFAULT_COMMAND_IW):
self._host = host
- self._iw_command = iw_command
+ self._command_iw = command_iw
+
+
+ def add_interface(self, phy, interface, interface_type):
+ """
+ Add an interface to a WiFi PHY.
+
+ @param phy: string name of PHY to add an interface to.
+ @param interface: string name of interface to add.
+ @param interface_type: string type of interface to add (e.g. 'monitor').
+
+ """
+ self._host.run('%s phy %s interface add %s type %s' %
+ (self._command_iw, phy, interface, interface_type))
+
+
+ def disconnect_station(self, interface):
+ """
+ Disconnect a STA from a network.
+
+ @param interface: string name of interface to disconnect.
+
+ """
+ self._host.run('%s dev %s disconnect' % (self._command_iw, interface))
+
+
+ def ibss_join(self, interface, ssid, frequency):
+ """
+ Join a WiFi interface to an IBSS.
+
+ @param interface: string name of interface to join to the IBSS.
+ @param ssid: string SSID of IBSS to join.
+ @param frequency: int frequency of IBSS in Mhz.
+
+ """
+ self._host.run('%s dev %s ibss join %s %d' %
+ (self._command_iw, interface, ssid, frequency))
+
+
+ def ibss_leave(self, interface):
+ """
+ Leave an IBSS.
+
+ @param interface: string name of interface to remove from the IBSS.
+
+ """
+ self._host.run('%s dev %s ibss leave' % (self._command_iw, interface))
+
+
+ def list_interfaces(self):
+ """@return list of string WiFi interface names on device."""
+ output = self._host.run('%s dev' % self._command_iw).stdout
+ interfaces = []
+ for line in output.splitlines():
+ m = re.match('[\s]*Interface (.*)', line)
+ if m:
+ interfaces.append(m.group(1))
+
+ return interfaces
+
+
+ def list_phys(self):
+ """
+ List WiFi PHYs on the given host.
+
+ @return list of IwPhy tuples.
+
+ """
+ output = self._host.run('%s list' % self._command_iw).stdout
+ current_phy = None
+ current_band = None
+ all_phys = []
+ for line in output.splitlines():
+ match_phy = re.search('Wiphy (.*)', line)
+ if match_phy:
+ current_phy = IwPhy(name=match_phy.group(1), bands=[])
+ all_phys.append(current_phy)
+ continue
+ match_band = re.search('Band (\d+):', line)
+ if match_band:
+ current_band = IwBand(num=int(match_band.group(1)),
+ frequencies=[],
+ mcs_indices=[])
+ current_phy.bands.append(current_band)
+ continue
+ if not all([current_band, current_phy, line.startswith('\t')]):
+ continue
+
+ mhz_match = re.search('(\d+) MHz', line)
+ if mhz_match:
+ current_band.frequencies.append(int(mhz_match.group(1)))
+ continue
+
+ # re_mcs needs to match something like:
+ # HT TX/RX MCS rate indexes supported: 0-15, 32
+ if re.search('HT TX/RX MCS rate indexes supported: ', line):
+ rate_string = line.split(':')[1].strip()
+ for piece in rate_string.split(','):
+ if piece.find('-') > 0:
+ # Must be a range like ' 0-15'
+ begin, end = piece.split('-')
+ for index in range(int(begin), int(end) + 1):
+ current_band.mcs_indices.append(index)
+ else:
+ # Must be a single rate like '32 '
+ current_band.mcs_indices.append(int(piece))
+ return all_phys
+
+
+ def remove_interface(self, interface, ignore_status=False):
+ """
+ Remove a WiFi interface from a PHY.
+
+ @param interface: string name of interface (e.g. mon0)
+ @param ignore_status: boolean True iff we should ignore failures
+ to remove the interface.
+
+ """
+ self._host.run('%s dev %s del' % (self._command_iw, interface),
+ ignore_status=ignore_status)
def scan(self, interface):
@@ -37,7 +159,7 @@
@returns a list of IwBss collections; None if the scan fails
"""
- command = str('%s %s scan' % (self._iw_command, interface))
+ command = str('%s %s scan' % (self._command_iw, interface))
scan = self._host.run(command, ignore_status=True)
if scan.exit_status == 240:
# The device was busy
@@ -73,6 +195,28 @@
return bss_list
+ def set_tx_power(self, interface, power):
+ """
+ Set the transmission power for an interface.
+
+ @param interface: string name of interface to set Tx power on.
+ @param power: string power parameter. (e.g. 'auto').
+
+ """
+ self._host.run('%s dev %s set txpower %s' %
+ (self._command_iw, interface, power))
+
+
+ def set_regulatory_domain(self, domain_string):
+ """
+ Set the regulatory domain of the current machine.
+
+ @param domain_string: string regulatory domain name (e.g. 'US').
+
+ """
+ self._host.run('%s reg set %s' % (self._command_iw, domain_string))
+
+
def wait_for_scan_result(self, interface, bss=None, ssid=None,
timeout_seconds=30):
"""Returns a IWBSS object for a network with the given bssed or ssid.
diff --git a/server/site_linux_router.py b/server/site_linux_router.py
index aa45297..25fd942 100644
--- a/server/site_linux_router.py
+++ b/server/site_linux_router.py
@@ -114,7 +114,7 @@
self.stop_dhcp_servers()
# Place us in the US by default
- self.router.run("%s reg set US" % self.cmd_iw)
+ self.iw_runner.set_regulatory_domain('US')
def close(self):
@@ -537,9 +537,8 @@
self._build_ssid(config.ssid_suffix))
# Connect the station
self.router.run('%s link set %s up' % (self.cmd_ip, interface))
- self.router.run('%s dev %s ibss join %s %d' % (
- self.cmd_iw, interface, self.station['conf']['ssid'],
- config.frequency))
+ self.iw_runner.ibss_join(
+ interface, self.station['conf']['ssid'], config.frequency)
# Always start a local server.
self.start_local_server(interface)
# Remember that this interface is up.
@@ -730,11 +729,9 @@
local_servers = self.local_servers
self.local_servers = []
if self.station['type'] == 'ibss':
- self.router.run("%s dev %s ibss leave" %
- (self.cmd_iw, self.station['interface']))
+ self.iw_runner.ibss_leave(self.station['interface'])
else:
- self.router.run("%s dev %s disconnect" %
- (self.cmd_iw, self.station['interface']))
+ self.iw_runner.disconnect_station(self.station['interface'])
self.router.run("%s link set %s down" % (self.cmd_ip,
self.station['interface']))
@@ -784,8 +781,7 @@
interface = params.get('interface',
self.hostapd_instances[0]['interface'])
power = params.get('power', 'auto')
- self.router.run("%s dev %s set txpower %s" %
- (self.cmd_iw, interface, power))
+ self.iw_runner.set_tx_power(interface, power)
def deauth(self, params):
diff --git a/server/site_linux_system.py b/server/site_linux_system.py
index ae71c61..ff4f762 100644
--- a/server/site_linux_system.py
+++ b/server/site_linux_system.py
@@ -4,12 +4,12 @@
import datetime
import logging
-import re
import time
from autotest_lib.client.common_lib import error
from autotest_lib.server.cros import wifi_test_utils
from autotest_lib.server.cros.network import packet_capturer
+from autotest_lib.server.cros.network import iw_runner
class LinuxSystem(object):
"""Superclass for test machines running Linux.
@@ -41,7 +41,7 @@
def __init__(self, host, params, role):
# Command locations.
- self.cmd_iw = wifi_test_utils.must_be_installed(
+ cmd_iw = wifi_test_utils.must_be_installed(
host, params.get('cmd_iw', '/usr/sbin/iw'))
self.cmd_ip = wifi_test_utils.must_be_installed(
host, params.get('cmd_ip', '/usr/sbin/ip'))
@@ -63,11 +63,13 @@
host, params.get('cmd_ifconfig', '/sbin/ifconfig'))
self._packet_capturer = packet_capturer.get_packet_capturer(
self.host, host_description=role, cmd_ifconfig=cmd_ifconfig,
- cmd_ip=self.cmd_ip, cmd_iw=self.cmd_iw, cmd_netdump=cmd_netdump,
+ cmd_ip=self.cmd_ip, cmd_iw=cmd_iw, cmd_netdump=cmd_netdump,
ignore_failures=True)
+ self.iw_runner = iw_runner.IwRunner(host, command_iw=cmd_iw)
self.phys_for_frequency, self.phy_bus_type = self._get_phy_info()
self.wlanifs_in_use = []
+ self.wlanifs = {}
self._capabilities = None
@@ -85,29 +87,17 @@
@return phys_for_frequency, phy_bus_type tuple as described.
"""
- output = self.host.run('%s list' % self.cmd_iw).stdout
- re_wiphy = re.compile('Wiphy (.*)')
- re_mhz = re.compile('(\d+) MHz')
- in_phy = False
- phy_list = []
+ phys = self.iw_runner.list_phys()
phys_for_frequency = {}
- for line in output.splitlines():
- match_wiphy = re_wiphy.match(line)
- if match_wiphy:
- in_phy = True
- wiphyname = match_wiphy.group(1)
- phy_list.append(wiphyname)
- elif in_phy:
- if line[0] == '\t':
- match_mhz = re_mhz.search(line)
- if match_mhz:
- mhz = int(match_mhz.group(1))
- if mhz not in phys_for_frequency:
- phys_for_frequency[mhz] = [wiphyname]
- else:
- phys_for_frequency[mhz].append(wiphyname)
- else:
- in_phy = False
+ phy_list = []
+ for phy in phys:
+ phy_list.append(phy.name)
+ for band in phy.bands:
+ for mhz in band.frequencies:
+ if mhz not in phys_for_frequency:
+ phys_for_frequency[mhz] = [phy.name]
+ else:
+ phys_for_frequency[mhz].append(phy.name)
phy_bus_type = {}
for phy in phy_list:
@@ -133,14 +123,14 @@
"""
self.host.run('%s link set %s down' % (self.cmd_ip, interface))
- self.host.run('%s dev %s del' % (self.cmd_iw, interface))
+ self.iw_runner.remove_interface(interface)
if remove_monitor:
# Some old hostap implementations create a 'mon.<interface>' to
# handle management frame transmit/receive.
self.host.run('%s link set mon.%s down' % (self.cmd_ip, interface),
ignore_status=True)
- self.host.run('%s dev mon.%s del' % (self.cmd_iw, interface),
- ignore_status=True)
+ self.iw_runner.remove_interface('mon.%s' % interface,
+ ignore_status=True)
for phytype in self.wlanifs:
for phy in self.wlanifs[phytype]:
if self.wlanifs[phytype][phy] == interface:
@@ -150,14 +140,9 @@
def _remove_interfaces(self):
"""Remove all WiFi devices."""
+ for interface in self.iw_runner.list_interfaces():
+ self.iw_runner.remove_interface(interface)
self.wlanifs = {}
- # Remove all wifi devices.
- output = self.host.run('%s dev' % self.cmd_iw).stdout
- test = re.compile('[\s]*Interface (.*)')
- for line in output.splitlines():
- m = test.match(line)
- if m:
- self._remove_interface(m.group(1), False)
def close(self):
@@ -329,9 +314,7 @@
wlanif = '%s%d' % (phytype, len(self.wlanifs[phytype].keys()))
self.wlanifs[phytype][phy] = wlanif
- self.host.run('%s phy %s interface add %s type %s' %
- (self.cmd_iw, phy, wlanif, phytype))
-
+ self.iw_runner.add_interface(phy, wlanif, phytype)
self.wlanifs_in_use.append((phy, wlanif, phytype))
return wlanif