# 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, ' '))