463 lines
19 KiB
Python
463 lines
19 KiB
Python
# Copyright 2021 Red Hat, Inc.
|
|
#
|
|
# 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 configparser
|
|
import copy
|
|
import ddt
|
|
import os
|
|
import subprocess
|
|
from unittest import mock
|
|
|
|
from . import fakes
|
|
from . import test_main
|
|
import tripleo_repos.yum_config.constants as const
|
|
import tripleo_repos.yum_config.exceptions as exc
|
|
import tripleo_repos.yum_config.yum_config as yum_cfg
|
|
import tripleo_repos.utils as repos_utils
|
|
|
|
|
|
@ddt.ddt
|
|
class TestTripleOYumConfig(test_main.TestTripleoYumConfigBase):
|
|
"""Tests for TripleYumConfig class and its methods."""
|
|
|
|
def _create_yum_config_obj(self, dir_path=None, valid_options=None,
|
|
file_extension=None):
|
|
self.mock_object(os.path, 'isfile')
|
|
self.mock_object(os, 'access')
|
|
self.mock_object(os.path, 'isdir')
|
|
return yum_cfg.TripleOYumConfig(dir_path=dir_path,
|
|
valid_options=valid_options,
|
|
file_extension=file_extension)
|
|
|
|
def test_tripleo_yum_config_invalid_dir_path(self):
|
|
self.mock_object(os.path, 'isdir',
|
|
mock.Mock(return_value=False))
|
|
|
|
self.assertRaises(exc.TripleOYumConfigNotFound,
|
|
yum_cfg.TripleOYumConfig,
|
|
dir_path='fake_dir_path')
|
|
|
|
def test_read_config_file_path(self):
|
|
yum_config = self._create_yum_config_obj()
|
|
|
|
parser_mock = mock.Mock()
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(return_value=parser_mock))
|
|
read_mock = self.mock_object(parser_mock, 'read')
|
|
self.mock_object(parser_mock, 'sections',
|
|
mock.Mock(return_value=fakes.FAKE_SECTIONS))
|
|
|
|
config_parser, file_path = yum_config._read_config_file(
|
|
fakes.FAKE_FILE_PATH,
|
|
section=fakes.FAKE_SECTION1
|
|
)
|
|
|
|
self.assertEqual(parser_mock, config_parser)
|
|
self.assertEqual(fakes.FAKE_FILE_PATH, file_path)
|
|
read_mock.assert_called_once_with(fakes.FAKE_FILE_PATH)
|
|
|
|
def test_read_config_file_path_parse_error(self):
|
|
yum_config = self._create_yum_config_obj()
|
|
|
|
parser_mock = mock.Mock()
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(return_value=parser_mock))
|
|
read_mock = self.mock_object(parser_mock, 'read',
|
|
mock.Mock(side_effect=configparser.Error))
|
|
|
|
self.assertRaises(exc.TripleOYumConfigFileParseError,
|
|
yum_config._read_config_file,
|
|
fakes.FAKE_FILE_PATH,
|
|
section=fakes.FAKE_SECTION1)
|
|
|
|
read_mock.assert_called_once_with(fakes.FAKE_FILE_PATH)
|
|
|
|
def test_read_config_file_path_invalid_section(self):
|
|
yum_config = self._create_yum_config_obj()
|
|
|
|
parser_mock = mock.Mock()
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(return_value=parser_mock))
|
|
read_mock = self.mock_object(parser_mock, 'read')
|
|
self.mock_object(parser_mock, 'sections',
|
|
mock.Mock(return_value=fakes.FAKE_SECTIONS))
|
|
|
|
self.assertRaises(exc.TripleOYumConfigInvalidSection,
|
|
yum_config._read_config_file,
|
|
fakes.FAKE_FILE_PATH,
|
|
section='invalid_section')
|
|
|
|
read_mock.assert_called_once_with(fakes.FAKE_FILE_PATH)
|
|
|
|
def test_get_config_files(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
dir_path=fakes.FAKE_DIR_PATH,
|
|
file_extension='.conf')
|
|
parser_mocks = []
|
|
for i in range(3):
|
|
parser_mock = mock.Mock()
|
|
parser_mocks.append(parser_mock)
|
|
self.mock_object(parser_mock, 'read')
|
|
|
|
self.mock_object(parser_mocks[0], 'sections',
|
|
mock.Mock(return_value=[]))
|
|
# second file inside dir will have the expected sections
|
|
self.mock_object(parser_mocks[1], 'sections',
|
|
mock.Mock(return_value=fakes.FAKE_SECTIONS))
|
|
self.mock_object(parser_mocks[2], 'sections',
|
|
mock.Mock(return_value=[]))
|
|
self.mock_object(os, 'listdir',
|
|
mock.Mock(return_value=fakes.FAKE_DIR_FILES))
|
|
self.mock_object(os, 'access', mock.Mock(return_value=True))
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(side_effect=parser_mocks))
|
|
|
|
result = yum_config._get_config_files(fakes.FAKE_SECTION1)
|
|
expected_dir_path = [os.path.join(fakes.FAKE_DIR_PATH,
|
|
fakes.FAKE_DIR_FILES[1])]
|
|
|
|
self.assertEqual(expected_dir_path, result)
|
|
|
|
@mock.patch('builtins.open')
|
|
def test_update_section(self, open):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
config_parser = fakes.FakeConfigParser({fakes.FAKE_SECTION1: {}})
|
|
|
|
mock_read_config = self.mock_object(
|
|
yum_config, '_read_config_file',
|
|
mock.Mock(return_value=(config_parser, fakes.FAKE_FILE_PATH)))
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
|
|
yum_config.update_section(fakes.FAKE_SECTION1, updates,
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
mock_read_config.assert_called_once_with(fakes.FAKE_FILE_PATH,
|
|
section=fakes.FAKE_SECTION1)
|
|
|
|
def test_update_section_invalid_options(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
|
|
updates = {'invalid_option': 'new_fake_value'}
|
|
|
|
self.assertRaises(exc.TripleOYumConfigInvalidOption,
|
|
yum_config.update_section,
|
|
fakes.FAKE_SECTION1,
|
|
updates,
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
def test_update_section_file_not_found(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
mock_get_configs = self.mock_object(
|
|
yum_config, '_get_config_files',
|
|
mock.Mock(return_value=[]))
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
|
|
self.assertRaises(exc.TripleOYumConfigNotFound,
|
|
yum_config.update_section,
|
|
fakes.FAKE_SECTION1,
|
|
updates)
|
|
mock_get_configs.assert_called_once_with(fakes.FAKE_SECTION1)
|
|
|
|
@mock.patch('builtins.open')
|
|
def test_add_section(self, open):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
config_parser = fakes.FakeConfigParser({fakes.FAKE_SECTION1: {}})
|
|
|
|
mock_read_config = self.mock_object(
|
|
yum_config, '_read_config_file',
|
|
mock.Mock(return_value=(config_parser, fakes.FAKE_FILE_PATH)))
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
|
|
yum_config.add_section(fakes.FAKE_SECTION2, updates,
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
mock_read_config.assert_called_once_with(
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
@mock.patch('builtins.open')
|
|
def test_add_section_already_exists(self, open):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
config_parser = fakes.FakeConfigParser({fakes.FAKE_SECTION1: {}})
|
|
|
|
mock_read_config = self.mock_object(
|
|
yum_config, '_read_config_file',
|
|
mock.Mock(return_value=(config_parser, fakes.FAKE_FILE_PATH)))
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
|
|
self.assertRaises(exc.TripleOYumConfigInvalidSection,
|
|
yum_config.add_section,
|
|
fakes.FAKE_SECTION1, updates,
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
mock_read_config.assert_called_once_with(
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
@mock.patch('builtins.open')
|
|
def test_add_update_all_sections(self, open):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
config_parser = fakes.FakeConfigParser({fakes.FAKE_SECTION1: {}})
|
|
|
|
mock_read_config = self.mock_object(
|
|
yum_config, '_read_config_file',
|
|
mock.Mock(return_value=(config_parser, fakes.FAKE_FILE_PATH)))
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
|
|
yum_config.update_all_sections(updates, fakes.FAKE_FILE_PATH)
|
|
|
|
mock_read_config.assert_called_once_with(fakes.FAKE_FILE_PATH)
|
|
|
|
def test_source_env_file(self):
|
|
p_open_mock = mock.Mock()
|
|
mock_open = self.mock_object(subprocess, 'Popen',
|
|
mock.Mock(return_value=p_open_mock))
|
|
data_mock = mock.Mock()
|
|
self.mock_object(data_mock, 'decode',
|
|
mock.Mock(return_value=fakes.FAKE_ENV_OUTPUT))
|
|
self.mock_object(p_open_mock, 'communicate',
|
|
mock.Mock(return_value=[data_mock]))
|
|
env_update_mock = self.mock_object(os.environ, 'update')
|
|
|
|
yum_cfg.source_env_file('fake_source_file', update=True)
|
|
|
|
exp_env_dict = dict(
|
|
line.split("=", 1) for line in fakes.FAKE_ENV_OUTPUT.splitlines()
|
|
if len(line.split("=", 1)) > 1)
|
|
|
|
mock_open.assert_called_once_with(". fake_source_file; env",
|
|
stdout=subprocess.PIPE,
|
|
shell=True)
|
|
env_update_mock.assert_called_once_with(exp_env_dict)
|
|
|
|
def test_get_config_from_url_invalid_url(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
fake_context = mock.Mock()
|
|
self.mock_object(repos_utils, 'http_get',
|
|
mock.Mock(return_value=(fake_context, 404)))
|
|
|
|
self.assertRaises(exc.TripleOYumConfigUrlError,
|
|
yum_config.get_config_from_url,
|
|
fakes.FAKE_REPO_DOWN_URL)
|
|
|
|
def test_get_config_from_url(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
fake_context = mock.Mock()
|
|
self.mock_object(repos_utils, 'http_get',
|
|
mock.Mock(return_value=(fake_context, 200)))
|
|
parser_mock = mock.Mock()
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(return_value=parser_mock))
|
|
|
|
result = yum_config.get_config_from_url(fakes.FAKE_REPO_DOWN_URL)
|
|
|
|
self.assertEqual(parser_mock, result)
|
|
|
|
def test_get_options_from_url_section_not_found(self):
|
|
yum_config = self._create_yum_config_obj(
|
|
valid_options=fakes.FAKE_SUPP_OPTIONS)
|
|
fake_config = mock.Mock()
|
|
self.mock_object(fake_config, 'sections',
|
|
mock.Mock(return_value=[]))
|
|
mock_get_from_url = self.mock_object(
|
|
yum_config, 'get_config_from_url',
|
|
mock.Mock(return_value=fake_config))
|
|
|
|
self.assertRaises(exc.TripleOYumConfigInvalidSection,
|
|
yum_config.get_options_from_url,
|
|
fakes.FAKE_REPO_DOWN_URL,
|
|
fakes.FAKE_SECTION1)
|
|
|
|
mock_get_from_url.assert_called_once_with(fakes.FAKE_REPO_DOWN_URL)
|
|
|
|
|
|
@ddt.ddt
|
|
class TestTripleOYumRepoConfig(test_main.TestTripleoYumConfigBase):
|
|
"""Tests for TripleOYumRepoConfig class and its methods."""
|
|
|
|
def setUp(self):
|
|
super(TestTripleOYumRepoConfig, self).setUp()
|
|
self.config_obj = yum_cfg.TripleOYumRepoConfig(
|
|
dir_path='/tmp'
|
|
)
|
|
|
|
@ddt.data(True, False, None)
|
|
def test_yum_repo_config_update_section(self, enable):
|
|
self.mock_object(os.path, 'isfile')
|
|
self.mock_object(os, 'access')
|
|
self.mock_object(os.path, 'isdir')
|
|
|
|
mock_update = self.mock_object(yum_cfg.TripleOYumConfig,
|
|
'update_section')
|
|
|
|
updates = {fakes.FAKE_OPTION1: 'new_fake_value'}
|
|
expected_updates = copy.copy(updates)
|
|
if enable is not None:
|
|
expected_updates['enabled'] = '1' if enable else '0'
|
|
|
|
self.config_obj.update_section(
|
|
fakes.FAKE_SECTION1, set_dict=updates,
|
|
file_path=fakes.FAKE_FILE_PATH, enabled=enable)
|
|
|
|
mock_update.assert_called_once_with(fakes.FAKE_SECTION1,
|
|
expected_updates,
|
|
file_path=fakes.FAKE_FILE_PATH)
|
|
|
|
@mock.patch('builtins.open')
|
|
@ddt.data(None, fakes.FAKE_REPO_DOWN_URL)
|
|
def test_add_or_update_section(self, open, down_url):
|
|
mock_update = self.mock_object(
|
|
self.config_obj, 'update_section',
|
|
mock.Mock(side_effect=exc.TripleOYumConfigNotFound(
|
|
error_msg='error')))
|
|
mock_add_section = self.mock_object(self.config_obj, 'add_section')
|
|
extra_opt = {'key1': 'new value 1'}
|
|
mock_get_from_url = self.mock_object(
|
|
self.config_obj, 'get_options_from_url',
|
|
mock.Mock(return_value=extra_opt))
|
|
|
|
self.config_obj.add_or_update_section(
|
|
fakes.FAKE_SECTION1,
|
|
set_dict=fakes.FAKE_SET_DICT,
|
|
file_path=fakes.FAKE_FILE_PATH,
|
|
enabled=True,
|
|
create_if_not_exists=True,
|
|
from_url=down_url)
|
|
|
|
fake_set_dict = copy.deepcopy(fakes.FAKE_SET_DICT)
|
|
fake_set_dict['name'] = fakes.FAKE_SECTION1
|
|
if down_url:
|
|
fake_set_dict.update(extra_opt)
|
|
mock_get_from_url.assert_called_once_with(down_url,
|
|
fakes.FAKE_SECTION1)
|
|
|
|
mock_update.assert_called_once_with(fakes.FAKE_SECTION1,
|
|
set_dict=fake_set_dict,
|
|
file_path=fakes.FAKE_FILE_PATH,
|
|
enabled=True)
|
|
mock_add_section.assert_called_once_with(
|
|
fakes.FAKE_SECTION1,
|
|
fake_set_dict,
|
|
fakes.FAKE_FILE_PATH,
|
|
enabled=True)
|
|
|
|
@ddt.data((fakes.FAKE_FILE_PATH, False), (None, True))
|
|
@ddt.unpack
|
|
def test_add_or_update_section_file_not_found(self, fake_path,
|
|
create_if_not_exists):
|
|
mock_update = self.mock_object(
|
|
self.config_obj, 'update_section',
|
|
mock.Mock(side_effect=exc.TripleOYumConfigNotFound(
|
|
error_msg='error')))
|
|
|
|
self.assertRaises(
|
|
exc.TripleOYumConfigNotFound,
|
|
self.config_obj.add_or_update_section,
|
|
fakes.FAKE_SECTION1,
|
|
set_dict=fakes.FAKE_SET_DICT,
|
|
file_path=fake_path,
|
|
enabled=True,
|
|
create_if_not_exists=create_if_not_exists)
|
|
|
|
fake_set_dict = copy.deepcopy(fakes.FAKE_SET_DICT)
|
|
fake_set_dict['name'] = fakes.FAKE_SECTION1
|
|
mock_update.assert_called_once_with(fakes.FAKE_SECTION1,
|
|
set_dict=fake_set_dict,
|
|
file_path=fake_path,
|
|
enabled=True)
|
|
|
|
@ddt.data(None, False, True)
|
|
def test_add_section(self, enabled):
|
|
mock_add = self.mock_object(yum_cfg.TripleOYumConfig, 'add_section')
|
|
|
|
self.config_obj.add_section(
|
|
fakes.FAKE_SECTION1, fakes.FAKE_SET_DICT,
|
|
fakes.FAKE_FILE_PATH, enabled=enabled)
|
|
|
|
updated_dict = copy.deepcopy(fakes.FAKE_SET_DICT)
|
|
if enabled is not None:
|
|
updated_dict['enabled'] = '1' if enabled else '0'
|
|
mock_add.assert_called_once_with(fakes.FAKE_SECTION1, updated_dict,
|
|
fakes.FAKE_FILE_PATH)
|
|
|
|
@ddt.data(fakes.FAKE_FILE_PATH, None)
|
|
def test_add_or_update_all_sections_from_url(self, file_path):
|
|
add_or_update_section = self.mock_object(
|
|
self.config_obj, 'add_or_update_section')
|
|
fake_config = mock.Mock()
|
|
self.mock_object(fake_config, 'sections',
|
|
mock.Mock(return_value=[fakes.FAKE_SECTION1]))
|
|
options_from_url = {'key3': 'value3'}
|
|
self.mock_object(fake_config, 'items',
|
|
mock.Mock(return_value=options_from_url))
|
|
mock_get_from_url = self.mock_object(
|
|
self.config_obj, 'get_config_from_url',
|
|
mock.Mock(return_value=fake_config))
|
|
exp_file_path = (
|
|
file_path or os.path.join(
|
|
'/tmp', fakes.FAKE_REPO_DOWN_URL.split('/')[-1])
|
|
)
|
|
|
|
self.config_obj.add_or_update_all_sections_from_url(
|
|
fakes.FAKE_REPO_DOWN_URL,
|
|
file_path=file_path,
|
|
set_dict=fakes.FAKE_SET_DICT,
|
|
enabled=True,
|
|
create_if_not_exists=True)
|
|
|
|
mock_get_from_url.assert_called_once_with(fakes.FAKE_REPO_DOWN_URL)
|
|
expected_update_dict = copy.deepcopy(fakes.FAKE_SET_DICT)
|
|
expected_update_dict.update(options_from_url)
|
|
add_or_update_section.assert_called_once_with(
|
|
fakes.FAKE_SECTION1, set_dict=expected_update_dict,
|
|
file_path=exp_file_path, enabled=True,
|
|
create_if_not_exists=True)
|
|
|
|
|
|
@ddt.ddt
|
|
class TestTripleOYumGlobalConfig(test_main.TestTripleoYumConfigBase):
|
|
"""Tests for TripleOYumGlobalConfig class and its methods."""
|
|
|
|
@mock.patch('builtins.open')
|
|
def test_create_yum_global_config_create_yum_conf(self, open):
|
|
self.mock_object(os, 'access')
|
|
self.mock_object(os.path, 'isdir')
|
|
self.mock_object(os.path, 'isfile',
|
|
mock.Mock(side_effect=[False, True]))
|
|
|
|
fake_cfg_parser = mock.Mock()
|
|
mock_read = self.mock_object(fake_cfg_parser, 'read')
|
|
mock_add = self.mock_object(fake_cfg_parser, 'add_section')
|
|
mock_write = self.mock_object(fake_cfg_parser, 'write')
|
|
self.mock_object(configparser, 'ConfigParser',
|
|
mock.Mock(return_value=fake_cfg_parser))
|
|
|
|
cfg_obj = yum_cfg.TripleOYumGlobalConfig()
|
|
|
|
self.assertIsNotNone(cfg_obj)
|
|
mock_read.assert_called_once_with(const.YUM_GLOBAL_CONFIG_FILE_PATH)
|
|
mock_add.assert_called_once_with('main')
|
|
mock_write.assert_called_once()
|