salt/tests/integration/cloud/clouds/test_virtualbox.py

# -*- coding: utf-8 -*-
# This code assumes vboxapi.py from VirtualBox distribution
# being in PYTHONPATH, or installed system-wide

# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals

import logging
import os
import socket

from salt.config import cloud_providers_config, vm_profiles_config

# Import Salt Libs
from salt.ext import six
from salt.ext.six.moves import range
from salt.utils.virtualbox import (
    HAS_LIBS,
    XPCOM_ATTRIBUTES,
    machine_get_machinestate_str,
    vb_clone_vm,
    vb_create_machine,
    vb_destroy_machine,
    vb_get_box,
    vb_get_network_addresses,
    vb_machine_exists,
    vb_start_vm,
    vb_stop_vm,
    vb_wait_for_network_address,
    vb_xpcom_to_attribute_dict,
)
from tests.integration.cloud.helpers.virtualbox import (
    BASE_BOX_NAME,
    BOOTABLE_BASE_BOX_NAME,
    CONFIG_NAME,
    DEPLOY_PROFILE_NAME,
    INSTANCE_NAME,
    PROFILE_NAME,
    PROVIDER_NAME,
    VirtualboxCloudTestCase,
    VirtualboxTestCase,
)
from tests.support.runtests import RUNTIME_VARS

# Import Salt Testing Libs
from tests.support.unit import TestCase, skipIf

log = logging.getLogger(__name__)

# As described in the documentation of list_nodes (this may change with time)
MINIMAL_MACHINE_ATTRIBUTES = [
    "id",
    "image",
    "size",
    "state",
    "private_ips",
    "public_ips",
]


class VirtualboxProviderTest(VirtualboxCloudTestCase):
    """
    Integration tests for the Virtualbox cloud provider using the Virtualbox driver
    """

    def run_cloud_destroy(self, machine_name):
        """
        Calls salt-cloud to destroy a machine and returns the destroyed machine object (should be None)
        @param machine_name:
        @type str:
        @return:
        @rtype: dict
        """
        output = self.run_cloud(
            "-d {0} --assume-yes --log-level=debug".format(machine_name)
        )
        return output.get(CONFIG_NAME, {}).get(PROVIDER_NAME, {})

    def setUp(self):
        """
        Sets up the test requirements
        """
        super(VirtualboxProviderTest, self).setUp()

        # check if appropriate cloud provider and profile files are present
        profile_str = "virtualbox-config"
        providers = self.run_cloud("--list-providers")
        log.debug("providers: %s", providers)

        if profile_str not in providers:
            self.skipTest(
                "Configuration file for {0} was not found. Check {0}.conf files "
                "in tests/integration/files/conf/cloud.*.d/ to run these tests.".format(
                    PROVIDER_NAME
                )
            )

        # check if personal access token, ssh_key_file, and ssh_key_names are present
        config_path = os.path.join(
            RUNTIME_VARS.FILES, "conf", "cloud.providers.d", PROVIDER_NAME + ".conf"
        )
        log.debug("config_path: %s", config_path)
        providers = cloud_providers_config(config_path)
        log.debug("config: %s", providers)
        config_path = os.path.join(
            RUNTIME_VARS.FILES, "conf", "cloud.profiles.d", PROVIDER_NAME + ".conf"
        )
        profiles = vm_profiles_config(config_path, providers)
        profile = profiles.get(PROFILE_NAME)
        if not profile:
            self.skipTest(
                "Profile {0} was not found. Check {1}.conf files "
                "in tests/integration/files/conf/cloud.profiles.d/ to run these tests.".format(
                    PROFILE_NAME, PROVIDER_NAME
                )
            )
        base_box_name = profile.get("clonefrom")

        if base_box_name != BASE_BOX_NAME:
            self.skipTest(
                "Profile {0} does not have a base box to clone from. Check {1}.conf files "
                "in tests/integration/files/conf/cloud.profiles.d/ to run these tests."
                'And add a "clone_from: {2}" to the profile'.format(
                    PROFILE_NAME, PROVIDER_NAME, BASE_BOX_NAME
                )
            )

    @classmethod
    def setUpClass(cls):
        vb_create_machine(BASE_BOX_NAME)

    @classmethod
    def tearDownClass(cls):
        vb_destroy_machine(BASE_BOX_NAME)

    def test_cloud_create(self):
        """
        Simply create a machine and make sure it was created
        """
        machines = self.run_cloud(
            "-p {0} {1} --log-level=debug".format(PROFILE_NAME, INSTANCE_NAME)
        )
        self.assertIn(INSTANCE_NAME, machines.keys())

    def test_cloud_list(self):
        """
        List all machines in virtualbox and make sure the requested attributes are included
        """
        machines = self.run_cloud_function("list_nodes")

        expected_attributes = MINIMAL_MACHINE_ATTRIBUTES
        names = machines.keys()
        self.assertGreaterEqual(len(names), 1, "No machines found")
        for name, machine in six.iteritems(machines):
            if six.PY3:
                self.assertCountEqual(expected_attributes, machine.keys())
            else:
                self.assertItemsEqual(expected_attributes, machine.keys())

        self.assertIn(BASE_BOX_NAME, names)

    def test_cloud_list_full(self):
        """
        List all machines and make sure full information in included
        """
        machines = self.run_cloud_function("list_nodes_full")
        expected_minimal_attribute_count = len(MINIMAL_MACHINE_ATTRIBUTES)

        names = machines.keys()
        self.assertGreaterEqual(len(names), 1, "No machines found")
        for name, machine in six.iteritems(machines):
            self.assertGreaterEqual(
                len(machine.keys()), expected_minimal_attribute_count
            )

        self.assertIn(BASE_BOX_NAME, names)

    def test_cloud_list_select(self):
        """
        List selected attributes of all machines
        """
        machines = self.run_cloud_function("list_nodes_select")
        # TODO find out how to get query.selection from the  "cloud" config
        expected_attributes = ["id"]

        names = machines.keys()
        self.assertGreaterEqual(len(names), 1, "No machines found")
        for name, machine in six.iteritems(machines):
            if six.PY3:
                self.assertCountEqual(expected_attributes, machine.keys())
            else:
                self.assertItemsEqual(expected_attributes, machine.keys())

        self.assertIn(BASE_BOX_NAME, names)

    def test_cloud_destroy(self):
        """
        Test creating an instance on virtualbox with the virtualbox driver
        """
        # check if instance with salt installed returned
        self.test_cloud_create()
        ret = self.run_cloud_destroy(INSTANCE_NAME)

        # destroy the instance
        self.assertIn(INSTANCE_NAME, ret.keys())

    def test_function_show_instance(self):
        kw_function_args = {"image": BASE_BOX_NAME}
        machines = self.run_cloud_function("show_image", kw_function_args, timeout=30)
        expected_minimal_attribute_count = len(MINIMAL_MACHINE_ATTRIBUTES)
        self.assertIn(BASE_BOX_NAME, machines)
        machine = machines[BASE_BOX_NAME]
        self.assertGreaterEqual(len(machine.keys()), expected_minimal_attribute_count)

    def tearDown(self):
        """
        Clean up after tests
        """
        if vb_machine_exists(INSTANCE_NAME):
            vb_destroy_machine(INSTANCE_NAME)


@skipIf(
    HAS_LIBS and vb_machine_exists(BOOTABLE_BASE_BOX_NAME) is False,
    "Bootable VM '{0}' not found. Cannot run tests.".format(BOOTABLE_BASE_BOX_NAME),
)
class VirtualboxProviderHeavyTests(VirtualboxCloudTestCase):
    """
    Tests that include actually booting a machine and doing operations on it that might be lengthy.
    """

    def assertIsIpAddress(self, ip_str):
        """
        Is it either a IPv4 or IPv6 address

        @param ip_str:
        @type ip_str: str
        @raise AssertionError
        """
        try:
            socket.inet_aton(ip_str)
        except Exception:  # pylint: disable=broad-except
            try:
                socket.inet_pton(socket.AF_INET6, ip_str)
            except Exception:  # pylint: disable=broad-except
                self.fail("{0} is not a valid IP address".format(ip_str))

    def setUp(self):
        """
        Sets up the test requirements
        """
        # check if appropriate cloud provider and profile files are present
        provider_str = CONFIG_NAME
        providers = self.run_cloud("--list-providers")
        log.debug("providers: %s", providers)

        if provider_str not in providers:
            self.skipTest(
                "Configuration file for {0} was not found. Check {0}.conf files "
                "in tests/integration/files/conf/cloud.*.d/ to run these tests.".format(
                    PROVIDER_NAME
                )
            )

        # check if personal access token, ssh_key_file, and ssh_key_names are present
        config_path = os.path.join(
            RUNTIME_VARS.FILES, "conf", "cloud.providers.d", PROVIDER_NAME + ".conf"
        )
        log.debug("config_path: %s", config_path)
        providers = cloud_providers_config(config_path)
        log.debug("config: %s", providers)
        config_path = os.path.join(
            RUNTIME_VARS.FILES, "conf", "cloud.profiles.d", PROVIDER_NAME + ".conf"
        )
        profiles = vm_profiles_config(config_path, providers)
        profile = profiles.get(DEPLOY_PROFILE_NAME)
        if not profile:
            self.skipTest(
                "Profile {0} was not found. Check {1}.conf files "
                "in tests/integration/files/conf/cloud.profiles.d/ to run these tests.".format(
                    DEPLOY_PROFILE_NAME, PROVIDER_NAME
                )
            )
        base_box_name = profile.get("clonefrom")

        if base_box_name != BOOTABLE_BASE_BOX_NAME:
            self.skipTest(
                "Profile {0} does not have a base box to clone from. Check {1}.conf files "
                "in tests/integration/files/conf/cloud.profiles.d/ to run these tests."
                'And add a "clone_from: {2}" to the profile'.format(
                    PROFILE_NAME, PROVIDER_NAME, BOOTABLE_BASE_BOX_NAME
                )
            )

    def tearDown(self):
        try:
            vb_stop_vm(BOOTABLE_BASE_BOX_NAME)
        except Exception:  # pylint: disable=broad-except
            pass

        if vb_machine_exists(INSTANCE_NAME):
            try:
                vb_stop_vm(INSTANCE_NAME)
                vb_destroy_machine(INSTANCE_NAME)
            except Exception as e:  # pylint: disable=broad-except
                log.warning("Possibly dirty state after exception", exc_info=True)

    def test_deploy(self):
        machines = self.run_cloud(
            "-p {0} {1} --log-level=debug".format(DEPLOY_PROFILE_NAME, INSTANCE_NAME)
        )
        self.assertIn(INSTANCE_NAME, machines.keys())
        machine = machines[INSTANCE_NAME]
        self.assertIn("deployed", machine)
        self.assertTrue(machine["deployed"], "Machine wasn't deployed :(")

    def test_start_stop_action(self):
        res = self.run_cloud_action("start", BOOTABLE_BASE_BOX_NAME, timeout=10)
        log.info(res)

        machine = res.get(BOOTABLE_BASE_BOX_NAME)
        self.assertIsNotNone(machine)
        expected_state = "Running"
        state = machine.get("state")
        self.assertEqual(state, expected_state)

        res = self.run_cloud_action("stop", BOOTABLE_BASE_BOX_NAME, timeout=10)
        log.info(res)

        machine = res.get(BOOTABLE_BASE_BOX_NAME)
        self.assertIsNotNone(machine)

        expected_state = "PoweredOff"
        state = machine.get("state")
        self.assertEqual(state, expected_state)

    def test_restart_action(self):
        pass

    def test_network_addresses(self):
        # Machine is off
        ip_addresses = vb_get_network_addresses(machine_name=BOOTABLE_BASE_BOX_NAME)

        network_count = len(ip_addresses)
        self.assertEqual(network_count, 0)

        # Machine is up again
        vb_start_vm(BOOTABLE_BASE_BOX_NAME)
        ip_addresses = vb_wait_for_network_address(
            20, machine_name=BOOTABLE_BASE_BOX_NAME
        )
        network_count = len(ip_addresses)
        self.assertGreater(network_count, 0)

        for ip_address in ip_addresses:
            self.assertIsIpAddress(ip_address)


@skipIf(HAS_LIBS is False, "The 'vboxapi' library is not available")
class BaseVirtualboxTests(TestCase):
    def test_get_manager(self):
        self.assertIsNotNone(vb_get_box())


class CreationDestructionVirtualboxTests(VirtualboxTestCase):
    def test_vm_creation_and_destruction(self):
        vm_name = BASE_BOX_NAME
        vb_create_machine(vm_name)
        self.assertMachineExists(vm_name)

        vb_destroy_machine(vm_name)
        self.assertMachineDoesNotExist(vm_name)


class CloneVirtualboxTests(VirtualboxTestCase):
    def setUp(self):
        self.vbox = vb_get_box()

        self.name = "SaltCloudVirtualboxTestVM"
        vb_create_machine(self.name)
        self.assertMachineExists(self.name)

    def tearDown(self):
        vb_destroy_machine(self.name)
        self.assertMachineDoesNotExist(self.name)

    def test_create_machine(self):
        vb_name = "NewTestMachine"
        machine = vb_clone_vm(name=vb_name, clone_from=self.name)
        self.assertEqual(machine.get("name"), vb_name)
        self.assertMachineExists(vb_name)

        vb_destroy_machine(vb_name)
        self.assertMachineDoesNotExist(vb_name)


@skipIf(
    HAS_LIBS and vb_machine_exists(BOOTABLE_BASE_BOX_NAME) is False,
    "Bootable VM '{0}' not found. Cannot run tests.".format(BOOTABLE_BASE_BOX_NAME),
)
class BootVirtualboxTests(VirtualboxTestCase):
    def test_start_stop(self):
        for i in range(2):
            machine = vb_start_vm(BOOTABLE_BASE_BOX_NAME, 20000)
            self.assertEqual(machine_get_machinestate_str(machine), "Running")

            machine = vb_stop_vm(BOOTABLE_BASE_BOX_NAME)
            self.assertEqual(machine_get_machinestate_str(machine), "PoweredOff")


class XpcomConversionTests(TestCase):
    @classmethod
    def _mock_xpcom_object(cls, interface_name=None, attributes=None):
        class XPCOM(object):
            def __str__(self):
                return "<XPCOM component '<unknown>' (implementing {0})>".format(
                    interface_name
                )

        o = XPCOM()

        if attributes and isinstance(attributes, dict):
            for key, value in six.iteritems(attributes):
                setattr(o, key, value)
        return o

    def test_unknown_object(self):
        xpcom = XpcomConversionTests._mock_xpcom_object()

        ret = vb_xpcom_to_attribute_dict(xpcom)
        self.assertDictEqual(ret, dict())

    def test_imachine_object_default(self):

        interface = "IMachine"
        imachine = XpcomConversionTests._mock_xpcom_object(interface)

        ret = vb_xpcom_to_attribute_dict(imachine, interface_name=interface)
        expected_attributes = XPCOM_ATTRIBUTES[interface]

        self.assertIsNotNone(expected_attributes, "%s is unknown")

        for key in ret:
            self.assertIn(key, expected_attributes)

    def test_override_attributes(self):

        expected_dict = {"herp": "derp", "lol": "rofl", "something": 12345}

        xpc = XpcomConversionTests._mock_xpcom_object(attributes=expected_dict)

        ret = vb_xpcom_to_attribute_dict(xpc, attributes=expected_dict.keys())
        self.assertDictEqual(ret, expected_dict)

    def test_extra_attributes(self):

        interface = "IMachine"
        expected_extras = {
            "extra": "extra",
        }
        expected_machine = dict(
            [(attribute, attribute) for attribute in XPCOM_ATTRIBUTES[interface]]
        )
        expected_machine.update(expected_extras)

        imachine = XpcomConversionTests._mock_xpcom_object(
            interface, attributes=expected_machine
        )

        ret = vb_xpcom_to_attribute_dict(
            imachine, interface_name=interface, extra_attributes=expected_extras.keys()
        )
        self.assertDictEqual(ret, expected_machine)

        ret_keys = ret.keys()
        for key in expected_extras:
            self.assertIn(key, ret_keys)

    def test_extra_nonexistent_attributes(self):
        expected_extra_dict = {"nonexistent": ""}
        xpcom = XpcomConversionTests._mock_xpcom_object()

        ret = vb_xpcom_to_attribute_dict(
            xpcom, extra_attributes=expected_extra_dict.keys()
        )
        self.assertDictEqual(ret, expected_extra_dict)

    def test_extra_nonexistent_attribute_with_default(self):
        expected_extras = [("nonexistent", list)]
        expected_extra_dict = {"nonexistent": []}
        xpcom = XpcomConversionTests._mock_xpcom_object()

        ret = vb_xpcom_to_attribute_dict(xpcom, extra_attributes=expected_extras)
        self.assertDictEqual(ret, expected_extra_dict)