client/common_lib/cros/tendo/buffet_config.py

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