blob: 5cf6a0008bae00cd26f820173670f5e4aa182107 [file] [log] [blame]
# Copyright (c) 2013 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.
# This test is used to verify the functionality of bluetooth device.
# The functionality under test are:
# 1. Detect the specified number of bluetooth adapter on dut.
# 2. Scan remote bluetooth device and try to find at least one device.
# 3. If a remote device keyword is given, the test will only care
# the devices whose 'Name' contains keyword. This applies to item 4 as well.
# 4. If an RSSI threshold value is given, check that the largest average RSSI
# among all scanned devices >= threshold.
# 5. Try to pair and connect with the bluetooth input device. Now it supports
# mouse.
# Check the ARGS in BluetoothTest for the detail of arguments.
import glob
import logging
import os
import sys
import threading
import time
import unittest
import factory_common # pylint: disable=unused-import
from cros.factory.device import device_utils
from cros.factory.test import event_log
from cros.factory.test import factory
from cros.factory.test import factory_task
from cros.factory.test.i18n import test_ui as i18n_test_ui
from cros.factory.test import shopfloor
from cros.factory.test import test_ui
from cros.factory.test import ui_templates
from cros.factory.test.utils import bluetooth_utils
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import process_utils
from cros.factory.utils import sync_utils
_TEST_TITLE = i18n_test_ui.MakeI18nLabel('Bluetooth functional Test')
_MSG_DETECT_ADAPTER = i18n_test_ui.MakeI18nLabel('Detect bluetooth adapter')
_MSG_TURN_ON_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
'Enable the connection ability of bluetooth device and press Enter',
_MSG_INTO_FIXTURE = i18n_test_ui.MakeI18nLabelWithClass(
'Place the base into the fixture, '
'and press the space key on the test host.', 'start-font-size')
_MSG_RESET_MAGNET = i18n_test_ui.MakeI18nLabelWithClass(
'Please re-attach the magnet, and press the space key on the test host.',
_MSG_START_CHARGE = i18n_test_ui.MakeI18nLabelWithClass(
'Turn on charging by pressing the green button, '
'take the keyboard out and put it back, '
'and press the space key on the test host.', 'start-font-size')
_MSG_READ_BATTERY_1 = i18n_test_ui.MakeI18nLabelWithClass(
'Read battery level for the 1st time.', 'start-font-size')
_MSG_READ_BATTERY_2 = i18n_test_ui.MakeI18nLabelWithClass(
'Read battery level for the 2nd time.', 'start-font-size')
_MSG_BATTERY_CHARGE_TEST = i18n_test_ui.MakeI18nLabelWithClass(
'Check if the battery has charged to a higher percentage',
_MSG_CHECK_BATTERY_LEVEL = i18n_test_ui.MakeI18nLabelWithClass(
'Check battery level.', 'start-font-size')
_MSG_STOP_CHARGE = i18n_test_ui.MakeI18nLabelWithClass(
'Press the green button again to stop charging, '
'and press the space key on the test host.', 'start-font-size')
_MSG_OUT_OF_FIXTURE = i18n_test_ui.MakeI18nLabelWithClass(
'Take the base out of the fixture, '
'and press the space key on the test host.', 'start-font-size')
_MSG_READ_FIRMWARE_REVISION_STRING = i18n_test_ui.MakeI18nLabelWithClass(
'Read firmware revision string.', 'start-font-size')
_MSG_SCAN_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
'Scanning...', 'start-font-size')
_MSG_DETECT_RSSI = lambda count, total: (
'Detect RSSI (count {count}/{total})',
'start-font-size', count=count, total=total))
_MSG_TURN_ON_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
'Enable the connection ability of input bluetooth device and press Enter',
_MSG_PAIR_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
'Pairing to input device now...', 'start-font-size')
_MSG_UNPAIR = i18n_test_ui.MakeI18nLabelWithClass(
'Press shift-p-a-i-r simultaneously on the base.', 'start-font-size')
_MSG_CONNECT_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
'Connecting to input device now...', 'start-font-size')
_MSG_TEST_INPUT = i18n_test_ui.MakeI18nLabelWithClass(
'Please test input. Press Escape to fail and Enter to pass',
_MSG_UNPAIRING = i18n_test_ui.MakeI18nLabelWithClass(
'Unpairing', 'start-font-size')
_MSG_AUTH_FAILED = i18n_test_ui.MakeI18nLabelWithClass(
'Authentication failed, retrying...', 'start-font-size')
BATTERY_LEVEL_KEY = 'battery_level'
READ_BATTERY_STEP_1 = 'read_battery_1'
READ_BATTERY_STEP_2 = 'read_battery_2'
def GetCurrentTime():
"""Get the current time."""
return time.strftime('%Y-%m-%d %H:%M:%S')
def ColonizeMac(mac):
""" Given a MAC address, normalize its colons.
Example: ABCDEF123456 -> AB:CD:EF:12:34:56
mac_no_colons = ''.join(mac.strip().split(':'))
groups = [mac_no_colons[x:x+2] for x in range(0, len(mac_no_colons), 2)]
return ':'.join(groups)
def MakePasskeyLabelPrompt(passkey):
"""Creates a label prompting the operator to enter a passkey"""
return i18n_test_ui.MakeI18nLabelWithClass(
'Enter passkey {key} then press enter on the base.',
def CheckInputCount():
"""Returns the number of input devices from probing /dev/input/event*."""
number_input = len(glob.glob('/dev/input/event*'))'Found %d input devices.', number_input)
return number_input
def WaitForInputCount(task, expected_input_count, timeout=10):
""" Waits for the number of input devices to reach the given count.
Returns true if the input count reaches the given amount, false
otherwise. On failure it fails the task.
task: The task that may be Failed
expected_input_count: The number of input devices that determines success
timeout: The maximum time in seconds that we will wait
end_time = time.time() + timeout
while time.time() < end_time:
input_count = CheckInputCount()
if input_count == expected_input_count:
return True
task.Fail('Input device count %d is different than expected %d.' %
(input_count, expected_input_count))
return False
def _ResetAdapter():
"""Reset the adapter every time when using the BT device.
This is because the adapter may be down anytime for some unknown reason.
cmd = 'hciconfig hci0 reset''Reset adapter and wait 5 seconds....: %s', cmd)
process_utils.Spawn(cmd.split(), log=True, check_call=True)
def _SaveLocalLog(log_file, data):
"""Save the log locally on a test host."""
log_dir = os.path.dirname(log_file)
if not os.path.isdir(log_dir):
with open(log_file, 'a') as log:
def _SaveAuxLogOnShopfloor(aux_log_file, data):
"""Save the local log file to shopfloor."""
shopfloor_client = shopfloor.GetShopfloorConnection()
shopfloor_client.SaveAuxLog(aux_log_file, str(data))
except Exception as e:
# It is only a logging error. Do not fail the test.
logging.warning('Save aux log failure: %s', e)
def _SaveLogs(log_file, aux_log_file, data):
"""Save the log files on the local test host and on the shopfloor."""
# Prepend the current timestamp to each line.
data = ''.join([(GetCurrentTime() + ' ' + line + '\n') if line else '\n'
for line in data.splitlines()])
if log_file:
_SaveLocalLog(log_file, data)
if aux_log_file:
with open(log_file) as log:
def RetryWithProgress(template, template_message, action_string,
max_retry_times, retry_interval, target, *args, **kwargs):
"""Runs target function with retries and shows retry times on progress bar.
template: an ui_template
template_message: The message to show when target function is running.
action_string: The string to describe the action in logging.
max_retry_times: the maximal retry times
retry_interval: the interval between retries
target: The target function. *args and **kwargs will be passed to target.
Return the return value of the target function.
def _UpdateProgressBar(retry_time, max_retry_time):
"""Updates the progress bar according to retry_time and max_retry_time."""
msg = 'Update progress bar with retry time: %d, max retry time: %d.', retry_time, max_retry_time)
template.SetProgressBarValue(int(100 * retry_time / max_retry_time))
target_result = sync_utils.Retry(max_retry_times, retry_interval,
_UpdateProgressBar, target, *args, **kwargs)
log_msg = '%s was done.' if target_result else '%s failed.', action_string)
return target_result
class DetectAdapterTask(factory_task.FactoryTask):
"""The task checking number of adapters.
Detects adapters from dbus and checks if the number of adapters matches the
expected number.
expected_adapter_count: The expected number of bluetooth adapters.
def __init__(self, test, expected_adapter_count):
super(DetectAdapterTask, self).__init__()
self._test = test
self._expected_adapter_count = expected_adapter_count
def Run(self):
adapters = self._test.dut.bluetooth.GetAdapters(
if len(adapters) == self._expected_adapter_count:
self.Fail('DetectAdapter: expect %d and find %d adapter(s).' %
(self._expected_adapter_count, len(adapters)))
class TurnOnTask(factory_task.FactoryTask):
"""The task to ask operator to turn on bluetooth device and press enter.
message: Html code containing message to show on the screen.
def __init__(self, test, message, key=test_ui.ENTER_KEY):
super(TurnOnTask, self).__init__()
self._test = test
self._message = message
self._key = key
def Cleanup(self):
"""Unbinds the Enter key after this task is done."""
def Run(self):
self._test.ui.AppendCSS('.start-font-size {font-size: 2em;}')
self._test.ui.BindKey(self._key, lambda _: self.Pass())'wait for the user to press a key')
def SetAndStartScanProgressBar(template, timeout_secs, scan_event=None):
"""Control progress bar fo a duration of timeout_secs."""
def UpdateProgressBar():
"""Updates progress bar for a duration of timeout_secs"""
start_time = time.time()
end_time = start_time + timeout_secs
while time.time() < end_time:
if scan_event and scan_event.isSet():
100 * (time.time() - start_time) / timeout_secs))
logging.debug('UpdateProgressBar stopped.')
return process_utils.StartDaemonThread(
target=UpdateProgressBar, name='ProgressThread')
class ScanDevicesTask(factory_task.FactoryTask):
"""The task to scan bluetooth devices around.
In this task, the test will control the first adapter from BluetoothManager
and scan devices around for timeout_secs. The task passed if there is at least
one device.
If target_addresses is provided, the test will also check if it can find
at least one device specified in target_addresses list.
This passes the strongest matching device mac to _test.SetStrongestRssiMac
Note: this task is intended to be executed on a DUT, i.e., a chromebook,
to test its bluetooth module. A typical test case is to see if it can detect
a bluetooth mouse placed around it.
def __init__(self, test):
super(ScanDevicesTask, self).__init__()
self._test = test
self._keyword = test.args.keyword
self._average_rssi_threshold = test.args.average_rssi_threshold
self._mac_to_scan = test.GetInputDeviceMac()
self._scan_counts = test.args.scan_counts
self._timeout_secs = test.args.scan_timeout_secs
self._progress_thread = None
def FilterByKeyword(self, devices):
"""Returns the devices filtered by self._keyword.
If self._keyword is None, leave devices as it is.
if self._keyword is None:
return devices
filtered_devices = {}
for mac, props in devices.iteritems():
if 'Name' not in props:
logging.warning('Device %s: %s does not have "Name" property.',
mac, props)
if self._keyword in props['Name']:
filtered_devices[mac] = props'Device %s: "Name" property %s matches keyword %s.',
mac, props['Name'], self._keyword)
return filtered_devices
def UpdateRssi(self, devices_rssis, devices):
"""Updates devices_rssis using RSSI property in devices.
devices_rssis: A dict. Keys are mac addresses and values are lists of
scanned RSSI value.
devices: A dict. Keys are mac addresses and values are dicts of
for mac, props in devices.iteritems():
if 'RSSI' not in props:
logging.warning('Device %s: %s does not have "RSSI" property.',
mac, props)
if mac in devices_rssis:
devices_rssis[mac] = [props['RSSI']]'UpdateRssi: %s', devices_rssis)
def Run(self):
bluetooth_manager = self._test.dut.bluetooth
adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
# Records RSSI of each scan and calculates average rssi.
candidate_rssis = {}
# Helper to check if the target MAC has been scanned
def has_scanned_target_mac():
return self._mac_to_scan and self._mac_to_scan in candidate_rssis
for _ in xrange(self._scan_counts):
self._progress_thread = SetAndStartScanProgressBar(self._test.template,
devices = bluetooth_manager.ScanDevices(adapter, self._timeout_secs)
self._progress_thread.join()'Found %d device(s).', len(devices))
for mac, props in devices.iteritems():
try:'Device found: %s. Name: %s, RSSI: %d',
mac, props['Name'], props['RSSI'])
except KeyError:
logging.exception('Name or RSSI is not available in %s', mac)
self.UpdateRssi(candidate_rssis, self.FilterByKeyword(devices))
# Optimization: if we are only interested in one particular address,
# then we can early-out as soon as we find it
if self._average_rssi_threshold is None and has_scanned_target_mac():"Address found, ending scan early")
break'Found %d candidate device(s) in %s scans.',
len(candidate_rssis), self._scan_counts)'Candidate devices scan results: %s',
dict((str(k), [int(r) for r in v])
for k, v in candidate_rssis.iteritems()))
if len(candidate_rssis) == 0:
self.Fail('ScanDevicesTask: Fail to find any candidate device.')
# Calculates maximum average RSSI.
max_average_rssi_mac, max_average_rssi = None, -sys.float_info.max
for mac, rssis in candidate_rssis.iteritems():
# typecast to str to avoid the weird dbus.String type
mac = str(mac)
average_rssi = float(sum(rssis)) / len(rssis)'Device %s has average RSSI: %f', mac, average_rssi)
event_log.Log('avg_rssi', mac=mac, average_rssi=average_rssi)
if average_rssi > max_average_rssi:
max_average_rssi_mac, max_average_rssi = mac, average_rssi'Device %s has the largest average RSSI: %f',
max_average_rssi_mac, max_average_rssi)
event_log.Log('bluetooth_scan_device', mac=str(max_average_rssi_mac),
meet=max_average_rssi >= self._average_rssi_threshold)
if self._mac_to_scan and not has_scanned_target_mac():
found_addresses = [str(k) for k in candidate_rssis]
self.Fail('Failed to find MAC address %s.'
'Scanned addresses: %s' % (self._mac_to_scan, found_addresses))
if self._average_rssi_threshold is None:
# Test is uninterested in RSSI thresholds
elif self._average_rssi_threshold > max_average_rssi:
factory.console.error('The largest average RSSI %f does not meet'
' threshold %f. Please ensure that the test BT '
'device is \'visible\' and close to the DUT '
max_average_rssi, self._average_rssi_threshold)
self.Fail('ScanDeviceTask: The largest average RSSI %f of device %s does'
' not meet threshold %f.' % (
max_average_rssi, str(max_average_rssi_mac),
else:'The largest average RSSI %f meets threshold %f.',
max_average_rssi, self._average_rssi_threshold)
class DetectRSSIofTargetMACTask(factory_task.FactoryTask):
"""The task to detect the RSSI strength at a given target MAC address.
In this task, a generic test host uses the first adapter from
BluetoothManager and scans devices around for timeout_secs. The task
passed if it can detect the RSSI strength at the target MAC.
Note: this task is intended to be executed on a generic test host to test
if the RSSI of a target device, e.g., a Ryu base, could be detected.
def __init__(self, test):
super(DetectRSSIofTargetMACTask, self).__init__()
self._test = test
self._mac_to_scan = test.GetInputDeviceMac()
self._scan_counts = test.args.scan_counts
self._timeout_secs = test.args.scan_timeout_secs
self._input_device_rssi_key = test.args.input_device_rssi_key
self._progress_thread = None
self._scan_rssi_event = threading.Event()
self.fail_msg = ''
self._average_rssi_lower_threshold = None
self._average_rssi_upper_threshold = None
def _DeriveRSSIThreshold(self, threshold, fid):
if isinstance(threshold, (int, float)):
return threshold
elif isinstance(threshold, dict):
if fid in threshold:
return threshold.get(fid)
self.fail_msg += 'Fixture ID "%s" is not legitimate!\n' % fid
self.fail_msg += 'Wrong type of RSSI threshold: %s\n' % str(threshold)
def Run(self):
fid = event_log.GetDeviceId()
self._average_rssi_lower_threshold = self._DeriveRSSIThreshold(
self._test.args.average_rssi_lower_threshold, fid)
self._average_rssi_upper_threshold = self._DeriveRSSIThreshold(
self._test.args.average_rssi_upper_threshold, fid)
if self.fail_msg:
bluetooth_manager = self._test.dut.bluetooth
adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)'mac (%s): %s', self._test.host_mac, adapter)
rssis = []
for i in xrange(1, 1 + self._scan_counts):
label = _MSG_DETECT_RSSI(i, self._scan_counts)
self._progress_thread = SetAndStartScanProgressBar(self._test.template,
devices = bluetooth_manager.ScanDevices(adapter,
for mac, props in devices.iteritems():
if mac == self._mac_to_scan and 'RSSI' in props:'RSSI of count %d: %.2f', i, props['RSSI'])
if len(rssis) == 0:
self.Fail('DetectRSSIofTargetMACTask: Fail to get RSSI from device %s.' %
average_rssi = float(sum(rssis)) / len(rssis)
factory.set_shared_data(self._input_device_rssi_key, average_rssi)'RSSIs at MAC %s: %s', self._mac_to_scan, rssis)'Average RSSI: %.2f', average_rssi)
fail_msg = ''
if (self._average_rssi_lower_threshold is not None and
average_rssi < self._average_rssi_lower_threshold):
fail_msg += ('Average RSSI %.2f less than the lower threshold %.2f\n' %
(average_rssi, self._average_rssi_lower_threshold))
if (self._average_rssi_upper_threshold is not None and
average_rssi > self._average_rssi_upper_threshold):
fail_msg += ('Average RSSI %.2f greater than the upper threshold %.2f' %
(average_rssi, self._average_rssi_upper_threshold))
# Convert dbus.Int16 in rssis below to regular integers.
status = (('pass' if fail_msg == '' else 'fail') +
' exp: [%.2f, %.2f]' % (self._average_rssi_lower_threshold,
data = ('Average RSSI: %.2f %s (%s)\n' %
(average_rssi, map(int, rssis), status))
_SaveLogs(self._test.log_file, self._test.aux_log_file, data)
if fail_msg:
class UnpairTask(factory_task.FactoryTask):
"""A task to unpair from bluetooth devices.
device_mac: None, or the MAC address of the device to unpair
name_fragment: None, or a substring of the name of the device(s) to unpair
def __init__(self, test, device_mac, name_fragment):
super(UnpairTask, self).__init__()
self._test = test
self._device_mac = device_mac
self._name_fragment = name_fragment
def _ShouldUnpairDevice(self, device_props):
"""Indicate if a device matches the filter, and so should be unpaired
If a name fragment or MAC address is given, the corresponding property
must match. If neither is given, all devices should be unpaired.
if self._device_mac and device_props["Address"] != self._device_mac:
return False
if (self._name_fragment and
self._name_fragment not in device_props.get('Name', '')):
return False
return device_props["Paired"]
def Run(self):
self._test.ui.AppendCSS('.start-font-size {font-size: 2em;}')
input_count_before_unpair = CheckInputCount()
bluetooth_manager = self._test.dut.bluetooth
adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
devices = bluetooth_manager.GetAllDevices(adapter).values()
devices_to_unpair = filter(self._ShouldUnpairDevice, devices)'Unpairing %d device(s)', len(devices_to_unpair))
for device_to_unpair in devices_to_unpair:
address = device_to_unpair["Address"]
bluetooth_manager.DisconnectAndUnpairDevice(adapter, address)
bluetooth_manager.RemovePairedDevice(adapter, address)
# Check that we unpaired what we thought we did
expected_input_count = input_count_before_unpair - len(devices_to_unpair)
if WaitForInputCount(self, expected_input_count):
class CheckDisconnectionOfPairedDeviceTask(factory_task.FactoryTask):
"""A task to check whether a paired device has disconnected.
device_mac: None, or the MAC address of the device to unpair
name_fragment: None, or a substring of the name of the device(s) to unpair
def __init__(self, test, device_mac):
super(CheckDisconnectionOfPairedDeviceTask, self).__init__()
self._test = test
self._device_mac = device_mac
def _ConnectedDevice(self, device_props):
"""Indicates if a device matches the filter, and so should be unpaired
If a name fragment or MAC address is given, the corresponding property
must match. If neither is given, all devices should be unpaired.
return (device_props["Address"] == self._device_mac and
int(device_props["Connected"]) >= 1)
def _CheckDisconnection(self):
bluetooth_manager = self._test.dut.bluetooth
adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
devices = bluetooth_manager.GetAllDevices(adapter).values()
connected_devices = filter(self._ConnectedDevice, devices)'Connected and paired %d device(s)', len(connected_devices))
return len(connected_devices) == 0
def Run(self):
flag_disconnection = RetryWithProgress(
self._test.template, _MSG_UNPAIR,
'Check disconnection of the paired base',
if flag_disconnection:
msg = 'Shift-P-A-I-R: done'
msg = 'Shift-P-A-I-R: not done'
_SaveLogs(self._test.log_file, self._test.aux_log_file, msg)
def _ExecuteFixtureMethod(fixture, operation, post_sleep=0):
"""Execute a method of the charge test fixture."""
# An operation is mapped to its corresponding fixture method defined in
# base_charge_fixture.BaseChargeFixture class.
'STOP_CHARGING': 'StopCharging',
'ENABLE_MAGNET': 'EnableMagnet',
'DISABLE_MAGNET': 'DisableMagnet'}
fixture_method = getattr(fixture, FIXTURE_METHOD_DICT.get(operation))'Executing fixture method: %s', fixture_method.__name__)
class FixtureControlTask(factory_task.FactoryTask):
"""The task to control the charge test fixture.
operation: the operation to be performed by the test fixture.
def __init__(self, test, operation, post_sleep=0):
# pylint: disable=super-init-not-called
self._fixture = test.fixture
self._operation = operation
self._post_sleep = post_sleep
def Run(self):
_ExecuteFixtureMethod(self._fixture, self._operation,
except Exception as e:
self.Fail('error in executing %s (%s)' % (self._operation, e))
def _SaveLocalBatteryLog(base_enclosure_serial_number, mac, step,
battery_level, log_filename):
"""Save the battery log on the local test host."""
with open(log_filename, 'a') as f:
f.write('%s %s %s [%s]: %s\n' %
(GetCurrentTime(), base_enclosure_serial_number, mac, step,
class ReadBatteryLevelTask(factory_task.FactoryTask):
"""A class to read battery level."""
def __init__(self, test, mac, step):
super(ReadBatteryLevelTask, self).__init__()
self._test = test
self._mac = mac
self._step = step
def Run(self):
self._test.template.SetState(self.MSG_DICT.get(self._step))'%s via %s ...', self._step, self._test.hci_device)
battery_level = int(RetryWithProgress(
self._test.template, self.MSG_DICT.get(self._step), self._step,
self._mac, 'battery level', hci_device=self._test.hci_device,
timeout=self._test.args.read_bluetooth_uuid_timeout_secs))'%s: %d', self._step, battery_level)
except bluetooth_utils.BluetoothUtilsError as e:
self.Fail('%s failed to get battery level: %s' % (self._step, e))
old_battery_level = factory.get_shared_data(self._step)
if (self._step == READ_BATTERY_STEP_1 and
(old_battery_level is None or battery_level < old_battery_level)):
# If the battery level at step 1 becomes higher over different rounds
# (when the operator keeps retesting it for any reasons),
# we only keep the lowest one. This is because we want to test if the
# battery could charge to a higher level at step 2 than step 1.
factory.set_shared_data(self._step, battery_level)
elif self._step == READ_BATTERY_STEP_2:
# We keep the latest battery level read at step 2.
factory.set_shared_data(self._step, battery_level)
if self._step == READ_BATTERY_STEP_1:
data = ('\nSN: %s\nMAC: %s\n' %
(self._test.args.base_enclosure_serial_number, self._mac))
data = ''
data += '%s: %s\n' % (self._step, battery_level)
_SaveLogs(self._test.log_file, self._test.aux_log_file, data)
if self._test.args.battery_log:
self._mac, self._step, battery_level,
class CheckBatteryLevelTask(factory_task.FactoryTask):
"""This battery level test checks whether the following condistions
are satisfied:
1. The battery levels are read twice.
2. battery_level_1 < battery_level_2
3. battery_level_1 >= expected_battery_level
def __init__(self, test):
super(CheckBatteryLevelTask, self).__init__()
self._test = test
def Run(self):
battery_level_1 = factory.get_shared_data(READ_BATTERY_STEP_1)
battery_level_2 = factory.get_shared_data(READ_BATTERY_STEP_2)'%s: %s', READ_BATTERY_STEP_1, str(battery_level_1))'%s: %s', READ_BATTERY_STEP_2, str(battery_level_2))
if not battery_level_1 or not battery_level_2:
fail_msg = 'Battery levels should be read twice. read_1: %s, read_2: %s'
elif (battery_level_1 > battery_level_2 or
(battery_level_1 == battery_level_2 and battery_level_1 < 100)):
fail_msg = 'Base battery is not charged up. read_1: %s, read_2: %s'
elif battery_level_1 < self._test.args.expected_battery_level:
# Note: battery_level_1 instead of battery_level_2 should be larger than
# the expected_battery_level since battery_level_2 is read while
# charging and its value is usually larger than its actual value.
fail_msg = 'Measured battery level %s is less than the expected level %s.'
self.Fail(fail_msg % (battery_level_1, battery_level_2))
class ChargeTestTask(factory_task.FactoryTask):
def __init__(self, test, mac, step):
super(ChargeTestTask, self).__init__()
self._test = test
self._mac = mac
self._step = step
def ReadBatteryLevel(self, step):
if self._test.args.use_charge_fixture:
_ExecuteFixtureMethod(self._test.fixture, 'ENABLE_MAGNET')'Begin reading battery level...')
value = bluetooth_utils.GattTool.GetDeviceInfo(
self._mac, 'battery level', hci_device=self._test.hci_device,
if self._test.args.use_charge_fixture:
_ExecuteFixtureMethod(self._test.fixture, 'DISABLE_MAGNET')'%s: %s', step, value)
return int(value)
def Run(self):
if self._step == READ_BATTERY_STEP_1:
battery_level = self.ReadBatteryLevel(self._step)
factory.set_shared_data(BATTERY_LEVEL_KEY, battery_level)
elif self._step == READ_BATTERY_STEP_2:
def _ReadAndCheckBatteryLevel():
battery_level2 = self.ReadBatteryLevel(self._step)
result = battery_level2 > battery_level1
if result:
factory.set_shared_data(BATTERY_LEVEL_KEY, battery_level2)
return result
battery_level1 = factory.get_shared_data(BATTERY_LEVEL_KEY)
# Check if the battery is charging for up to READ_BATTERY_MAX_RETRY_TIMES.
# Note: the magnet needs to be taken away and re-applied each time.
# This operation could be performed automatically with a charging
# test fixture; otherwise, it must be performed manually. Also
# note that there is a 5-second delay at reading the battery level.
count = 0
success_increased_level = False
while (not success_increased_level and
success_increased_level = _ReadAndCheckBatteryLevel()
count += 1
if success_increased_level:
self.Fail('ChargeTestTask: the battery is not charging!')
class CheckFirmwareRevisionTestTask(factory_task.FactoryTask):
"""A factory task class to read firmware revision string."""
def __init__(self, test, mac):
super(CheckFirmwareRevisionTestTask, self).__init__()
self._test = test
self._mac = mac
def Run(self):
self._test.template.SetState(_MSG_READ_FIRMWARE_REVISION_STRING)'Begin reading firmware revision string via %s...',
fw = RetryWithProgress(
bluetooth_utils.GattTool.GetDeviceInfo, self._mac,
'firmware revision string', hci_device=self._test.hci_device,
except bluetooth_utils.BluetoothUtilsError as e:
self.Fail('Failed to get firmware revision string: %s' % e)
return'Expected firmware: %s',
self._test.args.firmware_revision_string)'Actual firmware: %s', fw)
factory.set_shared_data(self._test.args.firmware_revision_string_key, fw)
data = 'FW: %s\n' % fw
_SaveLogs(self._test.log_file, self._test.aux_log_file, data)
if fw == self._test.args.firmware_revision_string:
self.Fail('Expected firmware: %s, actual firmware: %s' %
(self._test.args.firmware_revision_string, fw))
class InputTestTask(factory_task.FactoryTask):
"""The task to test bluetooth input device functionality.
The task will try to pair with the device given by the test,
and make the connection.
After the connection, the number of input event should plus one.
If it does not plus one, the task fails.
After connection, operator can try to use the input device and press Enter
to pass checking or Esc to fail the task.
In the end of test, the task will try to disconnect the device and remove the
device. If these procedures fail, the task fails.
finish_after_pair: Whether to end the test after pairing. If false,
the operator is prompted to test input, and then
the device is unpaired
def __init__(self, test, finish_after_pair):
super(InputTestTask, self).__init__()
self._test = test
self._target_mac = None
self._bt_manager = None
self._adapter = None
self._need_to_cleanup = True
self._finish_after_pair = finish_after_pair
def Cleanup(self):
"""Cleans up input device if it was not cleaned"""
if self._need_to_cleanup:
success_to_remove = self.RemoveInput()
if not success_to_remove:
logging.error('Fail to remove input in Cleanup')
def RemoveInput(self):
"""Disconnects the input device and removes it.
If disconnection and removal are both succeeded, return True, return False
return_value = True
self._bt_manager.SetDeviceConnected(self._adapter, self._target_mac,
False)'Turned off the connection')
except self._bt_manager.Error:
logging.exception('Fail to turn off the connection.')
return_value = False
self._bt_manager.RemovePairedDevice(self._adapter, self._target_mac)'Remove the device')
except self._bt_manager.Error:
logging.exception('Fail to remove the device.')
return_value = False
return return_value
def RemoveInputAndQuit(self, success):
"""Removes the input device and quits the task.
success: The task is passed by operator or not.
If the task is passed by operator and input has been removed successfully,
pass the task, fail the task otherwise.
success_to_remove = self.RemoveInput()
# No need to cleanup again after the task does Pass() or Fail() if removal
# succeeds here.
self._need_to_cleanup = not success_to_remove
if success:
if success_to_remove:
self.Fail('InputTestTask: Fail to remove input')
self.Fail('Failed by operator')
def OperatorTestInput(self):
"""Lets operator test the input and press key to pass/fail the task."""'InputTestTask: Test the input by operator now')
lambda _: self.RemoveInputAndQuit(True))
lambda _: self.RemoveInputAndQuit(False))
def Run(self):
def SaveLogAndFail(fail_reason):
"""Save the fail log and invoke Fail()."""
data = 'Pairing fail: %s\n' % fail_reason
_SaveLogs(self._test.log_file, self._test.aux_log_file, data)
input_count_before_connection = CheckInputCount()
self._bt_manager = self._test.dut.bluetooth
self._adapter = self._bt_manager.GetFirstAdapter(self._test.host_mac)
self._target_mac = self._test.GetInputDeviceMac()
if not self._target_mac:
SaveLogAndFail('InputTestTask: No MAC with which to pair')'Attempting pair with %s', self._target_mac)
self._bt_manager.DisconnectAndUnpairDevice(self._adapter, self._target_mac)
success_create_device = RetryWithProgress(
self._test.template, _MSG_PAIR_INPUT_DEVICE, 'create paired device',
self._bt_manager.CreatePairedDevice, self._adapter,
self._target_mac, self._DisplayPasskey, self._AuthenticationCancelled)
if not success_create_device:
SaveLogAndFail('InputTestTask: Fail to create paired device.')
success_connect_device = RetryWithProgress(
self._test.template, _MSG_CONNECT_INPUT_DEVICE, 'connect input device',
self._bt_manager.SetDeviceConnected, self._adapter,
self._target_mac, True)
if not success_connect_device:
SaveLogAndFail('InputTestTask: Fail to connect device.')
if not WaitForInputCount(self, input_count_before_connection + 1):
if self._finish_after_pair:
# We leave the device paired
self._need_to_cleanup = False
data = 'Pairing finished\n'
_SaveLogs(self._test.log_file, self._test.aux_log_file, data)
def _DisplayPasskey(self, passkey):"Displaying passkey %s", passkey)
label = MakePasskeyLabelPrompt(passkey)
def _AuthenticationCancelled(self):
class BluetoothTest(unittest.TestCase):
ARGS = [
Arg('expected_adapter_count', int, 'Number of bluetooth adapters'
' on the machine.', default=0),
Arg('manufacturer_id', int, 'manufacturer id', optional=True),
Arg('detect_adapters_retry_times', int, 'Maximum retry time to'
' detect adapters', default=10),
Arg('detect_adapters_interval_secs', int, 'Interval in seconds between'
' each retry to detect adapters', default=2),
Arg('read_bluetooth_uuid_timeout_secs', int,
'Timeout to read bluetooth characteristics via uuid', default=None,
Arg('scan_devices', bool, 'Scan bluetooth device.',
Arg('prompt_scan_message', bool, 'Prompts a message to tell user to'
' enable remote devices discovery mode', default=True),
Arg('keyword', str, 'Only cares remote devices whose "Name" contains'
' keyword.', default=None, optional=True),
Arg('average_rssi_threshold', float, 'Checks the largest average RSSI'
' among scanned device is equal to or greater than '
' average_rssi_threshold.',
default=None, optional=True),
Arg('scan_counts', int, 'Number of scans to calculate average RSSI',
Arg('scan_timeout_secs', int, 'Timeout to do one scan', default=5),
Arg('input_device_mac', str, 'The mac address of bluetooth input device',
default=None, optional=True),
Arg('input_device_mac_key', str, 'A key for factory shared data '
'containing the mac address', default=None, optional=True),
Arg('input_device_rssi_key', str, 'A key for factory shared data '
'containing the rssi value', default=None, optional=True),
Arg('firmware_revision_string_key', str,
'A key of factory shared data containing firmware revision string',
Arg('firmware_revision_string', str,
'the firmware revision string', optional=True),
Arg('average_rssi_lower_threshold', (float, dict), 'Checks the average'
' RSSI of the target mac is equal to or greater than this threshold.',
default=None, optional=True),
Arg('average_rssi_upper_threshold', (float, dict), 'Checks the average'
' RSSI of the target mac is equal to or less than this threshold.',
default=None, optional=True),
Arg('pair_with_match', bool, 'Whether to pair with the strongest match.',
default=False, optional=True),
Arg('finish_after_pair', bool, 'Whether the test should end immediately '
'after pairing completes', default=False),
Arg('unpair', bool, 'Whether to unpair matching devices instead of pair',
default=False, optional=True),
Arg('check_shift_pair_keys', bool,
'check if shift-p-a-i-r keys are pressed.',
default=False, optional=True),
Arg('check_battery_charging', bool,
'Whether to check if the battery is charging',
default=False, optional=True),
Arg('read_battery_level', int, 'read the battery level',
default=None, optional=True),
Arg('check_battery_level', bool, 'Whether to check the battery level',
default=False, optional=True),
Arg('prompt_into_fixture', bool, 'Prompt the user to place the base into '
'the test fixture', default=False, optional=True),
Arg('use_charge_fixture', bool, 'whether a charge fixture is employed',
default=False, optional=True),
Arg('reset_fixture', bool, 'whether to reset the fixture',
default=False, optional=True),
Arg('start_charging', bool, 'Prompt the user to start charging the base',
default=False, optional=True),
Arg('enable_magnet', bool, 'enable the base',
default=False, optional=True),
Arg('reset_magnet', bool, 'reset the base',
default=False, optional=True),
Arg('stop_charging', bool, 'Prompt the user to stop charging the base',
default=False, optional=True),
Arg('base_enclosure_serial_number', unicode,
'the base enclosure serial number', default=None, optional=True),
Arg('battery_log', str,
'the battery log file', default=None, optional=True),
Arg('expected_battery_level', int,
'the expected battery level', default=100, optional=True),
Arg('log_path', str, 'the directory of the log on the local test host',
Arg('aux_log_path', str, 'the path of the aux log on shopfloor',
Arg('test_host_id_file', str, 'the file storing the id of the test host',
def SetStrongestRssiMac(self, mac_addr):
self._strongest_rssi_mac = mac_addr
def GetInputDeviceMac(self):
"""Gets the input device MAC to pair with, or None if None
This may be specified in the arguments, or computed at scan time.
if self._input_device_mac:
return self._input_device_mac
return self._strongest_rssi_mac
def setUp(self):
self.dut = device_utils.CreateDUTInterface()
self.ui = test_ui.UI()
self.ui.AppendCSS('.start-font-size {font-size: 2em;}')
self.template = ui_templates.TwoSections(self.ui)
self._task_list = []
self._strongest_rssi_mac = None
self.fixture = None
if self.args.input_device_mac_key:
self._input_device_mac = (
self._input_device_mac = self.args.input_device_mac
self.btmgmt = bluetooth_utils.BtMgmt(self.args.manufacturer_id)
self.hci_device = self.btmgmt.GetHciDevice()
self.host_mac = self.btmgmt.GetMac()'manufacturer_id %s: %s %s',
self.args.manufacturer_id, self.hci_device, self.host_mac)
if self.args.base_enclosure_serial_number:
if (self.args.test_host_id_file and
with open(self.args.test_host_id_file) as f:
test_host_id =
test_host_id = None
filename = '.'.join([self.args.base_enclosure_serial_number,
self.log_file = None
self.aux_log_file = None
if self.args.log_path:
self.log_file = os.path.join(self.args.log_path, filename)
# Note: aux_log_file is generated from log_file
# Not all projects would generate aux_log_file.
if self.args.aux_log_path:
self.aux_log_file = os.path.join(self.args.aux_log_path, filename)
def tearDown(self):
"""Close the charge test fixture."""
if self.args.use_charge_fixture:
def runTest(self):
if self.args.use_charge_fixture:
# Import this module only when a test station needs it.
# A base SMT test station does not need to use the charge fixture.
# pylint: disable=E0611
from cros.factory.test.fixture import base_charge_fixture
# Note: only reset the fixture in InitializeFixture test.
# This will stop charging and disable the magnet initially.
# For the following tests, do not reset the fixture so that
# the charging could be continued across tests in the test list
# defined in the base_host. The purpose is to keep charing the
# battery while executing other tests.
self.fixture = base_charge_fixture.BaseChargeFixture(
if self.args.expected_adapter_count:
self, self.args.expected_adapter_count))
if self.args.scan_devices:
if self.args.prompt_scan_message:
self._task_list.append(TurnOnTask(self, _MSG_TURN_ON_DEVICE))
if self.args.input_device_rssi_key:
if self.args.prompt_into_fixture:
TurnOnTask(self, _MSG_INTO_FIXTURE, test_ui.SPACE_KEY))
if self.args.read_battery_level == 1:
ReadBatteryLevelTask(self, self._input_device_mac,
if self.args.enable_magnet and self.args.use_charge_fixture:
self._task_list.append(FixtureControlTask(self, 'ENABLE_MAGNET'))
if self.args.reset_magnet:
if self.args.use_charge_fixture:
FixtureControlTask(self, 'DISABLE_MAGNET', post_sleep=1))
self._task_list.append(FixtureControlTask(self, 'ENABLE_MAGNET'))
TurnOnTask(self, _MSG_RESET_MAGNET, test_ui.SPACE_KEY))
if self.args.start_charging:
if self.args.use_charge_fixture:
# Let it charge for a little while.
FixtureControlTask(self, 'START_CHARGING'))
TurnOnTask(self, _MSG_START_CHARGE, test_ui.SPACE_KEY))
if self.args.check_shift_pair_keys:
CheckDisconnectionOfPairedDeviceTask(self, self._input_device_mac))
if self.args.unpair:
UnpairTask(self, self._input_device_mac, self.args.keyword))
if self.args.firmware_revision_string:
CheckFirmwareRevisionTestTask(self, self._input_device_mac))
if self.args.pair_with_match:
self._task_list.append(InputTestTask(self, self.args.finish_after_pair))
if self.args.read_battery_level == 2:
ReadBatteryLevelTask(self, self._input_device_mac,
if self.args.check_battery_level:
if self.args.stop_charging:
if self.args.use_charge_fixture:
self._task_list.append(FixtureControlTask(self, 'STOP_CHARGING'))
TurnOnTask(self, _MSG_STOP_CHARGE, test_ui.SPACE_KEY))
factory_task.FactoryTaskManager(self.ui, self._task_list).Run()