blob: f037bdca5ec4e20644cf2f648a4df8aaa61cf9e9 [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.
"""Enviroment objects that handles external test operations."""
from __future__ import print_function
import cPickle as pickle
import hashlib
import logging
import os
import subprocess
import threading
import time
import factory_common # pylint: disable=W0611
from cros.factory.goofy_split import connection_manager
from cros.factory.system.service_manager import GetServiceStatus
from cros.factory.system.service_manager import SetServiceStatus
from cros.factory.system.service_manager import Status
from cros.factory.test import state
from cros.factory.test import utils
from import chrome_debugger
class Environment(object):
"""Abstract base class for external test operations, e.g., run an autotest,
shutdown, or reboot.
The Environment is assumed not to be thread-safe: callers must grab the lock
before calling any methods. This is primarily necessary because we mock out
this Environment with mox, and unfortunately mox is not thread-safe.
TODO(jsalz): Try to write a thread-safe wrapper for mox.
lock = threading.Lock()
def shutdown(self, operation):
"""Shuts the machine down (from a ShutdownStep).
operation: 'reboot', 'full_reboot', or 'halt'.
True if Goofy should gracefully exit, or False if Goofy
should just consider the shutdown to have suceeded (e.g.,
in the chroot).
raise NotImplementedError()
def controller_ready_for_ui(self):
"""Hooks called when Goofy controller is ready for UI connection."""
def launch_chrome(self):
"""Launches Chrome.
The Chrome subprocess (or None if none).
raise NotImplementedError()
def spawn_autotest(self, name, args, env_additions, result_file):
"""Spawns a process to run an autotest.
name: Name of the autotest to spawn.
args: Command-line arguments.
env_additions: Additions to the environment.
result_file: Expected location of the result file.
raise NotImplementedError()
def create_connection_manager(self, wlans, scan_wifi_period_secs):
"""Creates a ConnectionManager."""
raise NotImplementedError()
def terminate(self):
"""Terminates and cleans up environment."""
class DUTEnvironment(Environment):
"""A real environment on a device under test."""
def __init__(self):
super(DUTEnvironment, self).__init__()
self.goofy = None # Must be assigned later by goofy.
self.has_sockets = None # Must be assigned later by goofy.
def shutdown(self, operation):
def prepare_shutdown():
"""Prepares for a clean shutdown."""
respawn_services = ['syslog',
for service in respawn_services:
if GetServiceStatus(service) == Status.START:
SetServiceStatus(service, Status.STOP)
assert operation in ['reboot', 'full_reboot', 'halt']'Shutting down: %s', operation)
if operation == 'full_reboot':
subprocess.check_call(['ectool', 'reboot_ec', 'cold', 'at-shutdown'])
subprocess.check_call(['shutdown', '-h', 'now'])
commands = dict(reboot=['shutdown', '-r', 'now'],
halt=['shutdown', '-h', 'now'])
# TODO(hungte) Current implementation will raise SIGTERM so goofy can't
# really gracefully shutdown. We should do "on exit" instead.
assert False, 'Never reached (should %s)' % operation
def spawn_autotest(self, name, args, env_additions, result_file):
return self.goofy.autotest_prespawner.spawn(args, env_additions)
def override_chrome_start_pages(self):
# TODO(hungte) Remove this workaround (mainly for
override_chrome_start_file = '/usr/local/factory/init/override_chrome_start'
if not os.path.exists(override_chrome_start_file):
url = (open(override_chrome_start_file).read() or
('http://%s:%s' % (state.DEFAULT_FACTORY_STATE_ADDRESS,
(host, unused_colon, port) = url.partition('http://')[2].partition(':')'Override chrome start pages as: %s', url)
chrome = chrome_debugger.ChromeRemoteDebugger()
utils.WaitFor(chrome.IsReady, 30)
# Wait for state server to be ready.
state_server = state.get_instance(address=host, port=int(port))
def is_state_server_ready():
return state_server.IsReadyForUIConnection()
return False
utils.WaitFor(is_state_server_ready, 30)
def launch_chrome(self):
utils.WaitFor(self.has_sockets, 30)
subprocess.check_call(['initctl', 'emit', 'login-prompt-visible'])
# TODO: Find a way to disable two-finger scrolling when not using X
if utils.HasX():
# Disable X-axis two-finger scrolling on touchpad.
# If we have a touchpad, then disable touchscreen so that operators won't
# accidentally roll back to previous webpage. If we have no touchpad, we
# assume that the touch UI is required.
if utils.GetTouchpadDeviceIds():
for device_id in utils.GetTouchscreenDeviceIds():
utils.SetXinputDeviceEnabled(device_id, False)
def create_connection_manager(self, wlans, scan_wifi_period_secs):
return connection_manager.ConnectionManager(wlans,
class FakeChrootEnvironment(Environment):
"""A chroot environment that doesn't actually shutdown or run autotests."""
def shutdown(self, operation):
assert operation in ['reboot', 'full_reboot', 'halt']
logging.warn('In chroot: skipping %s', operation)
return False
def spawn_autotest(self, name, args, env_additions, result_file):
logging.warn('In chroot: skipping autotest %s', name)
# Mark it as passed with 75% probability, or failed with 25%
# probability (depending on a hash of the autotest name).
pseudo_random = ord(hashlib.sha1(name).digest()[0]) / 256.0
passed = pseudo_random > .25
with open(result_file, 'w') as out:
pickle.dump((passed, '' if passed else 'Simulated failure'), out)
# Start a process that will return with a true exit status in
# 2 seconds (just like a happy autotest).
return subprocess.Popen(['sleep', '2'])
def launch_chrome(self):
logging.warn('In chroot; not launching Chrome. '
'Please open UI presenter app in Chrome.')
def create_connection_manager(self, wlans, scan_wifi_period_secs):
return connection_manager.DummyConnectionManager()