xref: /aosp_15_r20/external/autotest/server/hosts/factory_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3
# 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()