charm-nova-compute/unit_tests/test_lib_nova_compute_cloud_utils.py

# Copyright 2020 Canonical Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from unittest import TestCase
from unittest.mock import MagicMock, patch

import nova_compute.cloud_utils as cloud_utils


class NovaServiceMock:

    def __init__(self, id, host, binary):
        self.id = id
        self.host = host
        self.binary = binary


class NovaHypervisorMock:

    def __init__(self, hostname, running_vms=0):
        self.hypervisor_hostname = hostname
        self.running_vms = running_vms


class TestCloudUtils(TestCase):

    def __init__(self, methodName='runTest'):
        super(TestCloudUtils, self).__init__(methodName=methodName)

        self.nova_client = MagicMock()
        self.nova_cfg = {}

        nova_services = MagicMock()
        nova_services.list.return_value = []
        self.nova_client.services = nova_services

        nova_hypervisors = MagicMock()
        nova_hypervisors.list.return_value = []
        self.nova_client.hypervisors = nova_hypervisors

        self.neutron_client = MagicMock()

        self.unit_hostname = 'nova-compute-0'

    def setUp(self):
        to_patch = [
            'loading',
            'log',
            'nova_client_',
            '_nova_cfg',
        ]
        for object_ in to_patch:
            mock_ = patch.object(cloud_utils, object_, MagicMock())
            mock_.start()
            self.addCleanup(mock_.stop)

        cloud_utils._nova_cfg.return_value = self.nova_cfg
        cloud_utils.nova_client.return_value = self.nova_client

    def tearDown(self):
        # Cleanup any changes made to the self.nova_cfg
        self.nova_cfg = {}

    def test_os_credentials_content(self):
        """Test that function '_os_credentials' returns credentials
        in expected format"""
        credentials = cloud_utils._os_credentials()
        expected_keys = [
            'username',
            'password',
            'auth_url',
            'project_name',
            'project_domain_name',
            'user_domain_name',
        ]

        for key in expected_keys:
            self.assertIn(key, credentials.keys())

    def test_nova_service_id(self):
        """Test that `nova_service_id` returns expected nova service ID."""
        expected_id = 0
        other_host_id = 1
        other_host_name = "other-nova-compute-1"

        self.nova_client.services.list.return_value = [
            NovaServiceMock(expected_id, self.unit_hostname, 'nova-compute'),
            NovaServiceMock(other_host_id, other_host_name, 'nova-compute'),
        ]

        with patch.object(cloud_utils, "service_hostname",
                          return_value=self.unit_hostname):
            nova_id = cloud_utils.nova_service_id(self.nova_client)

        self.assertEqual(nova_id, expected_id)

    def test_nova_service_id_not_present(self):
        """Test that function 'nova_service_id' raises expected exception if
        current unit is not registered in 'nova-cloud-controller'"""
        nova_client = MagicMock()
        nova_services = MagicMock()
        nova_services.list.return_value = []
        nova_client.services = nova_services
        cloud_utils.nova_client.return_value = nova_client

        with patch.object(cloud_utils, "service_hostname",
                          return_value=self.unit_hostname):
            self.assertRaises(RuntimeError, cloud_utils.nova_service_id,
                              nova_client)

    def test_nova_service_id_multiple_services(self):
        """Test that function 'nova_service_id' will log warning and return
        first ID in the event that multiple nova-compute services are present
        on the same host"""
        first_id = 0
        second_id = 1
        warning_msg = 'Host "{}" has more than 1 nova-compute service ' \
                      'registered. Selecting one ID ' \
                      'randomly.'.format(self.unit_hostname)

        self.nova_client.services.list.return_value = [
            NovaServiceMock(first_id, self.unit_hostname, 'nova-compute'),
            NovaServiceMock(second_id, self.unit_hostname, 'nova-compute'),
        ]

        with patch.object(cloud_utils, "service_hostname",
                          return_value=self.unit_hostname):
            service_id = cloud_utils.nova_service_id(self.nova_client)

        self.assertEqual(service_id, first_id)
        cloud_utils.log.assert_called_with(warning_msg, cloud_utils.WARNING)

    def test_service_hostname_from_config(self):
        """Test that `service_hostname` func prioritizes hostname in config.

        In case that nova config contains "host" key in "DEFAULT" section,
        it should be used instead of calling `socket.gethostname()`.
        """
        expected_hostname = "nova-compute-0"
        self.nova_cfg["DEFAULT"] = {"host": expected_hostname}

        with patch.object(cloud_utils.socket, "gethostname",
                          return_value="foo"):
            self.assertEqual(cloud_utils.service_hostname(), expected_hostname)

    def test_service_hostname_from_gethostname(self):
        """Test that `service_hostname` falls back to socket.gethostname.

        In case that nova config does not contain "host" key in the "DEFAULT"
        section, this function should fall back to calling
        `socket.gethostname()`
        """
        expected_hostname = "nova-compute-0"
        self.nova_cfg["DEFAULT"] = {}

        with patch.object(cloud_utils.socket, "gethostname",
                          return_value=expected_hostname):
            self.assertEqual(cloud_utils.service_hostname(), expected_hostname)

    def test_running_vms(self):
        """Test that `running_vms` returns correct number of VMs."""
        expected_vms = 3
        expected_hostname = self.unit_hostname
        hostname_info = {"host_fqdn": expected_hostname}
        self.nova_client.hypervisors.list.return_value = [
            NovaHypervisorMock(expected_hostname, expected_vms),
            NovaHypervisorMock("other-nova-compute-0", 0)
        ]

        with patch.object(cloud_utils.HostInfoContext, '__call__',
                          return_value=hostname_info):
            vm_count = cloud_utils.running_vms(self.nova_client)

        self.assertEqual(vm_count, expected_vms)

    def test_running_vms_not_found(self):
        """Test error raised if the hypervisor is not find in the nova list."""
        hostname_info = {"host_fqdn": self.unit_hostname}
        expected_error = ("Nova compute node '{}' not found in the list of "
                          "hypervisors. Is the unit already removed from the"
                          " cloud?").format(self.unit_hostname)

        with patch.object(cloud_utils.HostInfoContext, '__call__',
                          return_value=hostname_info):
            with self.assertRaises(RuntimeError) as exc:
                cloud_utils.running_vms(self.nova_client)

        self.assertEqual(str(exc.exception), expected_error)