140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
# Copyright 2015 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
|
|
import random
|
|
import string
|
|
|
|
from autotest_lib.client.common_lib import utils
|
|
from autotest_lib.client.common_lib.cros.fake_device_server import oauth
|
|
from autotest_lib.client.common_lib.cros.fake_device_server import server
|
|
|
|
TEST_CONFIG_PATH = '/tmp/buffet.fake.conf'
|
|
TEST_STATE_PATH = '/tmp/buffet.fake.state'
|
|
|
|
LOCAL_SERVER_PORT = server.PORT
|
|
LOCAL_OAUTH_URL = 'http://localhost:%d/%s/' % (LOCAL_SERVER_PORT,
|
|
oauth.OAUTH_PATH)
|
|
LOCAL_SERVICE_URL = 'http://localhost:%d/' % LOCAL_SERVER_PORT
|
|
TEST_API_KEY = oauth.TEST_API_KEY
|
|
|
|
def build_unique_device_name():
|
|
"""@return a test-unique name for a device."""
|
|
RAND_CHARS = string.ascii_lowercase + string.digits
|
|
NUM_RAND_CHARS = 16
|
|
rand_token = ''.join([random.choice(RAND_CHARS)
|
|
for _ in range(NUM_RAND_CHARS)])
|
|
name = 'CrOS_%s' % rand_token
|
|
logging.debug('Generated unique device name %s', name)
|
|
return name
|
|
|
|
|
|
TEST_CONFIG = {
|
|
'client_id': 'this_is_my_client_id',
|
|
'client_secret': 'this_is_my_client_secret',
|
|
'api_key': TEST_API_KEY,
|
|
'oauth_url': LOCAL_OAUTH_URL,
|
|
'service_url': LOCAL_SERVICE_URL,
|
|
'model_id': 'AATST',
|
|
'wifi_auto_setup_enabled': 'false',
|
|
'name': build_unique_device_name()
|
|
}
|
|
|
|
|
|
def bool_to_flag(value):
|
|
"""Converts boolean value into lowercase string
|
|
|
|
@param value: Boolean value.
|
|
@return lower case string: 'true' or 'false'.
|
|
|
|
"""
|
|
return ('%s' % value).lower()
|
|
|
|
|
|
def format_options(options, separator):
|
|
"""Format dictionary as key1=value1{separator}key2=value2{separator}..
|
|
|
|
@param options: Dictionary with options.
|
|
@param separator: String to be used as separator between key=value strings.
|
|
@return formated string.
|
|
|
|
"""
|
|
return separator.join(['%s=%s' % (k, v) for (k, v) in options.iteritems()])
|
|
|
|
|
|
def naive_restart(host=None):
|
|
"""Restart Buffet without configuring it in any way.
|
|
|
|
@param host: Host object if we're interested in a remote host.
|
|
|
|
"""
|
|
run = utils.run if host is None else host.run
|
|
run('stop buffet', ignore_status=True)
|
|
run('start buffet')
|
|
|
|
|
|
|
|
class BuffetConfig(object):
|
|
"""An object that knows how to restart buffet in various configurations."""
|
|
|
|
def __init__(self,
|
|
log_verbosity=None,
|
|
test_definitions_dir=None,
|
|
enable_xmpp=False,
|
|
enable_ping=True,
|
|
disable_pairing_security=False,
|
|
device_whitelist=None,
|
|
options=None):
|
|
self.enable_xmpp = enable_xmpp
|
|
self.log_verbosity = log_verbosity
|
|
self.test_definitions_dir = test_definitions_dir
|
|
self.enable_ping = enable_ping
|
|
self.disable_pairing_security = disable_pairing_security
|
|
self.device_whitelist = device_whitelist
|
|
self.options = TEST_CONFIG.copy()
|
|
if options:
|
|
self.options.update(options)
|
|
|
|
|
|
def restart_with_config(self,
|
|
host=None,
|
|
clean_state=True):
|
|
"""Restart Buffet with this configuration.
|
|
|
|
@param host: Host object if we're interested in a remote host.
|
|
@param clean_state: boolean True to remove all existing state.
|
|
|
|
"""
|
|
run = utils.run if host is None else host.run
|
|
run('stop buffet', ignore_status=True)
|
|
flags = {
|
|
'BUFFET_ENABLE_XMPP': 'true' if self.enable_xmpp else 'false',
|
|
'BUFFET_CONFIG_PATH': TEST_CONFIG_PATH,
|
|
'BUFFET_STATE_PATH': TEST_STATE_PATH,
|
|
'BUFFET_ENABLE_PING': bool_to_flag(self.enable_ping),
|
|
'BUFFET_DISABLE_SECURITY':
|
|
bool_to_flag(self.disable_pairing_security),
|
|
}
|
|
if self.log_verbosity:
|
|
flags['BUFFET_LOG_LEVEL'] = self.log_verbosity
|
|
|
|
# Go through this convoluted shell magic here because we need to
|
|
# create this file on both remote and local hosts (see how run() is
|
|
# defined).
|
|
run('cat <<EOF >%s\n%s\nEOF\n' % (TEST_CONFIG_PATH,
|
|
format_options(self.options, '\n')))
|
|
|
|
if clean_state:
|
|
run('echo > %s' % TEST_STATE_PATH)
|
|
run('chown buffet:buffet %s' % TEST_STATE_PATH)
|
|
|
|
if self.test_definitions_dir:
|
|
flags['BUFFET_TEST_DEFINITIONS_PATH'] = self.test_definitions_dir
|
|
|
|
if self.device_whitelist:
|
|
flags['BUFFET_DEVICE_WHITELIST'] = ','.join(self.device_whitelist)
|
|
|
|
run('start buffet %s' % format_options(flags, ' '))
|
|
|