blob: 5d7d0cd4162303cdf6aba03981bc82bbd694bc70 [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 module for running automated factory tests on a DUT."""
import glob
import httplib
import jsonrpclib
import logging
import os
import socket
import subprocess
import threading
import time
import yaml
import factory_common # pylint: disable=W0611
from cros.factory.factory_flow.common import (
board_cmd_arg, bundle_dir_cmd_arg, dut_hostname_cmd_arg, FactoryFlowCommand)
from cros.factory.goofy import connection_manager
from cros.factory.goofy.goofy_rpc import RunState
from cros.factory.goofy.invocation import OVERRIDE_TEST_LIST_DARGS_FILE
from cros.factory.hacked_argparse import CmdArg
from cros.factory.test import state
from cros.factory.test import utils
from cros.factory.test.e2e_test.automator import AUTOMATION_FUNCTION_KWARGS_FILE
from cros.factory.utils import file_utils
from cros.factory.utils import net_utils
from cros.factory.utils import process_utils
from cros.factory.utils import ssh_utils
TEST_AUTOMATION_MESSAGE = """
*** Start factory test automation on DUT %(dut)s:
*** - Automation mode: %(automation_mode)s
*** - Test list: %(test_list)s
*** - Clear state: %(clear_states)s
"""
class RunAutomatedTestsError(Exception):
"""Run automated test error."""
pass
class RunAutomatedTests(FactoryFlowCommand):
"""Runs automated factory tests on DUT."""
args = [
board_cmd_arg,
bundle_dir_cmd_arg,
dut_hostname_cmd_arg,
CmdArg('--automation-mode', choices=['none', 'partial', 'full'],
default='partial', help=(
'mode of factory test automation to enable '
'(default: %(default)s)')),
CmdArg('--test-list', default=None, help='test list to run'),
CmdArg('--shopfloor-ip', help='IP of shop floor server'),
CmdArg('--shopfloor-port', type=int, help='port of shop floor server'),
CmdArg('--no-clear-states', dest='clear_states', action='store_false',
default=True, help='do not clear factory test states and VPD'),
CmdArg('--no-wait', dest='wait', action='store_false',
help='do not wait for automated factory tests to complete'),
CmdArg('--run', metavar='TEST_PATH',
help='run only the specified factory test'),
CmdArg('--timeout-mins', type=float, default=60,
help='minutes to wait before marking the test run as failed'),
CmdArg('--log-dir',
help=('path to the directory to store factory logs from DUT; '
'defaults to <output_dir>/logs/factory_logs')),
CmdArg('--device-data-yaml',
help=('a YAML file containing a dict of device data '
'key-value pairs to set on the DUT')),
CmdArg('--vpd-yaml',
help=('a YAML file containing a dict of RO and RW VPD '
'key-value pairs to set on the DUT')),
CmdArg('--test-list-dargs-yaml',
help=('a YAML file containing a dict of test paths to '
'their override dargs')),
CmdArg('--automation-function-kwargs-yaml',
help=('a YAML file containing a dict of test paths to '
'their kwargs for their automation functions')),
]
WAIT_FOR_GOOFY_TIMEOUT_SECS = 120
GOOFY_POLLING_INTERVAL = 5
ssh_tunnel = None
def Init(self):
if not self.options.log_dir:
self.options.log_dir = os.path.join(self.options.bundle, os.path.pardir,
'logs', 'factory_logs')
file_utils.TryMakeDirs(self.options.log_dir)
def Run(self):
self.EnableFactoryTestAutomation()
self.WaitForGoofy()
self.PrepareAutomationEnvironment()
self.StartAutomatedTests()
if self.options.wait:
self.WaitForTestsToFinish()
def TearDown(self):
if self.ssh_tunnel:
self.ssh_tunnel.Close()
def PrepareAutomationEnvironment(self):
"""Prepares settings for factory test automation on the DUT."""
def _CheckFile(path):
if not os.path.exists(path):
raise RunAutomatedTestsError('Unable to locate file %r' % path)
if self.options.device_data_yaml:
# Rsyncs the device data YAML file to the DUT and loads it.
_CheckFile(self.options.device_data_yaml)
TEMP_DEVICE_DATA_PATH = '/tmp/device_data'
DUT_FACTORY_BIN = '/usr/local/factory/bin/factory'
ssh_utils.SpawnRsyncToDUT(
[self.options.device_data_yaml,
'%s:%s' % (self.options.dut, TEMP_DEVICE_DATA_PATH)],
log=True, check_call=True)
ssh_utils.SpawnSSHToDUT(
[self.options.dut, DUT_FACTORY_BIN, 'device-data', '--set-yaml',
TEMP_DEVICE_DATA_PATH], log=True, check_call=True)
if self.options.vpd_yaml:
_CheckFile(self.options.vpd_yaml)
with open(self.options.vpd_yaml) as f:
vpd_yaml_dict = yaml.safe_load(f.read())
def GenerateVPDSetterArgs(vpd_dict):
result = []
for key, value in vpd_dict.iteritems():
result.extend(['-s', '"%s"="%s"' % (key, value)])
return result
vpd_section_args = {
'ro': 'RO_VPD',
'rw': 'RW_VPD',
}
for section in ('ro', 'rw'):
vpd_setters = GenerateVPDSetterArgs(vpd_yaml_dict[section])
ssh_utils.SpawnSSHToDUT(
[self.options.dut, 'vpd', '-i', vpd_section_args[section]] +
vpd_setters,
log=True, check_call=True)
if self.options.test_list_dargs_yaml:
# Rsyncs the test list dargs YAML file to the DUT.
_CheckFile(self.options.test_list_dargs_yaml)
TEMP_DARGS_PATH = '/tmp/test_list_dargs'
DUT_DARGS_PATH = os.path.join(
'/var/factory/state/',
os.path.basename(OVERRIDE_TEST_LIST_DARGS_FILE))
ssh_utils.SpawnRsyncToDUT(
[self.options.test_list_dargs_yaml,
'%s:%s' % (self.options.dut, TEMP_DARGS_PATH)],
log=True, check_call=True)
ssh_utils.SpawnSSHToDUT(
[self.options.dut, 'mv', TEMP_DARGS_PATH, DUT_DARGS_PATH],
log=True, check_call=True)
if self.options.automation_function_kwargs_yaml:
# Rsyncs the automation function kwargs YAML file to the DUT.
_CheckFile(self.options.automation_function_kwargs_yaml)
TEMP_KWARGS_PATH = '/tmp/automation_function_kwargs'
DUT_KWARGS_PATH = os.path.join(
'/var/factory/state/',
os.path.basename(AUTOMATION_FUNCTION_KWARGS_FILE))
ssh_utils.SpawnRsyncToDUT(
[self.options.automation_function_kwargs_yaml,
'%s:%s' % (self.options.dut, TEMP_KWARGS_PATH)],
log=True, check_call=True)
ssh_utils.SpawnSSHToDUT(
[self.options.dut, 'mv', TEMP_KWARGS_PATH, DUT_KWARGS_PATH],
log=True, check_call=True)
def EnableFactoryTestAutomation(self):
"""Enables factory test automation on the DUT."""
GOOFY_REMOTE = '/usr/local/factory/bin/goofy_remote'
FACTORY_RESTART = '/usr/local/factory/bin/factory_restart'
logging.info('Runing automated tests on %s', self.options.dut)
goofy_remote_args = [self.options.dut, GOOFY_REMOTE, self.options.dut,
'--local']
if self.options.test_list:
goofy_remote_args += ['--test_list=%s' % self.options.test_list]
if self.options.shopfloor_ip:
goofy_remote_args += ['-s', self.options.shopfloor_ip]
if self.options.shopfloor_port:
goofy_remote_args += ['--shopfloor_port=%s', self.options.shopfloor_port]
ssh_utils.SpawnSSHToDUT(goofy_remote_args, log=True, check_call=True)
ssh_utils.SpawnSSHToDUT(
[self.options.dut, FACTORY_RESTART, '--automation-mode',
'%s' % self.options.automation_mode, '--no-auto-run-on-start'] +
(['-a', '-d'] if self.options.clear_states else []),
log=True, check_call=True)
def GetGoofyProxy(self):
"""Creates a Goofy RPC proxy.
A SSH tunnel to DUT is created each time this function is called. This is to
ensure that the tunnel is alive, since DUT can go through reboot test and
make established tunnel unreachable. If a SSH tunnel has been established
then the tunnel will be closed before a new one is established.
Returns:
A Goofy RPC proxy instance, or None if the method fails to create a proxy.
"""
try:
if self.ssh_tunnel:
self.ssh_tunnel.Close()
# Ping the host first to make sure it is up.
utils.WaitFor(
lambda: connection_manager.PingHost(self.options.dut, timeout=1) == 0,
timeout_secs=30)
# Create a SSH tunnel to connect to the JSON RPC server on DUT.
local_port = net_utils.GetUnusedPort()
self.ssh_tunnel = ssh_utils.SSHTunnelToDUT(
self.options.dut, local_port, state.DEFAULT_FACTORY_STATE_PORT)
self.ssh_tunnel.Establish()
goofy_proxy = state.get_instance(address='localhost', port=local_port)
goofy_proxy.GetGoofyStatus() # Make sure the proxy works.
return goofy_proxy
except (utils.TimeoutError, # Cannot ping DUT.
subprocess.CalledProcessError, # Cannot create SSH tunnel.
socket.error, # Cannot connect to Goofy on DUT.
httplib.BadStatusLine): # Goofy RPC on DUT is not ready.
return None
def WaitForGoofy(self):
"""Waits for Goofy to come up."""
def PollGoofy():
goofy_proxy = self.GetGoofyProxy()
if goofy_proxy is None:
return False
return goofy_proxy.GetGoofyStatus()['status'] == 'RUNNING'
logging.info('Waiting for Goofy to come up')
utils.WaitFor(PollGoofy, timeout_secs=self.WAIT_FOR_GOOFY_TIMEOUT_SECS,
poll_interval=self.GOOFY_POLLING_INTERVAL)
def StartAutomatedTests(self):
"""Starts automated factory tests."""
goofy_proxy = self.GetGoofyProxy()
if goofy_proxy is None:
raise RunAutomatedTestsError('Unable to connect to Goofy on DUT')
if self.options.run:
goofy_proxy.RunTest(self.options.run)
else:
goofy_proxy.RestartAllTests()
def WaitForTestsToFinish(self):
"""Waits for the automated test run to finish."""
finished_tests = []
goofy_proxy = self.GetGoofyProxy()
if goofy_proxy is None:
raise RunAutomatedTestsError('Unable to connect to Goofy on DUT')
test_list = goofy_proxy.GetGoofyStatus()['test_list_id']
stop_event = threading.Event()
def KeepSyncingFactoryLogs(stop_event):
"""Keeps syncing factory logs from the DUT to the host."""
per_test_list_log_dir = os.path.join(self.options.log_dir, test_list)
file_utils.TryMakeDirs(per_test_list_log_dir)
while not stop_event.is_set():
ssh_utils.SpawnRsyncToDUT(
['-aP', '%s:/var/factory' % self.options.dut,
per_test_list_log_dir],
stdout=process_utils.OpenDevNull(),
stderr=process_utils.OpenDevNull(),
call=True)
time.sleep(1)
# Start a thread to keep syncing factory logs from the DUT.
sync_log_thread = threading.Thread(target=KeepSyncingFactoryLogs,
args=(stop_event,))
sync_log_thread.start()
print TEST_AUTOMATION_MESSAGE % dict(
dut=self.options.dut,
automation_mode=self.options.automation_mode,
test_list=test_list,
clear_states=self.options.clear_states)
def ParseFinishedTests(run_status):
"""Parses finish tests."""
finished = [
(t['path'], t['status']) for t in run_status['scheduled_tests']
if t['status'] in ('PASSED', 'FAILED')]
for t in finished:
if t not in finished_tests:
logging.info('[%s] %s', t[1], t[0])
finished_tests.append(t)
def WaitForRun():
"""Waits for test run to finish."""
goofy_proxy = self.GetGoofyProxy()
if goofy_proxy is None:
# Cannot connect to DUT; probably a reboot/suspend test is running,
# or DUT is finalized.
finalize_report_spec = glob.glob(
os.path.join(self.options.bundle, 'shopfloor', 'shopfloor_data',
'reports', time.strftime('logs.%Y%m%d'), '*.tar.xz'))
if finalize_report_spec:
logging.info(('Found finalize report %s; assuming DUT has been '
'finalized'), finalize_report_spec[0])
return True
return False
try:
# Fetch run status on the DUT through Goofy RPC.
run_status = goofy_proxy.GetTestRunStatus(None)
except (jsonrpclib.jsonrpc.ProtocolError, socket.error,
httplib.BadStatusLine):
# Time out waiting for response from Goofy RPC, or the SSH connection is
# gone.
return False
ParseFinishedTests(run_status)
return run_status['status'] in (RunState.FINISHED,
RunState.NOT_ACTIVE_RUN)
try:
logging.info('Waiting for automated test run to finish')
utils.WaitFor(WaitForRun, timeout_secs=self.options.timeout_mins * 60,
poll_interval=self.GOOFY_POLLING_INTERVAL)
if not finished_tests:
raise RunAutomatedTestsError('No test was run')
failed_tests = [t[0] for t in finished_tests if t[1] == 'FAILED']
if failed_tests:
raise RunAutomatedTestsError(
'The following test(s) failed:' + '\n'.join(failed_tests))
finally:
stop_event.set()
sync_log_thread.join()