python-tripleoclient/tripleoclient/tests/v1/overcloud_config/test_overcloud_config.py

#   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.

import mock

from mock import call
from mock import patch
from osc_lib.tests import utils

from tripleoclient.tests.v1.overcloud_config import fakes
from tripleoclient.v1 import overcloud_config


class TestOvercloudConfig(utils.TestCommand):

    def setUp(self):
        super(TestOvercloudConfig, self).setUp()

        self.cmd = overcloud_config.DownloadConfig(self.app, None)
        self.app.client_manager.workflow_engine = mock.Mock()
        self.app.client_manager.orchestration = mock.Mock()
        self.workflow = self.app.client_manager.workflow_engine

    @patch.object(overcloud_config.DownloadConfig, '_mkdir')
    @patch.object(overcloud_config.DownloadConfig, '_open_file')
    @mock.patch('tempfile.mkdtemp', autospec=True)
    def test_overcloud_config_generate_config(self,
                                              mock_tmpdir,
                                              mock_open,
                                              mock_mkdir):
        arglist = ['--name', 'overcloud', '--config-dir', '/tmp']
        verifylist = [
            ('name', 'overcloud'),
            ('config_dir', '/tmp')
        ]
        config_type_list = ['config_settings', 'global_config_settings',
                            'logging_sources', 'monitoring_subscriptions',
                            'service_config_settings',
                            'service_metadata_settings',
                            'service_names',
                            'upgrade_batch_tasks', 'upgrade_tasks']
        fake_role = [role for role in
                     fakes.FAKE_STACK['outputs'][1]['output_value']]

        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
        clients = self.app.client_manager
        orchestration_client = clients.orchestration
        orchestration_client.stacks.get.return_value = fakes.create_tht_stack()
        mock_tmpdir.return_value = "/tmp/tht"
        self.cmd.take_action(parsed_args)
        expected_mkdir_calls = [call('/tmp/tht/%s' % r) for r in fake_role]
        mock_mkdir.assert_has_calls(expected_mkdir_calls, any_order=True)
        expected_calls = []
        for config in config_type_list:
            for role in fake_role:
                if config == 'step_config':
                    expected_calls += [call('/tmp/tht/%s/%s.pp' %
                                            (role, config))]
                else:
                    expected_calls += [call('/tmp/tht/%s/%s.yaml' %
                                            (role, config))]
        mock_open.assert_has_calls(expected_calls, any_order=True)

    @patch.object(overcloud_config.DownloadConfig, '_mkdir')
    @patch.object(overcloud_config.DownloadConfig, '_open_file')
    @mock.patch('tempfile.mkdtemp', autospec=True)
    def test_overcloud_config_one_config_type(self,
                                              mock_tmpdir,
                                              mock_open,
                                              mock_mkdir):

        arglist = ['--name', 'overcloud', '--config-dir', '/tmp',
                   '--config-type', ['config_settings']]
        verifylist = [
            ('name', 'overcloud'),
            ('config_dir', '/tmp'),
            ('config_type', ['config_settings'])
        ]
        expected_config_type = 'config_settings'
        fake_role = [role for role in
                     fakes.FAKE_STACK['outputs'][1]['output_value']]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)

        clients = self.app.client_manager
        orchestration_client = clients.orchestration
        orchestration_client.stacks.get.return_value = fakes.create_tht_stack()
        mock_tmpdir.return_value = "/tmp/tht"
        self.cmd.take_action(parsed_args)
        expected_mkdir_calls = [call('/tmp/tht/%s' % r) for r in fake_role]
        expected_calls = [call('/tmp/tht/%s/%s.yaml'
                          % (r, expected_config_type))
                          for r in fake_role]
        mock_mkdir.assert_has_calls(expected_mkdir_calls, any_order=True)
        mock_open.assert_has_calls(expected_calls, any_order=True)

    @mock.patch('os.mkdir')
    @mock.patch('six.moves.builtins.open')
    @mock.patch('tempfile.mkdtemp', autospec=True)
    def test_overcloud_config_wrong_config_type(self, mock_tmpdir,
                                                mock_open, mock_mkdir):

        arglist = [
            '--name', 'overcloud',
            '--config-dir',
            '/tmp', '--config-type', ['bad_config']]
        verifylist = [
            ('name', 'overcloud'),
            ('config_dir', '/tmp'),
            ('config_type', ['bad_config'])
        ]
        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
        clients = self.app.client_manager
        mock_tmpdir.return_value = "/tmp/tht"
        orchestration_client = clients.orchestration
        orchestration_client.stacks.get.return_value = fakes.create_tht_stack()
        self.assertRaises(
            KeyError,
            self.cmd.take_action, parsed_args)

    @mock.patch('tripleoclient.utils.get_role_data', autospec=True)
    @mock.patch('os.mkdir')
    @mock.patch('six.moves.builtins.open')
    @mock.patch('tempfile.mkdtemp', autospec=True)
    def test_overcloud_config_upgrade_tasks(self, mock_tmpdir,
                                            mock_open,
                                            mock_mkdir,
                                            mock_get_role_data):

        clients = self.app.client_manager
        orchestration_client = clients.orchestration
        orchestration_client.stacks.get.return_value = fakes.create_tht_stack()
        mock_tmpdir.return_value = "/tmp/tht"
        fake_role = [role for role in
                     fakes.FAKE_STACK['outputs'][1]['output_value']]
        fake_tasks = {'FakeController': [{'name': 'Stop fake service',
                                          'service': 'name=fake '
                                          'state=stopped',
                                          'tags': 'step1',
                                          'when': 'step == 1'}],
                      'FakeCompute': [{'name': 'Stop fake service',
                                       'service': 'name=fake state=stopped',
                                       'tags': 'step1',
                                       'when': 'step == 1'},
                                      {'name': 'Stop nova-'
                                       'compute service',
                                       'service':
                                       'name=openstack-nova-'
                                       'compute state=stopped',
                                       'tags': 'step1',
                                       'when': 'step == 1'}]
                      }
        mock_get_role_data.return_value = fake_role

        for role in fake_role:
            filepath = "/tmp/tht/%s/upgrade_tasks_playbook" % role
            with mock.patch('os.fdopen'), \
                    mock.patch('os.mkdir'), mock.patch('os.open') as open:
                playbook_tasks = self.cmd._write_playbook_get_tasks(
                    fakes.FAKE_STACK['outputs'][1]['output_value'][role]
                    ['upgrade_tasks'], role, filepath)
                self.assertEqual(fake_tasks[role], playbook_tasks)
                open.assert_called()