1241 lines
45 KiB
Python
1241 lines
45 KiB
Python
# Copyright 2015 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 fixtures
|
|
import mock
|
|
import os
|
|
import requests
|
|
import shutil
|
|
import six
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
import yaml
|
|
|
|
from osc_lib import exceptions as oscexc
|
|
from six.moves.urllib import parse
|
|
from tripleo_common.image import image_uploader
|
|
from tripleo_common.image import kolla_builder
|
|
from tripleoclient import constants
|
|
from tripleoclient.tests.v1.test_plugin import TestPluginV1
|
|
from tripleoclient.v1 import container_image
|
|
|
|
# TODO(sbaker) Remove after a tripleo-common release contains this attribute
|
|
CLEANUP = (
|
|
CLEANUP_FULL, CLEANUP_PARTIAL, CLEANUP_NONE
|
|
) = (
|
|
'full', 'partial', 'none'
|
|
)
|
|
if not hasattr(image_uploader, 'CLEANUP'):
|
|
setattr(image_uploader, 'CLEANUP', CLEANUP)
|
|
setattr(image_uploader, 'CLEANUP_FULL', CLEANUP_FULL)
|
|
setattr(image_uploader, 'CLEANUP_PARTIAL', CLEANUP_PARTIAL)
|
|
setattr(image_uploader, 'CLEANUP_NONE', CLEANUP_NONE)
|
|
|
|
|
|
class TestContainerImageUpload(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImageUpload, self).setUp()
|
|
|
|
# Get the command object to test
|
|
self.cmd = container_image.UploadImage(self.app, None)
|
|
|
|
@mock.patch('tripleo_common.utils.locks.processlock.'
|
|
'ProcessLock')
|
|
@mock.patch('sys.exit')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_container_image_upload_noargs(self, mock_manager, exit_mock,
|
|
mock_lock):
|
|
arglist = []
|
|
verifylist = []
|
|
|
|
mock_lockobj = mock.MagicMock()
|
|
mock_lock.return_value = mock_lockobj
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
# argparse will complain that --config-file is missing and exit with 2
|
|
exit_mock.assert_called_with(2)
|
|
|
|
@mock.patch('tripleo_common.utils.locks.processlock.'
|
|
'ProcessLock')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_container_image_upload_conf_files(self, mock_manager, mock_lock):
|
|
arglist = [
|
|
'--config-file',
|
|
'/tmp/foo.yaml',
|
|
'--config-file',
|
|
'/tmp/bar.yaml'
|
|
]
|
|
verifylist = []
|
|
|
|
mock_lockobj = mock.MagicMock()
|
|
mock_lock.return_value = mock_lockobj
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_manager.assert_called_once_with(
|
|
['/tmp/foo.yaml', '/tmp/bar.yaml'], cleanup='full',
|
|
lock=mock_lockobj)
|
|
mock_manager.return_value.upload.assert_called_once_with()
|
|
|
|
|
|
class TestContainerImagePush(TestPluginV1):
|
|
def setUp(self):
|
|
super(TestContainerImagePush, self).setUp()
|
|
|
|
lock = mock.patch('tripleo_common.utils.locks.processlock.ProcessLock')
|
|
self.mock_lock = lock.start()
|
|
self.addCleanup(lock.stop)
|
|
|
|
self.cmd = container_image.TripleOContainerImagePush(self.app, None)
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.UploadTask')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action(self, mock_manager, mock_task, mock_get_uc_registry):
|
|
arglist = ['docker.io/namespace/foo']
|
|
verifylist = [('image_to_push', 'docker.io/namespace/foo')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock upload task
|
|
mock_uploadtask = mock.Mock()
|
|
mock_task.return_value = mock_uploadtask
|
|
|
|
# mock add upload task action
|
|
mock_add_upload = mock.Mock()
|
|
data = []
|
|
mock_add_upload.return_value = data
|
|
mock_uploader.add_upload_task = mock_add_upload
|
|
|
|
# mock run tasks action
|
|
mock_run_tasks = mock.Mock()
|
|
mock_uploader.run_tasks = mock_run_tasks
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_task.assert_called_once_with(
|
|
image_name='namespace/foo',
|
|
pull_source='docker.io',
|
|
push_destination='uc.ctlplane.somedomain',
|
|
append_tag=parsed_args.append_tag,
|
|
modify_role=None,
|
|
modify_vars=None,
|
|
cleanup=False,
|
|
multi_arch=parsed_args.multi_arch)
|
|
|
|
mock_add_upload.assert_called_once_with(mock_uploadtask)
|
|
mock_run_tasks.assert_called_once()
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.UploadTask')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action_local(self, mock_manager, mock_task,
|
|
mock_get_uc_registry):
|
|
arglist = ['docker.io/namespace/foo', '--local']
|
|
verifylist = [('image_to_push', 'docker.io/namespace/foo'),
|
|
('local', True)]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock upload task
|
|
mock_uploadtask = mock.Mock()
|
|
mock_task.return_value = mock_uploadtask
|
|
|
|
# mock add upload task action
|
|
mock_add_upload = mock.Mock()
|
|
data = []
|
|
mock_add_upload.return_value = data
|
|
mock_uploader.add_upload_task = mock_add_upload
|
|
|
|
# mock run tasks action
|
|
mock_run_tasks = mock.Mock()
|
|
mock_uploader.run_tasks = mock_run_tasks
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_task.assert_called_once_with(
|
|
image_name='containers-storage:docker.io/namespace/foo',
|
|
pull_source=None,
|
|
push_destination='uc.ctlplane.somedomain',
|
|
append_tag=parsed_args.append_tag,
|
|
modify_role=None,
|
|
modify_vars=None,
|
|
cleanup=False,
|
|
multi_arch=parsed_args.multi_arch)
|
|
|
|
mock_add_upload.assert_called_once_with(mock_uploadtask)
|
|
mock_run_tasks.assert_called_once()
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.UploadTask')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action_local_path(self, mock_manager, mock_task,
|
|
mock_get_uc_registry):
|
|
arglist = ['containers-storage:docker.io/namespace/foo']
|
|
verifylist = [('image_to_push',
|
|
'containers-storage:docker.io/namespace/foo')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock upload task
|
|
mock_uploadtask = mock.Mock()
|
|
mock_task.return_value = mock_uploadtask
|
|
|
|
# mock add upload task action
|
|
mock_add_upload = mock.Mock()
|
|
data = []
|
|
mock_add_upload.return_value = data
|
|
mock_uploader.add_upload_task = mock_add_upload
|
|
|
|
# mock run tasks action
|
|
mock_run_tasks = mock.Mock()
|
|
mock_uploader.run_tasks = mock_run_tasks
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_task.assert_called_once_with(
|
|
image_name='containers-storage:docker.io/namespace/foo',
|
|
pull_source=None,
|
|
push_destination='uc.ctlplane.somedomain',
|
|
append_tag=parsed_args.append_tag,
|
|
modify_role=None,
|
|
modify_vars=None,
|
|
cleanup=False,
|
|
multi_arch=parsed_args.multi_arch)
|
|
|
|
mock_add_upload.assert_called_once_with(mock_uploadtask)
|
|
mock_run_tasks.assert_called_once()
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.UploadTask')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action_oserror(self, mock_manager, mock_task,
|
|
mock_get_uc_registry):
|
|
arglist = ['docker.io/namespace/foo']
|
|
verifylist = [('image_to_push', 'docker.io/namespace/foo')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock upload task
|
|
mock_uploadtask = mock.Mock()
|
|
mock_task.return_value = mock_uploadtask
|
|
|
|
# mock add upload task action
|
|
mock_add_upload = mock.Mock()
|
|
data = []
|
|
mock_add_upload.return_value = data
|
|
mock_uploader.add_upload_task = mock_add_upload
|
|
|
|
# mock run tasks action
|
|
mock_run_tasks = mock.Mock()
|
|
mock_run_tasks.side_effect = OSError('Fail')
|
|
mock_uploader.run_tasks = mock_run_tasks
|
|
|
|
self.assertRaises(oscexc.CommandError,
|
|
self.cmd.take_action,
|
|
parsed_args)
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.UploadTask')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action_all_options(self, mock_manager, mock_task,
|
|
mock_get_uc_registry):
|
|
arglist = ['--registry-url', '127.0.0.1:8787',
|
|
'--append-tag', 'test',
|
|
'--source-username', 'sourceuser',
|
|
'--source-password', 'sourcepassword',
|
|
'--username', 'user',
|
|
'--password', 'password',
|
|
'--dry-run',
|
|
'--multi-arch',
|
|
'--cleanup',
|
|
'docker.io/namespace/foo:tag']
|
|
verifylist = [('registry_url', '127.0.0.1:8787'),
|
|
('append_tag', 'test'),
|
|
('username', 'user'),
|
|
('password', 'password'),
|
|
('dry_run', True),
|
|
('multi_arch', True),
|
|
('cleanup', True),
|
|
('image_to_push', 'docker.io/namespace/foo:tag')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock upload task
|
|
mock_uploadtask = mock.Mock()
|
|
mock_task.return_value = mock_uploadtask
|
|
|
|
# mock add upload task action
|
|
mock_add_upload = mock.Mock()
|
|
data = []
|
|
mock_add_upload.return_value = data
|
|
mock_uploader.add_upload_task = mock_add_upload
|
|
|
|
# mock run tasks action
|
|
mock_run_tasks = mock.Mock()
|
|
mock_uploader.run_tasks = mock_run_tasks
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
source_url = parse.urlparse("docker://docker.io/namespace/foo:tag")
|
|
registry_url = parse.urlparse("docker://127.0.0.1:8787")
|
|
auth_calls = [mock.call(source_url,
|
|
parsed_args.source_username,
|
|
parsed_args.source_password),
|
|
mock.call(registry_url,
|
|
parsed_args.username,
|
|
parsed_args.password)]
|
|
mock_uploader.authenticate.assert_has_calls(auth_calls)
|
|
|
|
mock_task.assert_not_called()
|
|
mock_add_upload.assert_not_called()
|
|
mock_run_tasks.assert_not_called()
|
|
|
|
|
|
class TestContainerImageDelete(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImageDelete, self).setUp()
|
|
|
|
lock = mock.patch('tripleo_common.utils.locks.processlock.ProcessLock')
|
|
self.mock_lock = lock.start()
|
|
self.addCleanup(lock.stop)
|
|
|
|
self.cmd = container_image.TripleOContainerImageDelete(self.app, None)
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_oserror(self, mock_manager, mock_get_uc_registry):
|
|
|
|
arglist = ['-y', 'foo']
|
|
verifylist = [('yes', True),
|
|
('image_to_delete', 'foo')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return url object from uploader._image_to_url
|
|
mock_url = mock.Mock()
|
|
mock_url.geturl.return_value = 'munged-reg-url'
|
|
|
|
mock_uploader._image_to_url.return_value = mock_url
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
mock_uploader.delete.side_effect = OSError('Errno 13')
|
|
self.assertRaises(oscexc.CommandError,
|
|
self.cmd.take_action,
|
|
parsed_args)
|
|
mock_uploader.delete.assert_called_once_with('foo',
|
|
session=mock_session)
|
|
|
|
|
|
class TestContainerImageList(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImageList, self).setUp()
|
|
|
|
lock = mock.patch('tripleo_common.utils.locks.processlock.ProcessLock')
|
|
self.mock_lock = lock.start()
|
|
self.addCleanup(lock.stop)
|
|
|
|
self.cmd = container_image.TripleOContainerImageList(self.app, None)
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action(self, mock_manager, mock_get_uc_registry):
|
|
arglist = []
|
|
verifylist = []
|
|
|
|
mock_manager.return_value.uploader.return_value.list.return_value = \
|
|
['a', 'b']
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
rv = self.cmd.take_action(parsed_args)
|
|
actual = (('Image Name',), [('a',), ('b',)])
|
|
self.assertEqual(actual, rv)
|
|
|
|
@mock.patch('tripleo_common.image.image_uploader.get_undercloud_registry',
|
|
return_value='uc.ctlplane.somedomain')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action_auth(self, mock_manager, mock_get_uc_registry):
|
|
# check arg parsing items
|
|
arglist = ['--registry-url', 'reg-url',
|
|
'--username', 'foo',
|
|
'--password', 'bar']
|
|
verifylist = [('registry_url', 'reg-url'),
|
|
('username', 'foo'),
|
|
('password', 'bar')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return url object from uploader._image_to_url
|
|
mock_url = mock.Mock()
|
|
mock_url.geturl.return_value = 'munged-reg-url'
|
|
|
|
mock_uploader._image_to_url.return_value = mock_url
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
|
|
# mock image list function
|
|
mock_uploader.list.return_value = ['a', 'b']
|
|
|
|
rv = self.cmd.take_action(parsed_args)
|
|
|
|
# check various functions are called with expected inputs
|
|
mock_mgr.uploader.assert_called_with('python')
|
|
mock_uploader._image_to_url.assert_called_with('reg-url')
|
|
mock_uploader.authenticate.assert_called_with(mock_url, 'foo', 'bar')
|
|
mock_uploader.list.assert_called_with('munged-reg-url',
|
|
session=mock_session)
|
|
|
|
# check data format for lister
|
|
actual = (('Image Name',), [('a',), ('b',)])
|
|
self.assertEqual(actual, rv)
|
|
|
|
|
|
class TestContainerImageShow(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImageShow, self).setUp()
|
|
|
|
lock = mock.patch('tripleo_common.utils.locks.processlock.ProcessLock')
|
|
self.mock_lock = lock.start()
|
|
self.addCleanup(lock.stop)
|
|
|
|
self.cmd = container_image.TripleOContainerImageShow(self.app, None)
|
|
|
|
@mock.patch('tripleoclient.v1.container_image.TripleOContainerImageShow.'
|
|
'format_image_inspect')
|
|
@mock.patch('tripleo_common.image.image_uploader.ImageUploadManager')
|
|
def test_take_action(self, mock_manager, mock_formatter):
|
|
|
|
arglist = ['foo']
|
|
verifylist = [('image_to_inspect', 'foo')]
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
# mock manager object
|
|
mock_mgr = mock.Mock()
|
|
mock_manager.return_value = mock_mgr
|
|
|
|
# mock uploader object
|
|
mock_uploader = mock.Mock()
|
|
mock_mgr.uploader.return_value = mock_uploader
|
|
|
|
# mock return url object from uploader._image_to_url
|
|
mock_url = mock.Mock()
|
|
mock_url.geturl.return_value = 'munged-reg-url'
|
|
|
|
mock_uploader._image_to_url.return_value = mock_url
|
|
|
|
# mock return session object from uploader.authenticate
|
|
mock_session = mock.Mock()
|
|
mock_uploader.authenticate.return_value = mock_session
|
|
mock_inspect = mock.Mock()
|
|
data = {'Name': 'a', 'Layers': 'b'}
|
|
mock_inspect.return_value = data
|
|
mock_uploader.inspect = mock_inspect
|
|
|
|
# mock format image inspect
|
|
formatted_data = (['Name', 'Layers'], ['a', 'b'])
|
|
mock_formatter.return_value = formatted_data
|
|
|
|
rv = self.cmd.take_action(parsed_args)
|
|
|
|
mock_formatter.assert_called_once_with(data)
|
|
self.assertEqual(formatted_data, rv)
|
|
|
|
def test_format_image_inspect(self):
|
|
test_data = {'Name': 'foo', 'Layers': 'bar'}
|
|
self.assertEqual(self.cmd.format_image_inspect(test_data),
|
|
(['Name', 'Layers'], ['foo', 'bar']))
|
|
|
|
|
|
class TestContainerImagePrepare(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImagePrepare, self).setUp()
|
|
|
|
# Get the command object to test
|
|
self.cmd = container_image.PrepareImageFiles(self.app, None)
|
|
kolla_builder.DEFAULT_TEMPLATE_FILE = os.path.join(
|
|
'/tmp/overcloud_containers.yaml.j2'
|
|
)
|
|
self.roles_yaml = '''
|
|
- name: EnabledRole
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::AodhEvaluator
|
|
- name: RoleDisabledViaRolesData
|
|
CountDefault: 0
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::AodhApi
|
|
- name: RoleDisabledViaEnvironment
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::Disabled
|
|
- name: RoleOverwrittenViaEnvironment
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::Overwritten
|
|
'''
|
|
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'container_images_prepare_defaults', create=True)
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'container_images_prepare', create=True)
|
|
@mock.patch('requests.get')
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'build_service_filter')
|
|
def test_container_image_prepare_noargs(self, mock_bsf, mock_get, mock_cip,
|
|
mock_cipd):
|
|
mock_bsf.return_value = None
|
|
mock_cipd.return_value = {
|
|
'neutron_driver': 'ovn',
|
|
'name_suffix': '',
|
|
'tag': 'latest',
|
|
'namespace': 'docker.io/tripleomaster',
|
|
'name_prefix':
|
|
'centos-binary-'
|
|
}
|
|
arglist = []
|
|
verifylist = []
|
|
|
|
mock_cip.return_value = {'container_images.yaml': {}}
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_cip.assert_called_with(
|
|
excludes=[],
|
|
includes=[],
|
|
mapping_args={
|
|
'neutron_driver': 'ovn',
|
|
'name_suffix': '',
|
|
'tag': 'latest',
|
|
'namespace': 'docker.io/tripleomaster',
|
|
'name_prefix':
|
|
'centos-binary-'
|
|
},
|
|
output_env_file=None,
|
|
output_images_file='container_images.yaml',
|
|
push_destination=None,
|
|
service_filter=None,
|
|
tag_from_label=None,
|
|
modify_role=None,
|
|
modify_vars=None,
|
|
append_tag=None,
|
|
template_file=kolla_builder.DEFAULT_TEMPLATE_FILE
|
|
)
|
|
|
|
@mock.patch('tripleoclient.utils.fetch_roles_file')
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'container_images_prepare_defaults', create=True)
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'container_images_prepare', create=True)
|
|
@mock.patch('heatclient.common.template_utils.'
|
|
'process_multiple_environments_and_files')
|
|
@mock.patch('tripleo_common.image.kolla_builder.'
|
|
'build_service_filter')
|
|
@mock.patch('requests.get')
|
|
def test_container_image_prepare(self, mock_get, mock_bsf, pmef, mock_cip,
|
|
mock_cipd, mock_roles):
|
|
|
|
temp = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, temp)
|
|
images_file = os.path.join(temp, 'overcloud_containers.yaml')
|
|
env_file = os.path.join(temp, 'containers_env.yaml')
|
|
tmpl_file = os.path.join(temp, 'overcloud_containers.yaml.j2')
|
|
aodh_file = os.path.join(temp, 'docker', 'services', 'aodh.yaml')
|
|
roles_file = os.path.join(temp, 'roles_data.yaml')
|
|
with open(roles_file, 'w') as f:
|
|
f.write(self.roles_yaml)
|
|
modify_vars_file = os.path.join(temp, 'modify_vars.yaml')
|
|
with open(modify_vars_file, 'w') as f:
|
|
f.write('foo: bar')
|
|
mock_get.side_effect = requests.exceptions.SSLError('ouch')
|
|
mock_bsf.return_value = set(['OS::TripleO::Services::AodhEvaluator'])
|
|
|
|
resource_registry = {
|
|
'parameter_defaults': {
|
|
'NeutronMechanismDrivers': 'ovn',
|
|
},
|
|
'resource_registry': {
|
|
'OS::TripleO::Services::AodhEvaluator': aodh_file,
|
|
'OS::TripleO::Services::AodhApi': aodh_file
|
|
}
|
|
}
|
|
pmef.return_value = None, resource_registry
|
|
|
|
arglist = [
|
|
'--template-file',
|
|
tmpl_file,
|
|
'--tag',
|
|
'passed-ci',
|
|
'--namespace',
|
|
'192.0.2.0:8787/t',
|
|
'--prefix',
|
|
'os-',
|
|
'--suffix',
|
|
'foo',
|
|
'--output-images-file',
|
|
images_file,
|
|
'--output-env-file',
|
|
env_file,
|
|
'--set',
|
|
'ceph_namespace=myceph',
|
|
'--set',
|
|
'ceph_image=mydaemon',
|
|
'--set',
|
|
'ceph_tag=mytag',
|
|
'-e',
|
|
'environment/docker-ha.yaml',
|
|
'--roles-file',
|
|
roles_file,
|
|
'--modify-role',
|
|
'foo-role',
|
|
'--modify-vars',
|
|
modify_vars_file
|
|
]
|
|
self.cmd.app.command_options = arglist
|
|
verifylist = []
|
|
mock_cip.return_value = {images_file: [{
|
|
'imagename': '192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
}, {
|
|
'imagename': '192.0.2.0:8787/t/os-aodh-evaluatorfoo:passed-ci',
|
|
}], env_file: {
|
|
'DockerAodhApiImage':
|
|
'192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
'DockerAodhConfigImage':
|
|
'192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
'DockerAodhEvaluatorImage':
|
|
'192.0.2.0:8787/t/os-aodh-evaluatorfoo:passed-ci',
|
|
'DockerInsecureRegistryAddress': ['192.0.2.0:8787']
|
|
}
|
|
}
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
pmef.assert_called_once_with(['environment/docker-ha.yaml'],
|
|
env_path_is_object=mock.ANY,
|
|
object_request=mock.ANY)
|
|
mock_cip.assert_called_once_with(
|
|
excludes=[],
|
|
includes=[],
|
|
mapping_args={
|
|
'namespace': '192.0.2.0:8787/t',
|
|
'name_suffix': 'foo',
|
|
'ceph_tag': 'mytag',
|
|
'ceph_image': 'mydaemon',
|
|
'tag': 'passed-ci',
|
|
'neutron_driver': 'ovn',
|
|
'ceph_namespace': 'myceph',
|
|
'name_prefix': 'os-'
|
|
},
|
|
output_env_file=env_file,
|
|
output_images_file=images_file,
|
|
push_destination=None,
|
|
service_filter=set([
|
|
'OS::TripleO::Services::AodhEvaluator',
|
|
]),
|
|
tag_from_label=None,
|
|
modify_role='foo-role',
|
|
modify_vars={'foo': 'bar'},
|
|
append_tag=mock.ANY,
|
|
template_file=tmpl_file
|
|
)
|
|
ci_data = {
|
|
'container_images': [{
|
|
'imagename': '192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
}, {
|
|
'imagename': '192.0.2.0:8787/t/os-aodh-evaluatorfoo:passed-ci',
|
|
}]
|
|
}
|
|
env_data = {
|
|
'parameter_defaults': {
|
|
'DockerAodhApiImage':
|
|
'192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
'DockerAodhConfigImage':
|
|
'192.0.2.0:8787/t/os-aodh-apifoo:passed-ci',
|
|
'DockerAodhEvaluatorImage':
|
|
'192.0.2.0:8787/t/os-aodh-evaluatorfoo:passed-ci',
|
|
'DockerInsecureRegistryAddress': ['192.0.2.0:8787']
|
|
}
|
|
}
|
|
with open(images_file) as f:
|
|
self.assertEqual(ci_data, yaml.safe_load(f))
|
|
with open(env_file) as f:
|
|
self.assertEqual(env_data, yaml.safe_load(f))
|
|
|
|
|
|
class TestTripleoImagePrepare(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestTripleoImagePrepare, self).setUp()
|
|
# Get the command object to test
|
|
self.cmd = container_image.TripleOImagePrepare(self.app, None)
|
|
self.cmd.app_args = mock.Mock()
|
|
self.cmd.app_args.verbose_level = 3
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, self.temp_dir)
|
|
self.prepare_default_file = os.path.join(
|
|
self.temp_dir, 'prepare_env.yaml')
|
|
default_param = kolla_builder.CONTAINER_IMAGE_PREPARE_PARAM
|
|
self.default_env = {
|
|
'parameter_defaults': {
|
|
'ContainerImagePrepare': default_param
|
|
}
|
|
}
|
|
with open(self.prepare_default_file, 'w') as f:
|
|
yaml.safe_dump(self.default_env, f)
|
|
|
|
self.roles_yaml = '''
|
|
- name: EnabledRole
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::AodhEvaluator
|
|
- name: RoleDisabledViaRolesData
|
|
CountDefault: 0
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::AodhApi
|
|
- name: RoleDisabledViaEnvironment
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::Disabled
|
|
- name: RoleOverwrittenViaEnvironment
|
|
CountDefault: 1
|
|
ServicesDefault:
|
|
- OS::TripleO::Services::Overwritten
|
|
'''
|
|
self.roles_data_file = os.path.join(
|
|
self.temp_dir, 'roles_data.yaml')
|
|
with open(self.roles_data_file, 'w') as f:
|
|
f.write(self.roles_yaml)
|
|
|
|
@mock.patch('tripleoclient.utils.run_ansible_playbook',
|
|
autospec=True)
|
|
def test_tripleo_container_image_prepare(self, mock_playbook):
|
|
|
|
env_file = os.path.join(self.temp_dir, 'containers_env.yaml')
|
|
|
|
arglist = [
|
|
'--environment-file', self.prepare_default_file,
|
|
'--roles-file', self.roles_data_file,
|
|
'--output-env-file', env_file
|
|
]
|
|
verifylist = []
|
|
|
|
self.app.command_options = [
|
|
'tripleo', 'container', 'image', 'prepare'
|
|
] + arglist
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
self.cmd.take_action(parsed_args)
|
|
e_vars = {'roles_file': self.roles_data_file,
|
|
'environment_directories': [mock.ANY],
|
|
'environment_files': [self.temp_dir + '/prepare_env.yaml'],
|
|
'cleanup': 'full', 'dry_run': False,
|
|
'log_file': 'container_image_prepare.log', 'debug': True,
|
|
'output_env_file': self.temp_dir + '/containers_env.yaml'}
|
|
|
|
mock_playbook.assert_called_with(
|
|
extra_vars=e_vars,
|
|
inventory='localhost,',
|
|
playbook='cli-container-image-prepare.yaml',
|
|
playbook_dir='/usr/share/ansible/tripleo-playbooks',
|
|
verbosity=3, workdir=mock.ANY)
|
|
|
|
@mock.patch('tripleoclient.utils.rel_or_abs_path')
|
|
@mock.patch('tripleoclient.utils.run_ansible_playbook',
|
|
autospec=True)
|
|
def test_tripleo_container_image_prepare_paths(self, mock_playbook,
|
|
mock_path):
|
|
arglist = ['-e', 'foo.yaml',
|
|
'-e', '/bar.yaml',
|
|
'--environment-directory', 'foo',
|
|
'--environment-directory', '/bar',
|
|
'-r', 'foo.yaml']
|
|
verifylist = []
|
|
self.app.command_options = [
|
|
'tripleo', 'container', 'image', 'prepare'
|
|
] + arglist
|
|
|
|
mock_path.return_value = ('/usr/share/openstack-tripleo-heat-templates'
|
|
'/foo.yaml')
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
self.cmd.take_action(parsed_args)
|
|
e_vars = {'roles_file': ('/usr/share/'
|
|
'openstack-tripleo-heat-templates/foo.yaml'),
|
|
'environment_directories': [
|
|
os.path.expanduser(constants.DEFAULT_ENV_DIRECTORY),
|
|
os.getcwd() + '/foo',
|
|
'/bar'
|
|
],
|
|
'environment_files': [
|
|
os.getcwd() + '/foo.yaml',
|
|
'/bar.yaml'
|
|
],
|
|
'cleanup': 'full', 'dry_run': False,
|
|
'log_file': 'container_image_prepare.log', 'debug': True}
|
|
|
|
mock_playbook.assert_called_with(
|
|
extra_vars=e_vars,
|
|
inventory='localhost,',
|
|
playbook='cli-container-image-prepare.yaml',
|
|
playbook_dir='/usr/share/ansible/tripleo-playbooks',
|
|
verbosity=3, workdir=mock.ANY)
|
|
|
|
|
|
class TestTripleoImagePrepareDefault(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestTripleoImagePrepareDefault, self).setUp()
|
|
# Get the command object to test
|
|
self.cmd = container_image.TripleOImagePrepareDefault(self.app, None)
|
|
|
|
def test_prepare_default(self):
|
|
arglist = []
|
|
verifylist = []
|
|
|
|
self.app.command_options = [
|
|
'tripleo', 'container', 'image', 'prepare', 'default'
|
|
] + arglist
|
|
self.cmd.app.stdout = six.StringIO()
|
|
cmd = container_image.TripleOImagePrepareDefault(self.app, None)
|
|
|
|
parsed_args = self.check_parser(cmd, arglist, verifylist)
|
|
cmd.take_action(parsed_args)
|
|
|
|
result = self.app.stdout.getvalue()
|
|
expected_param = kolla_builder.CONTAINER_IMAGE_PREPARE_PARAM
|
|
expected = {
|
|
'parameter_defaults': {
|
|
'ContainerImagePrepare': expected_param
|
|
}
|
|
}
|
|
self.assertEqual(expected, yaml.safe_load(result))
|
|
|
|
def test_prepare_default_local_registry(self):
|
|
temp = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, temp)
|
|
env_file = os.path.join(temp, 'containers_env.yaml')
|
|
|
|
arglist = ['--local-push-destination', '--output-env-file', env_file]
|
|
verifylist = []
|
|
|
|
self.app.command_options = [
|
|
'tripleo', 'container', 'image', 'prepare', 'default'
|
|
] + arglist
|
|
cmd = container_image.TripleOImagePrepareDefault(self.app, None)
|
|
parsed_args = self.check_parser(cmd, arglist, verifylist)
|
|
|
|
cmd.take_action(parsed_args)
|
|
|
|
with open(env_file) as f:
|
|
result = yaml.safe_load(f)
|
|
self.assertEqual(
|
|
True,
|
|
result['parameter_defaults']['ContainerImagePrepare']
|
|
[0]['push_destination']
|
|
)
|
|
|
|
def test_prepare_default_registyr_login(self):
|
|
temp = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, temp)
|
|
env_file = os.path.join(temp, 'containers_env.yaml')
|
|
|
|
arglist = ['--enable-registry-login', '--output-env-file', env_file]
|
|
verifylist = []
|
|
|
|
self.app.command_options = [
|
|
'tripleo', 'container', 'image', 'prepare', 'default'
|
|
] + arglist
|
|
cmd = container_image.TripleOImagePrepareDefault(self.app, None)
|
|
parsed_args = self.check_parser(cmd, arglist, verifylist)
|
|
|
|
cmd.take_action(parsed_args)
|
|
|
|
with open(env_file) as f:
|
|
result = yaml.safe_load(f)
|
|
self.assertEqual(
|
|
True,
|
|
result['parameter_defaults']['ContainerImageRegistryLogin']
|
|
)
|
|
|
|
|
|
class TestContainerImageBuild(TestPluginV1):
|
|
|
|
def setUp(self):
|
|
super(TestContainerImageBuild, self).setUp()
|
|
|
|
# Get the command object to test
|
|
self.cmd = container_image.BuildImage(self.app, None)
|
|
self.cmd.app.stdout = six.StringIO()
|
|
self.uuid = str(uuid.uuid4())
|
|
self.temp_dir = os.path.join(self.useFixture(
|
|
fixtures.TempDir()).join())
|
|
self.temp_dir_uuid = os.path.join(self.temp_dir, self.uuid)
|
|
|
|
# Default conf file
|
|
self.default_kolla_conf = os.path.join(
|
|
sys.prefix, 'share', 'tripleo-common', 'container-images',
|
|
'tripleo_kolla_config_overrides.conf')
|
|
|
|
@mock.patch('sys.exit')
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('os.makedirs')
|
|
def test_container_image_build_noargs(self, mock_mkdirs, mock_builder,
|
|
exit_mock):
|
|
arglist = []
|
|
verifylist = []
|
|
mock_builder.return_value.build_images.return_value = 'done'
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
f, path = tempfile.mkstemp(dir=self.temp_dir)
|
|
with mock.patch('tempfile.mkstemp') as mock_mkstemp:
|
|
mock_mkstemp.return_value = f, path
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
# argparse will complain that --config-file is missing and exit with 2
|
|
exit_mock.assert_called_with(2)
|
|
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('uuid.uuid4')
|
|
def test_container_image_build(self, mock_uuid, mock_builder):
|
|
arglist = [
|
|
'--config-file',
|
|
'/tmp/foo.yaml',
|
|
'--config-file',
|
|
'/tmp/bar.yaml',
|
|
'--work-dir',
|
|
self.temp_dir,
|
|
'--kolla-config-file',
|
|
'/tmp/kolla.conf'
|
|
]
|
|
verifylist = []
|
|
mock_builder.return_value.build_images.return_value = 'done'
|
|
mock_uuid.return_value = self.uuid
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
f, path = tempfile.mkstemp(dir=self.temp_dir)
|
|
with mock.patch('tempfile.mkstemp') as mock_mkstemp:
|
|
with mock.patch('os.chdir'):
|
|
mock_mkstemp.return_value = f, path
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_builder.assert_called_once_with([
|
|
'/tmp/foo.yaml', '/tmp/bar.yaml'])
|
|
mock_builder.return_value.build_images.assert_called_once_with([
|
|
self.default_kolla_conf, '/tmp/kolla.conf',
|
|
path
|
|
], [], False, self.temp_dir_uuid)
|
|
|
|
@mock.patch('os.chdir')
|
|
@mock.patch('os.fdopen', autospec=True)
|
|
@mock.patch('tempfile.mkstemp')
|
|
@mock.patch(
|
|
'tripleoclient.utils.get_from_cfg')
|
|
@mock.patch(
|
|
'tripleoclient.utils.getboolean_from_cfg')
|
|
@mock.patch('tripleo_common.image.builder.buildah.BuildahBuilder',
|
|
autospec=True)
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('tripleoclient.utils.get_read_config')
|
|
@mock.patch('os.remove')
|
|
def test_container_image_build_with_buildah(self, mock_remove,
|
|
mock_read_conf,
|
|
mock_builder, mock_buildah,
|
|
mock_kolla_boolean_cfg,
|
|
mock_kolla_cfg, mock_mkstemp,
|
|
mock_fdopen, mock_chdir):
|
|
arglist = [
|
|
'--config-file',
|
|
'/tmp/bar.yaml',
|
|
'--kolla-config-file',
|
|
'/tmp/kolla.conf',
|
|
'--use-buildah',
|
|
'--work-dir',
|
|
self.temp_dir,
|
|
]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
deps = '{"base": ["qrouterd"]}'
|
|
mock_builder.return_value.build_images.return_value = deps
|
|
|
|
mock_mkstemp.return_value = (1, '/tmp/whatever_file')
|
|
|
|
mock_bb = mock.MagicMock()
|
|
mock_bb.build_all = mock.MagicMock()
|
|
mock_buildah.return_value = mock_bb
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
cfg_files = list(parsed_args.kolla_config_files)
|
|
cfg_files.append('/tmp/whatever_file')
|
|
mock_read_conf.assert_any_call(cfg_files)
|
|
cfg_calls = [
|
|
mock.call(mock_read_conf.return_value, 'base'),
|
|
mock.call(mock_read_conf.return_value, 'type'),
|
|
mock.call(mock_read_conf.return_value, 'tag'),
|
|
mock.call(mock_read_conf.return_value, 'namespace'),
|
|
mock.call(mock_read_conf.return_value, 'registry'),
|
|
]
|
|
cfg_boolean_calls = [
|
|
mock.call(mock_read_conf.return_value, 'push'),
|
|
]
|
|
mock_kolla_cfg.assert_has_calls(cfg_calls)
|
|
mock_kolla_boolean_cfg.assert_has_calls(cfg_boolean_calls)
|
|
mock_bb.build_all.assert_called_once()
|
|
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('uuid.uuid4')
|
|
def test_container_image_build_with_exclude(self, mock_uuid, mock_builder):
|
|
arglist = [
|
|
'--config-file',
|
|
'/tmp/foo.yaml',
|
|
'--config-file',
|
|
'/tmp/bar.yaml',
|
|
'--kolla-config-file',
|
|
'/tmp/kolla.conf',
|
|
'--work-dir',
|
|
'/tmp/testing',
|
|
'--exclude',
|
|
'foo',
|
|
'--exclude',
|
|
'bar'
|
|
]
|
|
verifylist = []
|
|
mock_builder.return_value.build_images.return_value = 'done'
|
|
mock_uuid.return_value = '123'
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
f, path = tempfile.mkstemp(dir=self.temp_dir)
|
|
with mock.patch('tempfile.mkdtemp') as mock_mkd:
|
|
mock_mkd.return_value = '/tmp/testing'
|
|
with mock.patch('tempfile.mkstemp') as mock_mkstemp:
|
|
with mock.patch('os.chdir'):
|
|
mock_mkstemp.return_value = f, path
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
mock_builder.assert_called_once_with([
|
|
'/tmp/foo.yaml', '/tmp/bar.yaml'])
|
|
mock_builder.return_value.build_images.assert_called_once_with([
|
|
self.default_kolla_conf, '/tmp/kolla.conf',
|
|
path
|
|
], ['foo', 'bar'], False, '/tmp/testing/123')
|
|
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('os.remove')
|
|
def test_container_image_build_list_images(self, mock_remove,
|
|
mock_builder):
|
|
arglist = [
|
|
'--list-images',
|
|
'--config-file',
|
|
'/tmp/bar.yaml',
|
|
'--work-dir',
|
|
'/tmp/testing',
|
|
'--kolla-config-file',
|
|
'/tmp/kolla.conf'
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
deps = '{"base": ["qrouterd"]}'
|
|
mock_builder.return_value.build_images.return_value = deps
|
|
|
|
f, path = tempfile.mkstemp(dir=self.temp_dir)
|
|
with mock.patch('tempfile.mkstemp') as mock_mkstemp:
|
|
mock_mkstemp.return_value = f, path
|
|
self.cmd.take_action(parsed_args)
|
|
with open(path, 'r') as conf_file:
|
|
self.assertEqual(
|
|
conf_file.readlines(),
|
|
['[DEFAULT]\n', 'list_dependencies=true'])
|
|
self.assertEqual('- base\n- qrouterd\n',
|
|
self.cmd.app.stdout.getvalue())
|
|
|
|
@mock.patch('tripleo_common.image.kolla_builder.KollaImageBuilder',
|
|
autospec=True)
|
|
@mock.patch('os.remove')
|
|
@mock.patch('os.makedirs')
|
|
def test_container_image_build_list_deps(self, mock_mkdirs, mock_remove,
|
|
mock_builder):
|
|
arglist = [
|
|
'--config-file',
|
|
'/tmp/bar.yaml',
|
|
'--kolla-config-file',
|
|
'/tmp/kolla.conf',
|
|
'--list-dependencies',
|
|
]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
deps = '{"base": ["qrouterd"]}'
|
|
mock_builder.return_value.build_images.return_value = deps
|
|
|
|
f, path = tempfile.mkstemp(dir=self.temp_dir)
|
|
with mock.patch('tempfile.mkstemp') as mock_mkstemp:
|
|
mock_mkstemp.return_value = f, path
|
|
self.cmd.take_action(parsed_args)
|
|
with open(path, 'r') as conf_file:
|
|
self.assertEqual(
|
|
conf_file.readlines(),
|
|
['[DEFAULT]\n', 'list_dependencies=true'])
|
|
self.assertEqual('base:\n- qrouterd\n',
|
|
self.cmd.app.stdout.getvalue())
|
|
|
|
def test_images_from_deps(self):
|
|
deps = yaml.safe_load('''base:
|
|
- qdrouterd
|
|
- cron
|
|
- ceph-base:
|
|
- ceph-osd
|
|
- ceph-rgw
|
|
- ceph-mon
|
|
- cephfs-fuse
|
|
- ceph-mds
|
|
- redis
|
|
- etcd
|
|
- kubernetes-entrypoint
|
|
- kolla-toolbox
|
|
- telegraf
|
|
- openstack-base:
|
|
- swift-base:
|
|
- swift-proxy-server
|
|
- swift-account
|
|
- swift-container
|
|
- swift-object-expirer
|
|
- swift-rsyncd
|
|
- swift-object''')
|
|
|
|
images_yaml = '''- base
|
|
- qdrouterd
|
|
- cron
|
|
- ceph-base
|
|
- ceph-osd
|
|
- ceph-rgw
|
|
- ceph-mon
|
|
- cephfs-fuse
|
|
- ceph-mds
|
|
- redis
|
|
- etcd
|
|
- kubernetes-entrypoint
|
|
- kolla-toolbox
|
|
- telegraf
|
|
- openstack-base
|
|
- swift-base
|
|
- swift-proxy-server
|
|
- swift-account
|
|
- swift-container
|
|
- swift-object-expirer
|
|
- swift-rsyncd
|
|
- swift-object
|
|
'''
|
|
images = []
|
|
self.cmd.images_from_deps(images, deps)
|
|
self.assertEqual(yaml.safe_load(images_yaml), images)
|