blob: 6e2762e13e921cb57a40e8bfe56b342242e65191 [file] [log] [blame]
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""A factory test for checking Wifi antenna.
The test accepts a dict of wireless specs.
Each spec contains candidate services and the signal constraints.
For each spec, the test will connect to AP first.
And scan the signal quality to get signal strength for all antennas.
Then the test checks signal quality.
Be sure to set AP correctly.
1. Select one fixed channel instead of auto.
2. Disable the TX power control in AP.
3. Make sure SSID of AP is unique.
This test case can be used for Intel WP2 7260 chip.
import collections
import dbus
import logging
import re
import struct
import sys
import threading
import time
import unittest
import factory_common # pylint: disable=unused-import
from cros.factory.test import event_log
from cros.factory.test import factory
from cros.factory.test.i18n import test_ui as i18n_test_ui
from cros.factory.test import test_ui
from cros.factory.test import ui_templates
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import net_utils
from cros.factory.utils import process_utils
import flimflam
except ImportError:
_DEFAULT_WIRELESS_TEST_CSS = '.wireless-info {font-size: 2em;}'
_MSG_SWITCHING_AP = lambda ap: i18n_test_ui.MakeI18nLabelWithClass(
'Switching to AP {ap}: ', 'wireless-info', ap=ap)
_MSG_SCANNING = lambda device, freq: i18n_test_ui.MakeI18nLabelWithClass(
'Scanning on device {device} frequency {freq}...',
'wireless-info', device=device, freq=freq)
_MSG_SCANNING_DONE = lambda device, freq: i18n_test_ui.MakeI18nLabelWithClass(
'Done scanning on device {device} frequency {freq}...',
'wireless-info', device=device, freq=freq)
_MSG_SPACE = i18n_test_ui.MakeI18nLabelWithClass(
'Press space to start scanning.', 'wireless-info')
_MSG_PRECHECK = i18n_test_ui.MakeI18nLabelWithClass(
'Checking frequencies...', 'wireless-info')
_RE_IWSCAN = re.compile(r'freq: (\d+).*SSID: (.+)$')
_RE_WIPHY = re.compile(r'wiphy (\d+)')
_RE_BEACON = re.compile(r'(\d+) MHz.*Beacon \((.+)\)')
_ANTENNA_CONFIG = ['all', 'main', 'aux']
def FlimGetService(flim, name):
"""Get service by property.
flim: flimflam object
name: property name
timeout = time.time() + 10
while time.time() < timeout:
service = flim.FindElementByPropertySubstring('Service', 'Name', name)
if service:
return service
def FlimGetServiceProperty(service, prop):
"""Get property from a service.
service: flimflam service object
prop: property name
timeout = time.time() + 10
while time.time() < timeout:
properties = service.GetProperties()
except dbus.exceptions.DBusException as e:
logging.exception('Error reading service property')
return properties[prop]
raise e
def FlimConfigureService(flim, ssid, password):
"""Config wireless ssid and password.
ssid: ssid name
password: wifi key to authenticate
wlan_dict = {
'Type': dbus.String('wifi', variant_level=1),
'Mode': dbus.String('managed', variant_level=1),
'AutoConnect': dbus.Boolean(False, variant_level=1),
'SSID': dbus.String(ssid, variant_level=1)}
if password:
wlan_dict['Security'] = dbus.String('psk', variant_level=1)
wlan_dict['Passphrase'] = dbus.String(password, variant_level=1)
def IwScan(devname, sleep_retry_time_secs=2, max_retries=10):
"""Scans on device.
devname: device name.
sleep_retry_time_secs: The sleep time before a retry.
max_retries: The maximum retry time to scan.
A list of (ssid, frequency) tuple.
IwException if fail to scan for max_retries tries,
or fail because of reason other than device or resource busy (-16)
cmd = r"iw %s scan | grep -e 'freq\|SSID' | sed 'N;s/\n/ /'" % devname
try_count = 0
scan_result = []
while try_count < max_retries:
process = process_utils.Spawn(
cmd, read_stdout=True, log_stderr_on_error=True, log=True, shell=True)
stdout, stderr = process.communicate()
retcode = process.returncode
event_log.Log('iw_scaned', retcode=retcode, stderr=stderr)
if retcode == 0:
for line in stdout.splitlines():
m =
if m:
if len(scan_result) == 0:
try_count += 1
continue'IwScan success.')
return scan_result
elif retcode == 240: # Device or resource busy (-16)
try_count += 1
elif retcode == 234: # Invalid argument (-22)
raise Exception('Failed to iw scan, ret code: %d. stderr: %s'
'Frequency might be wrong.' %
(retcode, stderr))
raise Exception('Failed to iw scan, ret code: %d. stderr: %s' %
(retcode, stderr))
raise Exception('Failed to iw scan for %s tries' % max_retries)
class RadiotapPacket(object):
FIELD = collections.namedtuple('Field', ['name', 'struct', 'align'])
ANTENNA_SIGNAL_FIELD = FIELD('Antenna Signal', struct.Struct('b'), 0)
ANTENNA_INDEX_FIELD = FIELD('Antenna Index', struct.Struct('B'), 0)
FIELD('TSFT', struct.Struct('Q'), 8),
FIELD('Flags', struct.Struct('B'), 0),
FIELD('Rate', struct.Struct('B'), 0),
FIELD('Channel', struct.Struct('HH'), 2),
FIELD('FHSS', struct.Struct('BB'), 0),
FIELD('Antenna Noise', struct.Struct('b'), 0),
FIELD('Lock Quality', struct.Struct('H'), 2),
FIELD('TX Attenuation', struct.Struct('H'), 2),
FIELD('dB TX Attenuation', struct.Struct('H'), 2),
FIELD('dBm TX Power', struct.Struct('b'), 1),
FIELD('dB Antenna Signal', struct.Struct('B'), 0),
FIELD('dB Antenna Noise', struct.Struct('B'), 0),
FIELD('RX Flags', struct.Struct('H'), 2),
FIELD('MCS', struct.Struct('BBB'), 1),
FIELD('AMPDU status', struct.Struct('IHBB'), 4),
FIELD('VHT', struct.Struct('HBBBBBBBBH'), 2)]
MAIN_HEADER_FORMAT = struct.Struct('BBhI')
PARSE_INFO = collections.namedtuple('AntennaData', ['header_size',
# This is a variable-length header, but this is what we want to see.
def decode(packet_bytes):
"""Returns signal strength data for each antenna.
Format is {all_signal, {antenna_index, antenna_signal}}.
if len(packet_bytes) < RadiotapPacket.EXPECTED_HEADER_FORMAT.size:
return None
parts = RadiotapPacket.EXPECTED_HEADER_FORMAT.unpack_from(packet_bytes)
(_, _, _, present0, present1, present2) = parts
parse_info = RadiotapPacket.parse_header([present0, present1, present2])
required_bytes = parse_info.header_size + parse_info.data_bytes
if len(packet_bytes) < required_bytes:
return None
antenna_data = []
for datum in filter(bool, parse_info.antenna_offsets):
signal = datum.get(RadiotapPacket.ANTENNA_SIGNAL_FIELD)
if RadiotapPacket.ANTENNA_SIGNAL_FIELD not in datum:
signal_offset = datum[RadiotapPacket.ANTENNA_SIGNAL_FIELD]
signal, = RadiotapPacket.ANTENNA_SIGNAL_FIELD.struct.unpack_from(
packet_bytes[(signal_offset + parse_info.header_size):])
if RadiotapPacket.ANTENNA_INDEX_FIELD in datum:
index_offset = datum[RadiotapPacket.ANTENNA_INDEX_FIELD]
index, = RadiotapPacket.ANTENNA_SIGNAL_FIELD.struct.unpack_from(
packet_bytes[(index_offset + parse_info.header_size):])
antenna_data.append((index, signal))
return antenna_data
def parse_header(field_list):
"""Returns packet information of the radiotap header should have."""
header_size = RadiotapPacket.MAIN_HEADER_FORMAT.size
data_bytes = 0
antenna_offsets = []
for _, bitmask in enumerate(field_list):
for bit, field in enumerate(RadiotapPacket.FIELDS):
if bitmask & (1 << bit):
if field.align and (data_bytes % field.align):
data_bytes += field.align - (data_bytes % field.align)
if (field == RadiotapPacket.ANTENNA_SIGNAL_FIELD or
field == RadiotapPacket.ANTENNA_INDEX_FIELD):
antenna_offsets[-1][field] = data_bytes
data_bytes += field.struct.size
if not bitmask & (1 << RadiotapPacket.EXTENDED_BIT):
header_size += 4
raise NotImplementedError('Packet has too many extensions for me!')
# Offset the antenna fields by the header size.
return RadiotapPacket.PARSE_INFO(header_size, data_bytes, antenna_offsets)
class Capture(object):
"""Context for a live tcpdump packet capture for beacons."""
def __init__(self, device_name, phy):
self.monitor_process = None
self.created_device = None
self.parent_device = device_name
self.phy = phy
def create_device(self, monitor_device='antmon0'):
"""Creates a monitor device to monitor beacon."""
process_utils.Spawn(['iw', self.parent_device, 'interface', 'add',
monitor_device, 'type', 'monitor'], check_call=True)
self.created_device = monitor_device
def remove_device(self, device_name):
"""Removes monitor device."""
process_utils.Spawn(['iw', device_name, 'del'], check_call=True)
def get_signal(self):
"""Gets signal from tcpdump."""
while True:
line = self.monitor_process.stdout.readline()
m =
if m:
freq = int(
ssid =
packet_bytes = ''
while True:
line = self.monitor_process.stdout.readline()
if not line.startswith('\t0x'):
# Break up lines of the form "\t0x0000: abcd ef" into a string
# "\xab\xcd\exef".
parts = line[3:].split()
for part in parts[1:]:
packet_bytes += chr(int(part[:2], 16))
if len(part) > 2:
packet_bytes += chr(int(part[2:], 16))
packet = RadiotapPacket.decode(packet_bytes)
if packet:
return {'ssid': ssid, 'freq': freq, 'signal': packet}
def set_beacon_filter(self, value):
"""Sets beacon filter.
This function may only for Intel WP2 7260 chip.
with open('/sys/kernel/debug/ieee80211/%s/netdev:%s/iwlmvm/bf_params' %
(self.phy, self.parent_device), 'w') as f:
f.write('bf_enable_beacon_filter=%d\n' % value)
def __enter__(self):
if not self.created_device:
['ip', 'link', 'set', self.created_device, 'up'], check_call=True)
['iw', self.parent_device, 'set', 'power_save', 'off'], check_call=True)
self.monitor_process = process_utils.Spawn(
['tcpdump', '-nUxxi', self.created_device, 'type', 'mgt',
'subtype', 'beacon'], stdout=process_utils.PIPE)
return self
def __exit__(self, exception, value, traceback):
if self.created_device:
class WirelessRadiotapTest(unittest.TestCase):
"""Basic wireless test class.
_ui: Test ui.
_template: Test template.
_antenna: current antenna config.
_phy_name: wireless phy name to test.
_space_event: An event that space has been pressed. It will also be set
if test has been done.
_done: An event that test has been done.
ARGS = [
Arg('device_name', str,
'Wireless device name to test. '
'Set this correctly if check_antenna is True.', default='wlan0'),
Arg('services', list,
'A list of (service_ssid, freq, password) tuples like '
'``[(SSID1, FREQ1, PASS1), (SSID2, FREQ2, PASS2), '
'(SSID3, FREQ3, PASS3)]``. The test will only check the service '
'whose antenna_all signal strength is the largest. For example, if '
'(SSID1, FREQ1, PASS1) has the largest signal among the APs, '
'then only its results will be checked against the spec values.',
Arg('strength', dict,
'A dict of minimal signal strengths. For example, a dict like '
'``{"main": strength_1, "aux": strength_2, "all": strength_all}``. '
'The test will check signal strength according to the different '
'antenna configurations in this dict.',
Arg('scan_count', int,
'Number of scans to get average signal strength.', default=5),
Arg('switch_antenna_sleep_secs', int,
'The sleep time after switchingantenna and ifconfig up. Need to '
'decide this value carefully since itdepends on the platform and '
'antenna config to test.', default=10),
Arg('press_space_to_start', bool,
'Press space to start the test.', default=True)]
def setUp(self):
self._ui = test_ui.UI()
self._template = ui_templates.OneSection(self._ui)
self._phy_name = self.DetectPhyName()'phy name is %s.', self._phy_name)
self._space_event = threading.Event()
self._done = threading.Event()
net_utils.Ifconfig(self.args.device_name, True)
self._flim = flimflam.FlimFlam(dbus.SystemBus())
self._connect_service = None
def tearDown(self):
def ConnectService(self, service_name, password):
"""Associates a specified wifi AP.
Password can be '' or None.
self._connect_service = FlimGetService(self._flim, service_name)
if self._connect_service is None:'Unable to find service %s', service_name)
return False
if FlimGetServiceProperty(self._connect_service, 'IsActive'):
logging.warning('Already connected to %s', service_name)
else:'Connecting to %s', service_name)
FlimConfigureService(self._flim, service_name, password)
success, diagnostics = self._flim.ConnectService(
if not success:'Unable to connect to %s, diagnostics %s',
service_name, diagnostics)
return False
'Successfully connected to service %s', service_name)
return True
def DisconnectService(self):
"""Disconnect wifi AP."""
if self._connect_service:
'Disconnect to service %s',
FlimGetServiceProperty(self._connect_service, 'Name'))
self._connect_service = None
def DetectPhyName(self):
"""Detects the phy name for device_name device.
The phy name for device_name device.
output = process_utils.CheckOutput(
['iw', 'dev', self.args.device_name, 'info'])'info output: %s', output)
m =
return ('phy' + if m else None
def ChooseMaxStrengthService(self, services, service_strengths):
"""Chooses the service that has the largest signal strength among services.
services: A list of services.
service_strengths: A dict of strengths of each service.
The service that has the largest signal strength among services.
max_strength_service, max_strength = None, -sys.float_info.max
for service in services:
strength = service_strengths[service]['all']
if strength:'Service %s signal strength %f.', service,
event_log.Log('service_signal', service=service, strength=strength)
if strength > max_strength:
max_strength_service, max_strength = service, strength
else:'Service %s has no valid signal strength.',
if max_strength_service:'Service %s has the highest signal strength %f among %s.',
max_strength_service, max_strength, services)
return max_strength_service
logging.warning('Services %s are not valid.', services)
def ScanSignal(self, service, times=3):
"""Scans antenna signal strengths for a specified service.
Device should connect to the service before starting to capture signal.
Signal result only includes antenna information of this service
(ssid, freq).
service: (service_ssid, freq, password) tuple.
times: Number of times to scan to get average.
A list of signal result.
signal_list = []
(ssid, freq, password) = service
result = self.ConnectService(ssid, password)
if result is False:
return []
self._template.SetState(_MSG_SCANNING(self.args.device_name, freq))
with Capture(self.args.device_name, self._phy_name) as capture:
capture_times = 0
while capture_times < times:
signal_result = capture.get_signal()
if signal_result['ssid'] == ssid and signal_result['freq'] == freq:'%s', signal_result)
capture_times += 1
_MSG_SCANNING_DONE(self.args.device_name, freq))
return signal_list
def AverageSignals(self, antenna_info):
"""Averages signal strengths for each antenna of a service.
The dividend is the sum of signal strengths during all scans.
The divisor is the number of times in the scan result.
If a service is not scannable, its average value will be None.
antenna_info: A dict of each antenna information of a service.
A dict of average signal strength of each antenna.
{antenna1: signal1, antenna2: signal2}
# keys are services and values are averages
average_results = {}
# Averages the scanned strengths
for antenna in _ANTENNA_CONFIG:
average_results[antenna] = 0
for signal in antenna_info:
average_results['all'] += signal[0]
average_results['main'] += signal[1][1]
average_results['aux'] += signal[2][1]
for antenna in _ANTENNA_CONFIG:
average_results[antenna] = (
float(average_results[antenna]) / len(antenna_info)
if len(antenna_info) else None)
return average_results
def CheckSpec(self, service, spec_antenna_strength, average_signal):
"""Checks if the scan result of antenna config can meet test spec.
service: (service_ssid, freq, password) tuple.
spec_antenna_strength: A dict of minimal signal strengths.
average_signal: A dict of average signal strength of each service in
service. {service: {antenna1: signal1, antenna2: signal2}}
for antenna in _ANTENNA_CONFIG:
if spec_antenna_strength.get(antenna) is None:
spec_strength = spec_antenna_strength[antenna]
scanned_strength = average_signal[service][antenna]
'antenna_%s' % antenna, freq=service[1], rssi=scanned_strength,
meet=(scanned_strength and scanned_strength > spec_strength))
if not scanned_strength:
'Antenna %s, service: %s: Can not scan signal strength.' %
(antenna, service))
if scanned_strength < spec_strength:
'Antenna %s, service: %s: The scanned strength %f < spec strength'
' %f' % (antenna, service, scanned_strength, spec_strength))
'Antenna %s, service: %s: The scanned strength %f > spec strength'
' %f', antenna, service, scanned_strength, spec_strength)
def PreCheck(self, services):
"""Checks each service only has one frequency.
services: A list of (service_ssid, freq) tuples to scan.
wireless_services = {}
scan_result = IwScan(self.args.device_name)
set_all_ssids = set([service[0] for service in services])
for ssid, freq in scan_result:
if ssid in set_all_ssids:
if ssid not in wireless_services:
wireless_services[ssid] = freq
elif freq != wireless_services[ssid]:'There are more than one frequencies for ssid %s.' % ssid)
def PromptSpace(self):
"""Prompts a message to ask operator to press space."""
self._ui.BindKey(test_ui.SPACE_KEY, lambda _: self.OnSpacePressed())
self._ui.Run(blocking=False, on_finish=self.Done)
def Done(self):
"""The callback when ui is done.
This will be called when test is finished, or if operator presses
'Mark Failed'.
def OnSpacePressed(self):
"""The handler of space key."""'Space pressed by operator.')
def runTest(self):
if self.args.press_space_to_start:
# Prompts a message to tell operator to press space key when ready.
if self._done.isSet():
antenna_info = {}
for service in
antenna_info[service] = self.ScanSignal(service, self.args.scan_count)
average_signal = {}
for service, signals in antenna_info.iteritems():
average_signal[service] = self.AverageSignals(signals)
# Gets the service with the largest strength to test for each spec.
test_service = self.ChooseMaxStrengthService(,
if test_service is None:'Services %s are not valid.' %
self.CheckSpec(test_service, self.args.strength, average_signal)