You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
221 lines
7.8 KiB
221 lines
7.8 KiB
#!/usr/bin/python2
|
|
# Copyright 2016 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 unittest
|
|
|
|
import common
|
|
from autotest_lib.client.common_lib import error
|
|
from autotest_lib.server.hosts import base_label_unittest, factory
|
|
from autotest_lib.server.hosts import host_info
|
|
|
|
|
|
class MockHost(object):
|
|
"""Mock host object with no side effects."""
|
|
def __init__(self, hostname, **args):
|
|
self._init_args = args
|
|
self._init_args['hostname'] = hostname
|
|
|
|
|
|
def job_start(self):
|
|
"""Only method called by factory."""
|
|
pass
|
|
|
|
|
|
class MockConnectivity(object):
|
|
"""Mock connectivity object with no side effects."""
|
|
def __init__(self, hostname, **args):
|
|
pass
|
|
|
|
def run(self, *args, **kwargs):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
def _gen_mock_host(name, check_host=False):
|
|
"""Create an identifiable mock host closs.
|
|
"""
|
|
return type('mock_host_%s' % name, (MockHost,), {
|
|
'_host_cls_name': name,
|
|
'check_host': staticmethod(lambda host, timeout=None: check_host)
|
|
})
|
|
|
|
|
|
def _gen_mock_conn(name):
|
|
"""Create an identifiable mock connectivity class.
|
|
"""
|
|
return type('mock_conn_%s' % name, (MockConnectivity,),
|
|
{'_conn_cls_name': name})
|
|
|
|
|
|
def _gen_machine_dict(hostname='localhost', labels=[], attributes={}):
|
|
"""Generate a machine dictionary with the specified parameters.
|
|
|
|
@param hostname: hostname of machine
|
|
@param labels: list of host labels
|
|
@param attributes: dict of host attributes
|
|
|
|
@return: machine dict with mocked AFE Host object and fake AfeStore.
|
|
"""
|
|
afe_host = base_label_unittest.MockAFEHost(labels, attributes)
|
|
store = host_info.InMemoryHostInfoStore()
|
|
store.commit(host_info.HostInfo(labels, attributes))
|
|
return {'hostname': hostname,
|
|
'afe_host': afe_host,
|
|
'host_info_store': store}
|
|
|
|
|
|
class CreateHostUnittests(unittest.TestCase):
|
|
"""Tests for create_host function."""
|
|
|
|
def setUp(self):
|
|
"""Prevent use of real Host and connectivity objects due to potential
|
|
side effects.
|
|
"""
|
|
self._orig_types = factory.host_types
|
|
self._orig_dict = factory.OS_HOST_DICT
|
|
self._orig_cros_host = factory.cros_host.CrosHost
|
|
self._orig_local_host = factory.local_host.LocalHost
|
|
self._orig_ssh_host = factory.ssh_host.SSHHost
|
|
|
|
self.host_types = factory.host_types = []
|
|
self.os_host_dict = factory.OS_HOST_DICT = {}
|
|
factory.cros_host.CrosHost = _gen_mock_host('cros_host')
|
|
factory.local_host.LocalHost = _gen_mock_conn('local')
|
|
factory.ssh_host.SSHHost = _gen_mock_conn('ssh')
|
|
|
|
|
|
def tearDown(self):
|
|
"""Clean up mocks."""
|
|
factory.host_types = self._orig_types
|
|
factory.OS_HOST_DICT = self._orig_dict
|
|
factory.cros_host.CrosHost = self._orig_cros_host
|
|
factory.local_host.LocalHost = self._orig_local_host
|
|
factory.ssh_host.SSHHost = self._orig_ssh_host
|
|
|
|
|
|
def test_use_specified(self):
|
|
"""Confirm that the specified host class is used."""
|
|
machine = _gen_machine_dict()
|
|
host_obj = factory.create_host(
|
|
machine,
|
|
_gen_mock_host('specified'),
|
|
)
|
|
self.assertEqual(host_obj._host_cls_name, 'specified')
|
|
|
|
|
|
def test_detect_host_by_os_label(self):
|
|
"""Confirm that the host object is selected by the os label.
|
|
"""
|
|
machine = _gen_machine_dict(labels=['os:foo'])
|
|
self.os_host_dict['foo'] = _gen_mock_host('foo')
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._host_cls_name, 'foo')
|
|
|
|
|
|
def test_detect_host_by_os_type_attribute(self):
|
|
"""Confirm that the host object is selected by the os_type attribute
|
|
and that the os_type attribute is preferred over the os label.
|
|
"""
|
|
machine = _gen_machine_dict(labels=['os:foo'],
|
|
attributes={'os_type': 'bar'})
|
|
self.os_host_dict['foo'] = _gen_mock_host('foo')
|
|
self.os_host_dict['bar'] = _gen_mock_host('bar')
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._host_cls_name, 'bar')
|
|
|
|
|
|
def test_detect_host_by_check_host(self):
|
|
"""Confirm check_host logic chooses a host object when label/attribute
|
|
detection fails.
|
|
"""
|
|
machine = _gen_machine_dict()
|
|
self.host_types.append(_gen_mock_host('first', check_host=False))
|
|
self.host_types.append(_gen_mock_host('second', check_host=True))
|
|
self.host_types.append(_gen_mock_host('third', check_host=False))
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._host_cls_name, 'second')
|
|
|
|
|
|
def test_detect_host_fallback_to_cros_host(self):
|
|
"""Confirm fallback to CrosHost when all other detection fails.
|
|
"""
|
|
machine = _gen_machine_dict()
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._host_cls_name, 'cros_host')
|
|
|
|
|
|
def test_choose_connectivity_local(self):
|
|
"""Confirm local connectivity class used when hostname is localhost.
|
|
"""
|
|
machine = _gen_machine_dict(hostname='localhost')
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._conn_cls_name, 'local')
|
|
|
|
|
|
def test_choose_connectivity_ssh(self):
|
|
"""Confirm ssh connectivity class used when configured and hostname
|
|
is not localhost.
|
|
"""
|
|
machine = _gen_machine_dict(hostname='somehost')
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._conn_cls_name, 'ssh')
|
|
|
|
|
|
def test_argument_passthrough(self):
|
|
"""Confirm that detected and specified arguments are passed through to
|
|
the host object.
|
|
"""
|
|
machine = _gen_machine_dict(hostname='localhost')
|
|
host_obj = factory.create_host(machine, foo='bar')
|
|
self.assertEqual(host_obj._init_args['hostname'], 'localhost')
|
|
self.assertTrue('afe_host' in host_obj._init_args)
|
|
self.assertTrue('host_info_store' in host_obj._init_args)
|
|
self.assertEqual(host_obj._init_args['foo'], 'bar')
|
|
|
|
|
|
def test_global_ssh_params(self):
|
|
"""Confirm passing of ssh parameters set as globals.
|
|
"""
|
|
factory.ssh_user = 'foo'
|
|
factory.ssh_pass = 'bar'
|
|
factory.ssh_port = 1
|
|
factory.ssh_verbosity_flag = 'baz'
|
|
factory.ssh_options = 'zip'
|
|
machine = _gen_machine_dict()
|
|
try:
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._init_args['user'], 'foo')
|
|
self.assertEqual(host_obj._init_args['password'], 'bar')
|
|
self.assertEqual(host_obj._init_args['port'], 1)
|
|
self.assertEqual(host_obj._init_args['ssh_verbosity_flag'], 'baz')
|
|
self.assertEqual(host_obj._init_args['ssh_options'], 'zip')
|
|
finally:
|
|
del factory.ssh_user
|
|
del factory.ssh_pass
|
|
del factory.ssh_port
|
|
del factory.ssh_verbosity_flag
|
|
del factory.ssh_options
|
|
|
|
|
|
def test_host_attribute_ssh_params(self):
|
|
"""Confirm passing of ssh parameters from host attributes.
|
|
"""
|
|
machine = _gen_machine_dict(attributes={'ssh_user': 'somebody',
|
|
'ssh_port': 100,
|
|
'ssh_verbosity_flag': 'verb',
|
|
'ssh_options': 'options'})
|
|
host_obj = factory.create_host(machine)
|
|
self.assertEqual(host_obj._init_args['user'], 'somebody')
|
|
self.assertEqual(host_obj._init_args['port'], 100)
|
|
self.assertEqual(host_obj._init_args['ssh_verbosity_flag'], 'verb')
|
|
self.assertEqual(host_obj._init_args['ssh_options'], 'options')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|