path: root/nova/tests/unit/cmd/test_status.py

# Copyright 2016 IBM Corp.
#
#    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.

"""
Unit tests for the nova-status CLI interfaces.
"""

# NOTE(cdent): Additional tests of nova-status may be found in
# nova/tests/functional/test_nova_status.py. Those tests use the external
# PlacementFixture, which is only available in functional tests.

from io import StringIO
from unittest import mock

import fixtures
from keystoneauth1 import exceptions as ks_exc
from keystoneauth1 import loading as keystone
from keystoneauth1 import session
from oslo_upgradecheck import upgradecheck
from oslo_utils.fixture import uuidsentinel as uuids
from requests import models

from nova.cmd import status
import nova.conf
from nova import context
from nova import exception
# NOTE(mriedem): We only use objects as a convenience to populate the database
# in the tests, we don't use them in the actual CLI.
from nova import objects
from nova.objects import service
from nova import test
from nova.tests import fixtures as nova_fixtures


CONF = nova.conf.CONF


class TestNovaStatusMain(test.NoDBTestCase):
    """Tests for the basic nova-status command infrastructure."""

    def setUp(self):
        super(TestNovaStatusMain, self).setUp()
        self.output = StringIO()
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', self.output))

    @mock.patch.object(status.config, 'parse_args')
    @mock.patch.object(status, 'CONF')
    def _check_main(self, mock_CONF, mock_parse_args,
                    category_name='check', expected_return_value=0):
        mock_CONF.category.name = category_name
        return_value = status.main()

        self.assertEqual(expected_return_value, return_value)
        mock_CONF.register_cli_opt.assert_called_once_with(
            status.category_opt)

    @mock.patch.object(status.version, 'version_string_with_package',
                       return_value="x.x.x")
    def test_main_version(self, mock_version_string):
        self._check_main(category_name='version')
        self.assertEqual("x.x.x\n", self.output.getvalue())

    @mock.patch.object(status.cmd_common, 'print_bash_completion')
    def test_main_bash_completion(self, mock_print_bash):
        self._check_main(category_name='bash-completion')
        mock_print_bash.assert_called_once_with(status.CATEGORIES)

    @mock.patch.object(status.cmd_common, 'get_action_fn')
    def test_main(self, mock_get_action_fn):
        mock_fn = mock.Mock()
        mock_fn_args = [mock.sentinel.arg]
        mock_fn_kwargs = {'key': mock.sentinel.value}
        mock_get_action_fn.return_value = (mock_fn, mock_fn_args,
                                           mock_fn_kwargs)

        self._check_main(expected_return_value=mock_fn.return_value)
        mock_fn.assert_called_once_with(mock.sentinel.arg,
                                        key=mock.sentinel.value)

    @mock.patch.object(status.cmd_common, 'get_action_fn')
    def test_main_error(self, mock_get_action_fn):
        mock_fn = mock.Mock(side_effect=Exception('wut'))
        mock_get_action_fn.return_value = (mock_fn, [], {})

        self._check_main(expected_return_value=255)
        output = self.output.getvalue()
        self.assertIn('Error:', output)
        # assert the traceback is in the output
        self.assertIn('wut', output)


class TestPlacementCheck(test.NoDBTestCase):
    """Tests the nova-status placement checks.

    These are done with mock as the ability to replicate all failure
    domains otherwise is quite complicated. Using a devstack
    environment you can validate each of these tests are matching
    reality.
    """

    def setUp(self):
        super(TestPlacementCheck, self).setUp()
        self.output = StringIO()
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', self.output))
        self.cmd = status.UpgradeCommands()

    @mock.patch.object(keystone, "load_auth_from_conf_options")
    def test_no_auth(self, auth):
        """Test failure when no credentials are specified.

        Replicate in devstack: start devstack with or without
        placement engine, remove the auth section from the [placement]
        block in nova.conf.
        """
        auth.side_effect = ks_exc.MissingAuthPlugin()
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('No credentials specified', res.details)

    @mock.patch.object(keystone, "load_auth_from_conf_options")
    @mock.patch.object(session.Session, 'request')
    def _test_placement_get_interface(
            self, expected_interface, mock_get, mock_auth):

        def fake_request(path, method, *a, **kw):
            self.assertEqual(mock.sentinel.path, path)
            self.assertEqual('GET', method)
            self.assertIn('endpoint_filter', kw)
            self.assertEqual(expected_interface,
                             kw['endpoint_filter']['interface'])
            return mock.Mock(autospec=models.Response)

        mock_get.side_effect = fake_request
        self.cmd._placement_get(mock.sentinel.path)
        mock_auth.assert_called_once_with(status.CONF, 'placement')
        self.assertTrue(mock_get.called)

    def test_placement_get_interface_default(self):
        """Tests that we try internal, then public interface by default."""
        self._test_placement_get_interface(['internal', 'public'])

    def test_placement_get_interface_internal(self):
        """Tests that "internal" is specified for interface when configured."""
        self.flags(valid_interfaces='internal', group='placement')
        self._test_placement_get_interface(['internal'])

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_invalid_auth(self, get):
        """Test failure when wrong credentials are specified or service user
        doesn't exist.

        Replicate in devstack: start devstack with or without
        placement engine, specify random credentials in auth section
        from the [placement] block in nova.conf.

        """
        get.side_effect = ks_exc.Unauthorized()
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('Placement service credentials do not work', res.details)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_invalid_endpoint(self, get):
        """Test failure when no endpoint exists.

        Replicate in devstack: start devstack without placement
        engine, but create valid placement service user and specify it
        in auth section of [placement] in nova.conf.
        """
        get.side_effect = ks_exc.EndpointNotFound()
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('Placement API endpoint not found', res.details)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_discovery_failure(self, get):
        """Test failure when discovery for placement URL failed.

        Replicate in devstack: start devstack with placement
        engine, create valid placement service user and specify it
        in auth section of [placement] in nova.conf. Stop keystone service.
        """
        get.side_effect = ks_exc.DiscoveryFailure()
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('Discovery for placement API URI failed.', res.details)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_down_endpoint(self, get):
        """Test failure when endpoint is down.

        Replicate in devstack: start devstack with placement
        engine, disable placement engine apache config.
        """
        get.side_effect = ks_exc.NotFound()
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('Placement API does not seem to be running', res.details)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_valid_version(self, get):
        get.return_value = {
            "versions": [
                {
                    "min_version": "1.0",
                    "max_version": status.MIN_PLACEMENT_MICROVERSION,
                    "id": "v1.0"
                }
            ]
        }
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.SUCCESS, res.code)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_version_comparison_does_not_use_floats(self, get):
        # NOTE(rpodolyaka): previously _check_placement() coerced the version
        # numbers to floats prior to comparison, that would lead to failures
        # in cases like float('1.10') < float('1.4'). As we require 1.4+ now,
        # the _check_placement() call below will assert that version comparison
        # continues to work correctly when Placement API versions 1.10
        # (or newer) is released
        get.return_value = {
             "versions": [
                {
                    "min_version": "1.0",
                    "max_version": status.MIN_PLACEMENT_MICROVERSION,
                    "id": "v1.0"
                }
            ]
        }
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.SUCCESS, res.code)

    @mock.patch.object(status.UpgradeCommands, "_placement_get")
    def test_invalid_version(self, get):
        get.return_value = {
            "versions": [
                {
                    "min_version": "0.9",
                    "max_version": "0.9",
                    "id": "v1.0"
                }
            ]
        }
        res = self.cmd._check_placement()
        self.assertEqual(upgradecheck.Code.FAILURE, res.code)
        self.assertIn('Placement API version %s needed, you have 0.9' %
                      status.MIN_PLACEMENT_MICROVERSION, res.details)


class TestUpgradeCheckCellsV2(test.NoDBTestCase):
    """Tests for the nova-status upgrade cells v2 specific check."""

    # We'll setup the API DB fixture ourselves and slowly build up the
    # contents until the check passes.
    USES_DB_SELF = True

    def setUp(self):
        super(TestUpgradeCheckCellsV2, self).setUp()
        self.output = StringIO()
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', self.output))
        self.useFixture(nova_fixtures.Database(database='api'))
        self.cmd = status.UpgradeCommands()

    def test_check_no_cell_mappings(self):
        """The cells v2 check should fail because there are no cell mappings.
        """
        result = self.cmd._check_cellsv2()
        self.assertEqual(upgradecheck.Code.FAILURE, result.code)
        self.assertIn('There needs to be at least two cell mappings',
                      result.details)

    def _create_cell_mapping(self, uuid):
        cm = objects.CellMapping(
            context=context.get_admin_context(),
            uuid=uuid,
            name=uuid,
            transport_url='fake://%s/' % uuid,
            database_connection=uuid)
        cm.create()
        return cm

    def test_check_no_cell0_mapping(self):
        """We'll create two cell mappings but not have cell0 mapped yet."""
        for i in range(2):
            uuid = getattr(uuids, str(i))
            self._create_cell_mapping(uuid)

        result = self.cmd._check_cellsv2()
        self.assertEqual(upgradecheck.Code.FAILURE, result.code)
        self.assertIn('No cell0 mapping found', result.details)

    def test_check_no_host_mappings_with_computes(self):
        """Creates a cell0 and cell1 mapping but no host mappings and there are
        compute nodes in the cell database.
        """
        self._setup_cells()
        cn = objects.ComputeNode(
            context=context.get_admin_context(),
            host='fake-host',
            vcpus=4,
            memory_mb=8 * 1024,
            local_gb=40,
            vcpus_used=2,
            memory_mb_used=2 * 1024,
            local_gb_used=10,
            hypervisor_type='fake',
            hypervisor_version=1,
            cpu_info='{"arch": "x86_64"}')
        cn.create()

        result = self.cmd._check_cellsv2()
        self.assertEqual(upgradecheck.Code.FAILURE, result.code)
        self.assertIn('No host mappings found but there are compute nodes',
                      result.details)

    def test_check_no_host_mappings_no_computes(self):
        """Creates the cell0 and cell1 mappings but no host mappings and no
        compute nodes so it's assumed to be an initial install.
        """
        self._setup_cells()

        result = self.cmd._check_cellsv2()
        self.assertEqual(upgradecheck.Code.SUCCESS, result.code)
        self.assertIn('No host mappings or compute nodes were found',
                      result.details)

    def test_check_success(self):
        """Tests a successful cells v2 upgrade check."""
        # create the cell0 and first cell mappings
        self._setup_cells()
        # Start a compute service and create a hostmapping for it
        svc = self.start_service('compute')
        cell = self.cell_mappings[test.CELL1_NAME]
        hm = objects.HostMapping(context=context.get_admin_context(),
                                 host=svc.host,
                                 cell_mapping=cell)
        hm.create()

        result = self.cmd._check_cellsv2()
        self.assertEqual(upgradecheck.Code.SUCCESS, result.code)
        self.assertIsNone(result.details)


class TestUpgradeCheckCinderAPI(test.NoDBTestCase):

    def setUp(self):
        super(TestUpgradeCheckCinderAPI, self).setUp()
        self.cmd = status.UpgradeCommands()

    def test_cinder_not_configured(self):
        self.flags(auth_type=None, group='cinder')
        self.assertEqual(upgradecheck.Code.SUCCESS,
                         self.cmd._check_cinder().code)

    @mock.patch('nova.volume.cinder.is_microversion_supported',
                side_effect=exception.CinderAPIVersionNotAvailable(
                    version='3.44'))
    def test_microversion_not_available(self, mock_version_check):
        self.flags(auth_type='password', group='cinder')
        result = self.cmd._check_cinder()
        mock_version_check.assert_called_once()
        self.assertEqual(upgradecheck.Code.FAILURE, result.code)
        self.assertIn('Cinder API 3.44 or greater is required.',
                      result.details)

    @mock.patch('nova.volume.cinder.is_microversion_supported',
                side_effect=test.TestingException('oops'))
    def test_unknown_error(self, mock_version_check):
        self.flags(auth_type='password', group='cinder')
        result = self.cmd._check_cinder()
        mock_version_check.assert_called_once()
        self.assertEqual(upgradecheck.Code.WARNING, result.code)
        self.assertIn('oops', result.details)

    @mock.patch('nova.volume.cinder.is_microversion_supported')
    def test_microversion_available(self, mock_version_check):
        self.flags(auth_type='password', group='cinder')
        result = self.cmd._check_cinder()
        mock_version_check.assert_called_once()
        self.assertEqual(upgradecheck.Code.SUCCESS, result.code)


class TestUpgradeCheckOldCompute(test.NoDBTestCase):

    def setUp(self):
        super(TestUpgradeCheckOldCompute, self).setUp()
        self.cmd = status.UpgradeCommands()

    def test_no_compute(self):
        self.assertEqual(
            upgradecheck.Code.SUCCESS, self.cmd._check_old_computes().code)

    def test_only_new_compute(self):
        last_supported_version = service.SERVICE_VERSION_ALIASES[
            service.OLDEST_SUPPORTED_SERVICE_VERSION]
        with mock.patch(
                "nova.objects.service.get_minimum_version_all_cells",
                return_value=last_supported_version):
            self.assertEqual(
                upgradecheck.Code.SUCCESS, self.cmd._check_old_computes().code)

    def test_old_compute(self):
        too_old = service.SERVICE_VERSION_ALIASES[
            service.OLDEST_SUPPORTED_SERVICE_VERSION] - 1
        with mock.patch(
                "nova.objects.service.get_minimum_version_all_cells",
                return_value=too_old):
            result = self.cmd._check_old_computes()
            self.assertEqual(upgradecheck.Code.FAILURE, result.code)


class TestCheckMachineTypeUnset(test.NoDBTestCase):

    def setUp(self):
        super().setUp()
        self.cmd = status.UpgradeCommands()

    @mock.patch(
        'nova.virt.libvirt.machine_type_utils.get_instances_without_type',
        new=mock.Mock(return_value=[mock.Mock(spec=objects.Instance)]))
    def test_instances_found_without_hw_machine_type(self):
        result = self.cmd._check_machine_type_set()
        self.assertEqual(
            upgradecheck.Code.WARNING,
            result.code
        )

    @mock.patch(
        'nova.virt.libvirt.machine_type_utils.get_instances_without_type',
        new=mock.Mock(return_value=[]))
    def test_instances_not_found_without_hw_machine_type(self):
        result = self.cmd._check_machine_type_set()
        self.assertEqual(
            upgradecheck.Code.SUCCESS,
            result.code
        )