blob: a01e879c45e61e64d107c01673ccc3f33b1bf9c4 [file] [log] [blame]
# Copyright (c) 2012 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.
import logging, threading, time
from autotest_lib.client.common_lib.cros import crash_detector
from autotest_lib.server import autotest, test
from autotest_lib.client.common_lib import error
_LOWER_USB_PORT = 'usb_mux_sel3'
_UPPER_USB_PORT = 'usb_mux_sel1'
class platform_ExternalUsbPeripherals(test.test):
"""Uses servo to repeatedly connect/remove USB devices during boot."""
version = 1
def getPluggedUsbDevices(self):
"""Determines the external USB devices plugged
@returns plugged_list: List of plugged usb devices names
lsusb_output ='lsusb').stdout.strip()
items = lsusb_output.split('\n')
plugged_list = []
unnamed_device_count = 1
for item in items:
columns = item.split(' ')
if len(columns) == 6 or len(' '.join(columns[6:]).strip()) == 0:
logging.debug('Unnamed device located, adding generic name.')
name = 'Unnamed device %d' % unnamed_device_count
unnamed_device_count += 1
name = ' '.join(columns[6:]).strip()
return plugged_list
def plug_peripherals(self, on=True):
"""Setting USB mux_3 plug status
@param on: Connect the servo-usb port to DUT(plug) or to servo(unplug)
switch = 'dut_sees_usbkey'
if not on:
switch = 'servo_sees_usbkey', switch)
self.pluged_status = on
def action_login(self):
"""Login i.e. runs running client test
@exception TestFail failed to login within timeout.
def wait_for_cmd_output(self, cmd, check, timeout, timeout_msg):
"""Waits till command output is meta
@param cmd: executed command
@param check: string to be checked for in cmd output
@param timeout: max time in sec to wait for output
@param timeout_msg: timeout failure message
@returns True if check is found in command output; False otherwise
start_time = int(time.time())
time_delta = 0
command = '%s %s' % (cmd, check)
logging.debug('Command: %s', command)
while(, ignore_status=True).exit_status != 0):
time_delta = int(time.time()) - start_time
if time_delta > timeout:
self.add_failure('%s - %d sec' % (timeout_msg, timeout))
return False
logging.debug('Succeeded in :%d sec', time_delta)
return True
def suspend_for_time(self, suspend_time=_SUSPEND_TIME):
"""Calls the host method suspend with suspend_time argument.
@param suspend_time: time to suspend the device for.
except error.AutoservSuspendError:
def action_suspend(self):
"""Suspend i.e. powerd_dbus_suspend and wait
@returns boot_id for the following resume
boot_id =
thread = threading.Thread(target=self.suspend_for_time)
logging.debug('--- Suspended')
self.suspend_status = True
return boot_id
def action_resume(self, boot_id):
"""Resume i.e. press power key and wait
@param boot_id: boot id obtained prior to suspending
logging.debug('--- Resumed')
self.suspend_status = False
def close_lid(self):
"""Close lid through servo to suspend the device."""
boot_id ='Closing lid...')
self.suspend_status = True
return boot_id
def open_lid(self, boot_id):
"""Open lid through servo to resume."""'Opening lid...'), _LONG_TIMEOUT)
self.suspend_status = False
def crash_not_detected(self):
"""Finds new crash files and adds to failures list if any
@returns True if there were not crashes; False otherwise
crash_files = self.detect_crash.get_new_crash_files()
if crash_files:
self.add_failure('CRASH DETECTED: %s' % str(crash_files))
return False
return True
def check_plugged_usb_devices(self):
"""Checks the plugged peripherals match device list.
@returns True if expected USB peripherals are detected; False otherwise
result = True
if self.pluged_status and self.usb_list != None:
# Check for mandatory USb devices passed by usb_list flag
for usb_name in self.usb_list:
found = self.wait_for_cmd_output(
'lsusb | grep -E ', usb_name, _WAIT_DELAY * 4,
'Not detecting %s' % usb_name)
result = result and found
on_now = self.getPluggedUsbDevices()
if self.pluged_status:
if not self.diff_list.issubset(on_now):
missing = str(self.diff_list.difference(on_now))
self.add_failure('Missing connected peripheral(s) '
'when plugged: %s ' % missing)
result = False
present = self.diff_list.intersection(on_now)
if len(present) > 0:
self.add_failure('Still presented peripheral(s) '
'when unplugged: %s ' % str(present))
result = False
return result
def check_usb_peripherals_details(self):
"""Checks the effect from plugged in USB peripherals.
@returns True if command line output is matched successfuly; Else False
usb_check_result = True
for cmd in self.usb_checks.keys():
out_match_list = self.usb_checks.get(cmd)
if cmd.startswith('loggedin:'):
if not self.login_status:
cmd = cmd.replace('loggedin:','')
# Run the usb check command
for out_match in out_match_list:
match_result = self.wait_for_cmd_output(
cmd, out_match, _WAIT_DELAY * 4,
'USB CHECKS DETAILS failed at %s %s:' % (cmd, out_match))
usb_check_result = usb_check_result and match_result
return usb_check_result
def check_status(self):
"""Performs checks after each action:
- for USB detected devices
- for generated crash files
- peripherals effect checks on cmd line
@returns True if all of the iteration checks pass; False otherwise.
result = True
if not self.suspend_status:
# Detect the USB peripherals
result = self.check_plugged_usb_devices()
# Check for crash files
if self.crash_check:
result = result and self.crash_not_detected()
if self.pluged_status and (self.usb_checks != None):
# Check for plugged USB devices details
result = result and self.check_usb_peripherals_details()
return result
def add_failure(self, reason):
""" Adds a failure reason to list of failures to be reported at end
@param reason: failure reason to record
if self.action_step is not None:
self.fail_reasons.append('%s FAILS - %s' %
(self.action_step, reason))
def check_connected_peripherals(self):
""" Verifies there are connected usb devices on servo
@raise error.TestError: if no peripherals are shown
# Collect USB peripherals when unplugged
off_list = self.getPluggedUsbDevices()
# Collect USB peripherals when plugged
time.sleep(_WAIT_DELAY * 2)
on_list = self.getPluggedUsbDevices()
self.diff_list = set(on_list).difference(set(off_list))
if len(self.diff_list) == 0:
# Fail if no devices detected after
raise error.TestError('No connected devices were detected. Make '
'sure the devices are connected to USB_KEY '
'and DUT_HUB1_USB on the servo board.')
logging.debug('Connected devices list: %s', self.diff_list)
def prep_servo_for_test(self, stress_rack):
"""Connects servo to DUT and sets servo ports
@param stress_rack: either to prep servo for stress tests, where
usb_mux_1 port should be on. For usb peripherals on usb_mux_3,
the port is on, and the oe2,oe4 poers are off.
@returns port as string to plug/unplug the specific port
port = _LOWER_USB_PORT'dut')'dut_hub1_rst1','off')
if stress_rack:
port = _UPPER_USB_PORT, 'dut_sees_usbkey')
else:, 'servo_sees_usbkey')'usb_mux_oe2', 'off')'usb_mux_oe4', 'off')
return port
def cleanup(self):
"""Disconnect servo hub"""
def run_once(self, host, client_autotest, action_sequence, repeat,
usb_list=None, usb_checks=None,
crash_check=False, stress_rack=False):
self.client_autotest = client_autotest = host
self.autotest_client = autotest.Autotest(
self.usb_list = usb_list
self.usb_checks = usb_checks
self.crash_check = crash_check
self.suspend_status = False
self.login_status = False
self.fail_reasons = list()
self.action_step = None
self.plug_port = self.prep_servo_for_test(stress_rack)
# Unplug, plug, compare usb peripherals, and leave plugged.
action_sequence = action_sequence.upper()
actions = action_sequence.split(',')
boot_id = 0
self. detect_crash = crash_detector.CrashDetector(
# Run camera client test to gather media_V4L2_test binary.
if 'media_v4l2_test' in str(self.usb_checks):
for iteration in xrange(1, repeat + 1):
step = 0
for action in actions:
step += 1
action = action.strip()
self.action_step = 'STEP %d.%d. %s' % (iteration, step, action)
if action == 'RESUME':
elif action == 'OPENLID':
elif action == 'UNPLUG':
elif action == 'PLUG':
elif self.suspend_status == False:
if action.startswith('LOGIN'):
if self.login_status:
logging.debug('Skipping login. Already logged in.')
self.login_status = True
elif action == 'REBOOT':
time.sleep(_WAIT_DELAY * 3)
self.login_status = False
elif action == 'SUSPEND':
boot_id = self.action_suspend()
elif action == 'CLOSELID':
boot_id = self.close_lid()
else:'WRONG ACTION: %s .', self.action_step)
if self.fail_reasons:
raise error.TestFail('Failures reported: %s' %