# 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 datetime import fixtures import jinja2 import os import requests import sys import tarfile import tempfile from unittest import mock from kolla.cmd import build as build_cmd from kolla.common import utils as common_utils from kolla.image import build from kolla.image.kolla_worker import Image from kolla.image import tasks from kolla.image import utils from kolla.tests import base FAKE_IMAGE = Image( 'image-base', 'image-base:latest', '/fake/path', parent_name=None, parent=None, status=utils.Status.MATCHED) FAKE_IMAGE_CHILD = Image( 'image-child', 'image-child:latest', '/fake/path2', parent_name='image-base', parent=FAKE_IMAGE, status=utils.Status.MATCHED) FAKE_IMAGE_CHILD_UNMATCHED = Image( 'image-child-unmatched', 'image-child-unmatched:latest', '/fake/path3', parent_name='image-base', parent=FAKE_IMAGE, status=utils.Status.UNMATCHED) FAKE_IMAGE_CHILD_ERROR = Image( 'image-child-error', 'image-child-error:latest', '/fake/path4', parent_name='image-base', parent=FAKE_IMAGE, status=utils.Status.ERROR) FAKE_IMAGE_CHILD_BUILT = Image( 'image-child-built', 'image-child-built:latest', '/fake/path5', parent_name='image-base', parent=FAKE_IMAGE, status=utils.Status.BUILT) FAKE_IMAGE_GRANDCHILD = Image( 'image-grandchild', 'image-grandchild:latest', '/fake/path6', parent_name='image-child', parent=FAKE_IMAGE_CHILD, status=utils.Status.MATCHED) engine_client = "docker.DockerClient" if os.getenv("TEST_ENGINE") == "podman": engine_client = "podman.PodmanClient" class TasksTest(base.TestCase): def setUp(self): super(TasksTest, self).setUp() self.image = FAKE_IMAGE.copy() # NOTE(jeffrey4l): use a real, temporary dir self.image.path = self.useFixture(fixtures.TempDir()).path self.imageChild = FAKE_IMAGE_CHILD.copy() # NOTE(mandre) we want the local copy of FAKE_IMAGE as the parent self.imageChild.parent = self.image self.imageChild.path = self.useFixture(fixtures.TempDir()).path if "podman" in engine_client: self.conf.set_override('engine', 'podman') self.build_kwargs["dockerfile"] = self.image.path + '/Dockerfile' # NOTE(kevko): Squash implementation is different in podman # # - Check kolla/image/tasks.py for podman # - Check https://github.com/containers/buildah/issues/1234 self.build_kwargs["squash"] = False else: self.build_kwargs = {} # Existing tests exercise the docker-py SDK path; disable BuildKit so # they are not routed through _build_buildkit. self.conf.set_override('buildkit', False) # Mock build-date self.build_kwargs['labels'] = {'build-date': mock.ANY} @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_push_image(self, mock_client): self.engine_client = mock_client mock_client().api._auth_configs = {} pusher = tasks.PushTask(self.conf, self.image) pusher.run() mock_client().images.push.assert_called_once_with( self.image.canonical_name, decode=True, stream=True, auth_config={}) self.assertTrue(pusher.success) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_push_image_failure(self, mock_client): """failure on connecting Docker API""" self.engine_client = mock_client mock_client().api._auth_configs = {} mock_client().images.push.side_effect = Exception pusher = tasks.PushTask(self.conf, self.image) pusher.run() mock_client().images.push.assert_called_once_with( self.image.canonical_name, decode=True, stream=True, auth_config={}) self.assertFalse(pusher.success) self.assertEqual(utils.Status.PUSH_ERROR, self.image.status) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_push_image_failure_retry(self, mock_client): """failure on connecting Docker API, success on retry""" self.engine_client = mock_client mock_client().api._auth_configs = {} mock_client().images.push.side_effect = [Exception, []] pusher = tasks.PushTask(self.conf, self.image) pusher.run() mock_client().images.push.assert_called_once_with( self.image.canonical_name, decode=True, stream=True, auth_config={}) self.assertFalse(pusher.success) self.assertEqual(utils.Status.PUSH_ERROR, self.image.status) # Try again, this time without exception. pusher.reset() pusher.run() self.assertEqual(2, mock_client().images.push.call_count) self.assertTrue(pusher.success) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_push_image_failure_error(self, mock_client): """Docker connected, failure to push""" self.engine_client = mock_client mock_client().api._auth_configs = {} mock_client().images.push.return_value = [{'errorDetail': {'message': 'mock push fail'}}] pusher = tasks.PushTask(self.conf, self.image) pusher.run() mock_client().images.push.assert_called_once_with( self.image.canonical_name, decode=True, stream=True, auth_config={}) self.assertFalse(pusher.success) self.assertEqual(utils.Status.PUSH_ERROR, self.image.status) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_push_image_failure_error_retry(self, mock_client): """Docker connected, failure to push, success on retry""" self.engine_client = mock_client mock_client().api._auth_configs = {} mock_client().images.push.return_value = [{'errorDetail': {'message': 'mock push fail'}}] pusher = tasks.PushTask(self.conf, self.image) pusher.run() mock_client().images.push.assert_called_once_with( self.image.canonical_name, decode=True, stream=True, auth_config={}) self.assertFalse(pusher.success) self.assertEqual(utils.Status.PUSH_ERROR, self.image.status) # Try again, this time without exception. mock_client().images.push.return_value = [{'stream': 'mock push passes'}] pusher.reset() pusher.run() self.assertEqual(2, mock_client().images.push.call_count) self.assertTrue(pusher.success) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_build_image(self, mock_client): self.engine_client = mock_client push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once_with( path=self.image.path, tag=self.image.canonical_name, network_mode='host', nocache=False, rm=True, pull=True, forcerm=True, platform=None, buildargs=None, **self.build_kwargs) self.assertTrue(builder.success) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_build_image_with_network_mode(self, mock_client): self.engine_client = mock_client push_queue = mock.Mock() self.conf.set_override('network_mode', 'bridge') builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once_with( path=self.image.path, tag=self.image.canonical_name, network_mode='bridge', nocache=False, rm=True, pull=True, forcerm=True, platform=None, buildargs=None, **self.build_kwargs) self.assertTrue(builder.success) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_build_image_with_build_arg(self, mock_client): self.engine_client = mock_client build_args = { 'HTTP_PROXY': 'http://localhost:8080', 'NO_PROXY': '127.0.0.1' } self.conf.set_override('build_args', build_args) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once_with( path=self.image.path, tag=self.image.canonical_name, network_mode='host', nocache=False, rm=True, pull=True, forcerm=True, platform=None, buildargs=build_args, **self.build_kwargs) self.assertTrue(builder.success) @mock.patch.dict(os.environ, {'http_proxy': 'http://FROM_ENV:8080'}, clear=True) @mock.patch(engine_client) def test_build_arg_from_env(self, mock_client): push_queue = mock.Mock() self.engine_client = mock_client build_args = { 'http_proxy': 'http://FROM_ENV:8080', } builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once_with( path=self.image.path, tag=self.image.canonical_name, network_mode='host', nocache=False, rm=True, pull=True, forcerm=True, platform=None, buildargs=build_args, **self.build_kwargs) self.assertTrue(builder.success) @mock.patch.dict(os.environ, {'http_proxy': 'http://FROM_ENV:8080'}, clear=True) @mock.patch(engine_client) def test_build_arg_precedence(self, mock_client): self.engine_client = mock_client build_args = { 'http_proxy': 'http://localhost:8080', } self.conf.set_override('build_args', build_args) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once_with( path=self.image.path, tag=self.image.canonical_name, network_mode='host', nocache=False, rm=True, pull=True, forcerm=True, platform=None, buildargs=build_args, **self.build_kwargs) self.assertTrue(builder.success) @mock.patch(engine_client) @mock.patch('requests.get') def test_requests_get_timeout(self, mock_get, mock_client): self.engine_client = mock_client self.image.source = { 'source': 'http://fake/source', 'type': 'url', 'name': 'fake-image-base', 'enabled': True } push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) mock_get.side_effect = requests.exceptions.Timeout get_result = builder.process_source(self.image, self.image.source) self.assertIsNone(get_result) self.assertEqual(self.image.status, utils.Status.ERROR) mock_get.assert_called_once_with(self.image.source['source'], timeout=120) self.assertFalse(builder.success) @mock.patch('os.utime') @mock.patch('shutil.copyfile') @mock.patch('shutil.rmtree') @mock.patch(engine_client) @mock.patch('requests.get') def test_process_source(self, mock_get, mock_client, mock_rmtree, mock_copyfile, mock_utime): for source in [{'source': 'http://fake/source1', 'type': 'url', 'name': 'fake-image-base1', 'reference': 'http://fake/reference1', 'enabled': True}, {'source': 'http://fake/source2', 'type': 'git', 'name': 'fake-image-base2', 'reference': 'http://fake/reference2', 'enabled': True}, {'source': 'http://fake/source3', 'type': 'local', 'name': 'fake-image-base3', 'reference': 'http://fake/reference3', 'enabled': True}, {'source': 'http://fake/source4', 'type': None, 'name': 'fake-image-base4', 'reference': 'http://fake/reference4', 'enabled': True}, {'source': 'http://fake/source${version}', 'type': 'url', 'name': 'fake-image-base5', 'version': '5', 'enabled': True}, {'source': 'http://fake/source${debian_arch}', 'type': 'url', 'name': 'fake-image-base6', 'enabled': True}]: self.image.source = source push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) get_result = builder.process_source(self.image, self.image.source) self.assertEqual(self.image.status, utils.Status.ERROR) self.assertFalse(builder.success) if source['type'] != 'local': self.assertIsNone(get_result) else: self.assertIsNotNone(get_result) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_local_directory(self, mock_client): tmpdir = tempfile.mkdtemp() file_name = 'test.txt' file_path = os.path.join(tmpdir, file_name) saved_umask = os.umask(0o077) try: with open(file_path, 'w') as f: f.write('Hello') self.dc = mock_client self.image.plugins = [{ 'name': 'fake-image-base-plugin-test', 'type': 'local', 'enabled': True, 'source': tmpdir} ] push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() except IOError: print('IOError') else: os.remove(file_path) finally: os.umask(saved_umask) os.rmdir(tmpdir) self.assertTrue(builder.success) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_malicious_tar(self, mock_client): tmpdir = tempfile.mkdtemp() file_name = 'test.txt' archive_name = 'my_archive.tar' file_path = os.path.join(tmpdir, file_name) archive_path = os.path.join(tmpdir, archive_name) # Ensure the file is read/write by the creator only saved_umask = os.umask(0o077) try: with open(file_path, 'w') as f: f.write('Hello') with tarfile.open(archive_path, 'w') as tar: tar.add(file_path, arcname='../test.txt') self.dc = mock_client self.image.plugins = [{ 'name': 'fake-image-base-plugin-test', 'type': 'local', 'enabled': True, 'source': archive_path} ] push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() self.assertFalse(builder.success) except IOError: print('IOError') else: os.remove(file_path) os.remove(archive_path) finally: os.umask(saved_umask) os.rmdir(tmpdir) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_malicious_tar_gz(self, mock_client): tmpdir = tempfile.mkdtemp() file_name = 'test.txt' archive_name = 'my_archive.tar.gz' file_path = os.path.join(tmpdir, file_name) archive_path = os.path.join(tmpdir, archive_name) # Ensure the file is read/write by the creator only saved_umask = os.umask(0o077) try: with open(file_path, 'w') as f: f.write('Hello') with tarfile.open(archive_path, 'w:gz') as tar: tar.add(file_path, arcname='../test.txt') self.dc = mock_client self.image.plugins = [{ 'name': 'fake-image-base-plugin-test', 'type': 'local', 'enabled': True, 'source': archive_path} ] push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() self.assertFalse(builder.success) except IOError: print('IOError') else: os.remove(file_path) os.remove(archive_path) finally: os.umask(saved_umask) os.rmdir(tmpdir) @mock.patch('os.path.exists') @mock.patch('os.utime') @mock.patch('shutil.rmtree') def test_process_git_source_existing_dir(self, mock_rmtree, mock_utime, mock_path_exists): source = {'source': 'http://fake/source1', 'type': 'git', 'name': 'fake-image1', 'reference': 'fake/reference1', 'enabled': True} self.image.source = source self.image.path = "fake_image_path" mock_path_exists.return_value = True push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) get_result = builder.process_source(self.image, self.image.source) mock_rmtree.assert_called_with( "fake_image_path/fake-image1-archive-fake-reference1") self.assertEqual(self.image.status, utils.Status.ERROR) self.assertFalse(builder.success) self.assertIsNone(get_result) @mock.patch(engine_client) def test_followups_docker_image(self, mock_client): self.imageChild.source = { 'source': 'http://fake/source', 'type': 'url', 'name': 'fake-image-base' } self.imageChild.children.append(FAKE_IMAGE_CHILD_UNMATCHED) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.imageChild, push_queue) builder.success = True self.conf.push = FAKE_IMAGE_CHILD_UNMATCHED get_result = builder.followups self.assertEqual(1, len(get_result)) @mock.patch(engine_client) def test_followups_buildkit_skips_push(self, mock_client): """PushTask must not be added when buildkit is active (docker).""" self.conf.set_override('buildkit', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.imageChild, push_queue) builder.success = True self.conf.push = True followups = builder.followups push_tasks = [f for f in followups if isinstance(f, tasks.PushIntoQueueTask)] self.assertEqual(0, len(push_tasks)) @mock.patch(engine_client) def test_followups_buildkit_true_podman_includes_push(self, mock_client): """PushTask must be added for Podman even when buildkit=True.""" self.conf.set_override('buildkit', True) self.conf.set_override('engine', 'podman') push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.imageChild, push_queue) builder.success = True self.conf.push = True followups = builder.followups push_tasks = [f for f in followups if isinstance(f, tasks.PushIntoQueueTask)] self.assertEqual(1, len(push_tasks)) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_no_platform_no_push(self, mock_run_cmd): """No --platform and no push: command uses --load.""" self.conf.set_override('buildkit', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=True, buildargs=None) mock_run_cmd.assert_called_once() cmd = mock_run_cmd.call_args[0][0] self.assertIn('--load', cmd) self.assertNotIn('--push', cmd) self.assertNotIn('--platform', cmd) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_no_platform_with_push(self, mock_run_cmd): """No platform + push=True: buildx pushes directly.""" self.conf.set_override('buildkit', True) self.conf.set_override('push', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) mock_run_cmd.assert_called_once() cmd = mock_run_cmd.call_args[0][0] self.assertIn('--push', cmd) self.assertNotIn('--load', cmd) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_single_platform_no_push(self, mock_run_cmd): """Single platform without push: --load, no push_tag, no extra push.""" self.conf.set_override('buildkit', True) self.conf.set_override('platform', 'linux/amd64') push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) mock_run_cmd.assert_called_once() cmd = mock_run_cmd.call_args[0][0] self.assertIn('--platform', cmd) self.assertIn('linux/amd64', cmd) self.assertIn('--load', cmd) self.assertNotIn('--push', cmd) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_single_platform_with_push(self, mock_run_cmd): """Single platform + push: --load then push canonical tag.""" self.conf.set_override('buildkit', True) self.conf.set_override('platform', 'linux/amd64') self.conf.set_override('push', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) self.assertEqual(2, mock_run_cmd.call_count) build_cmd = mock_run_cmd.call_args_list[0][0][0] push_cmd = mock_run_cmd.call_args_list[1][0][0] self.assertIn('--load', build_cmd) self.assertNotIn('--push', build_cmd) self.assertNotIn('image-base:latest-amd64', build_cmd) self.assertEqual( ['docker', 'push', self.image.canonical_name], push_cmd) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_multi_platform_with_push(self, mock_run_cmd): """Multi-platform + push: single --push invocation, no separate push""" self.conf.set_override('buildkit', True) self.conf.set_override('platform', 'linux/amd64,linux/arm64') self.conf.set_override('push', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) mock_run_cmd.assert_called_once() cmd = mock_run_cmd.call_args[0][0] self.assertIn('--push', cmd) self.assertNotIn('--load', cmd) self.assertIn('linux/amd64,linux/arm64', cmd) self.assertEqual(utils.Status.BUILT, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_multi_platform_no_push(self, mock_run_cmd): """Multi-platform without push: fails fast (--load unsupported).""" self.conf.set_override('buildkit', True) self.conf.set_override('platform', 'linux/amd64,linux/arm64') push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) mock_run_cmd.assert_not_called() self.assertEqual(utils.Status.ERROR, self.image.status) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_with_named_builder(self, mock_run_cmd): """--buildkit-builder is forwarded as --builder to the command.""" self.conf.set_override('buildkit', True) self.conf.set_override('buildkit_builder', 'mybuilder') push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) cmd = mock_run_cmd.call_args[0][0] self.assertIn('--builder', cmd) idx = cmd.index('--builder') self.assertEqual('mybuilder', cmd[idx + 1]) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_no_cache(self, mock_run_cmd): """cache=False adds --no-cache to the command.""" self.conf.set_override('buildkit', True) self.conf.set_override('cache', False) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) cmd = mock_run_cmd.call_args[0][0] self.assertIn('--no-cache', cmd) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_with_buildargs(self, mock_run_cmd): """Build args are forwarded as --build-arg KEY=VALUE pairs.""" self.conf.set_override('buildkit', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() buildargs = {'HTTP_PROXY': 'http://proxy:8080', 'NO_PROXY': '127.0.0.1'} builder._build_buildkit(self.image, pull=False, buildargs=buildargs) cmd = mock_run_cmd.call_args[0][0] self.assertIn('--build-arg', cmd) self.assertIn('HTTP_PROXY=http://proxy:8080', cmd) self.assertIn('NO_PROXY=127.0.0.1', cmd) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_with_labels(self, mock_run_cmd): """Image labels are forwarded as --label key=value pairs.""" self.conf.set_override('buildkit', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() self.image.labels = {'maintainer': 'kolla', 'version': '1.0'} builder._build_buildkit(self.image, pull=False, buildargs=None) cmd = mock_run_cmd.call_args[0][0] self.assertIn('--label', cmd) self.assertIn('maintainer=kolla', cmd) self.assertIn('version=1.0', cmd) @mock.patch.object(tasks.BuildTask, '_run_cmd') def test_build_buildkit_error_sets_status(self, mock_run_cmd): """An exception from _run_cmd marks the image as ERROR.""" self.conf.set_override('buildkit', True) mock_run_cmd.side_effect = Exception('build failed') push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) self.image.start = datetime.datetime.now() builder._build_buildkit(self.image, pull=False, buildargs=None) self.assertEqual(utils.Status.ERROR, self.image.status) @mock.patch('subprocess.Popen') def test_run_cmd_success(self, mock_popen): """Zero exit code does not raise.""" mock_proc = mock.MagicMock() mock_proc.__enter__.return_value = mock_proc mock_proc.stdout = [] mock_proc.returncode = 0 mock_popen.return_value = mock_proc push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder._run_cmd(['docker', 'buildx', 'build', self.image.path]) @mock.patch('subprocess.Popen') def test_run_cmd_failure_raises(self, mock_popen): """Non-zero exit code raises CalledProcessError.""" import subprocess as sp # nosec mock_proc = mock.MagicMock() mock_proc.__enter__.return_value = mock_proc mock_proc.stdout = [] mock_proc.returncode = 1 mock_popen.return_value = mock_proc push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) cmd = ['docker', 'buildx', 'build', self.image.path] self.assertRaises(sp.CalledProcessError, builder._run_cmd, cmd) @mock.patch(engine_client) def test_build_image_buildkit_dispatches(self, mock_client): """When buildkit=True and engine=docker, calls _build_buildkit.""" self.conf.set_override('buildkit', True) push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) with mock.patch.object(builder, '_build_buildkit') as mock_bk: builder.run() mock_bk.assert_called_once() @mock.patch(engine_client) def test_build_image_no_buildkit_uses_sdk(self, mock_client): """When buildkit=False, builder uses docker-py SDK (images.build).""" push_queue = mock.Mock() builder = tasks.BuildTask(self.conf, self.image, push_queue) builder.run() mock_client().images.build.assert_called_once() class KollaWorkerTest(base.TestCase): config_file = 'default.conf' def setUp(self): super(KollaWorkerTest, self).setUp() image = FAKE_IMAGE.copy() image.status = None image_child = FAKE_IMAGE_CHILD.copy() image_child.status = None image_unmatched = FAKE_IMAGE_CHILD_UNMATCHED.copy() image_error = FAKE_IMAGE_CHILD_ERROR.copy() image_built = FAKE_IMAGE_CHILD_BUILT.copy() # NOTE(mandre) we want the local copy of FAKE_IMAGE as the parent for i in [image_child, image_unmatched, image_error, image_built]: i.parent = image self.images = [image, image_child, image_unmatched, image_error, image_built] patcher = mock.patch(engine_client) self.addCleanup(patcher.stop) self.mock_client = patcher.start() def test_supported_base_distro(self): build_base = ['centos', 'debian', 'ubuntu'] for base_distro in build_base: self.conf.set_override('base', base_distro) # should no exception raised build.KollaWorker(self.conf) def test_build_image_list_adds_plugins(self): kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() kolla.create_dockerfiles() kolla.build_image_list() expected_plugin = { 'name': 'neutron-server-plugin-networking-arista', 'reference': 'master', 'source': 'https://opendev.org/x/networking-arista', 'type': 'git', 'enabled': True, 'sha256': None } found = False for image in kolla.images: if image.name == 'neutron-server': for plugin in image.plugins: if plugin == expected_plugin: found = True break break if not found: self.fail('Can not find the expected neutron arista plugin') def test_build_image_list_skips_disabled_plugins(self): self.conf.set_override('enabled', False, 'neutron-base-plugin-networking-baremetal') kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() kolla.create_dockerfiles() kolla.build_image_list() disabled_plugin = 'neutron-base-plugin-networking-baremetal' found = False for image in kolla.images: if image.name == 'neutron-server': for plugin in image.plugins: if plugin == disabled_plugin: found = True break break if found: self.fail('Found disabled neutron networking-baremetal plugin') def test_build_image_list_plugin_parsing(self): """Ensure regex used to parse plugins adds them to the correct image""" kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() kolla.create_dockerfiles() kolla.build_image_list() for image in kolla.images: if image.name == 'base': self.assertEqual(len(image.plugins), 0, 'base image should not have any plugins ' 'registered') break else: self.fail('Expected to find the base image in this test') def test_set_time(self): kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.set_time() @mock.patch.dict(os.environ, {'http_proxy': 'http://test-proxy:8080'}, clear=False) @mock.patch(engine_client) def test_create_dockerfiles_env_http_proxy(self, mock_client): tmpdir = self.useFixture(fixtures.TempDir()).path base_dir = os.path.join(tmpdir, 'base') os.makedirs(base_dir) with open(os.path.join(base_dir, 'Dockerfile.j2'), 'w') as f: f.write( '{% if env.http_proxy %}' 'ARG http_proxy={{ env.http_proxy }}' '{% endif %}\n' ) with mock.patch.object(build.KollaWorker, '_get_images_dir', return_value=tmpdir): kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() kolla.create_dockerfiles() dockerfile = os.path.join(kolla.working_dir, 'base', 'Dockerfile') with open(dockerfile) as f: content = f.read() self.assertIn('http://test-proxy:8080', content) @mock.patch(engine_client) def test_create_dockerfiles_ssti_blocked(self, mock_client): tmpdir = self.useFixture(fixtures.TempDir()).path base_dir = os.path.join(tmpdir, 'base') os.makedirs(base_dir) with open(os.path.join(base_dir, 'Dockerfile.j2'), 'w') as f: f.write( "{{ self.__init__.__globals__['__builtins__']" "['__import__']('os').popen('id').read() }}\n" ) with mock.patch.object(build.KollaWorker, '_get_images_dir', return_value=tmpdir): kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() self.assertRaises(jinja2.exceptions.SecurityError, kolla.create_dockerfiles) @mock.patch(engine_client) def test_create_dockerfiles_template_override_ssti_blocked( self, mock_client): tmpdir = self.useFixture(fixtures.TempDir()).path base_dir = os.path.join(tmpdir, 'base') os.makedirs(base_dir) with open(os.path.join(base_dir, 'Dockerfile.j2'), 'w') as f: f.write('FROM {{ base_distro }}:{{ base_distro_tag }}\n') override_file = os.path.join( self.useFixture(fixtures.TempDir()).path, 'template_override.j2') with open(override_file, 'w') as f: f.write( "{{ self.__init__.__globals__['__builtins__']" "['__import__']('os').popen('id').read() }}\n" ) self.conf.set_override('template_override', [override_file]) with mock.patch.object(build.KollaWorker, '_get_images_dir', return_value=tmpdir): kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() kolla.find_dockerfiles() self.assertRaises(jinja2.exceptions.SecurityError, kolla.create_dockerfiles) def _get_matched_images(self, images): return [image for image in images if image.status == utils.Status.MATCHED] def test_skip_parents(self): self.conf.set_override('skip_parents', True) kolla = build.KollaWorker(self.conf) kolla.images = self.images[:2] for i in kolla.images: i.status = utils.Status.UNPROCESSED if i.parent: i.parent.children.append(i) kolla.filter_images() self.assertEqual(utils.Status.MATCHED, kolla.images[1].status) self.assertEqual(utils.Status.SKIPPED, kolla.images[1].parent.status) def test_skip_parents_regex(self): self.conf.set_override('regex', ['image-child']) self.conf.set_override('skip_parents', True) kolla = build.KollaWorker(self.conf) kolla.images = self.images[:2] for i in kolla.images: i.status = utils.Status.UNPROCESSED if i.parent: i.parent.children.append(i) kolla.filter_images() self.assertEqual(utils.Status.MATCHED, kolla.images[1].status) self.assertEqual(utils.Status.SKIPPED, kolla.images[1].parent.status) def test_skip_parents_match_grandchildren(self): self.conf.set_override('skip_parents', True) kolla = build.KollaWorker(self.conf) image_grandchild = FAKE_IMAGE_GRANDCHILD.copy() image_grandchild.parent = self.images[1] self.images[1].children.append(image_grandchild) kolla.images = self.images[:2] + [image_grandchild] for i in kolla.images: i.status = utils.Status.UNPROCESSED if i.parent: i.parent.children.append(i) kolla.filter_images() self.assertEqual(utils.Status.MATCHED, kolla.images[2].status) self.assertEqual(utils.Status.SKIPPED, kolla.images[2].parent.status) self.assertEqual(utils.Status.SKIPPED, kolla.images[1].parent.status) def test_skip_parents_match_grandchildren_regex(self): self.conf.set_override('regex', ['image-grandchild']) self.conf.set_override('skip_parents', True) kolla = build.KollaWorker(self.conf) image_grandchild = FAKE_IMAGE_GRANDCHILD.copy() image_grandchild.parent = self.images[1] self.images[1].children.append(image_grandchild) kolla.images = self.images[:2] + [image_grandchild] for i in kolla.images: i.status = utils.Status.UNPROCESSED if i.parent: i.parent.children.append(i) kolla.filter_images() self.assertEqual(utils.Status.MATCHED, kolla.images[2].status) self.assertEqual(utils.Status.SKIPPED, kolla.images[2].parent.status) self.assertEqual(utils.Status.SKIPPED, kolla.images[1].parent.status) @mock.patch.object(Image, 'in_engine_cache') def test_skip_existing(self, mock_in_cache): mock_in_cache.side_effect = [True, False] self.conf.set_override('skip_existing', True) kolla = build.KollaWorker(self.conf) kolla.images = self.images[:2] for i in kolla.images: i.status = utils.Status.UNPROCESSED kolla.filter_images() self.assertEqual(utils.Status.SKIPPED, kolla.images[0].status) self.assertEqual(utils.Status.MATCHED, kolla.images[1].status) def test_without_profile(self): kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.filter_images() self.assertEqual(5, len(self._get_matched_images(kolla.images))) def test_build_rpm_setup(self): """checking the length of list of docker commands""" self.conf.set_override('rpm_setup_config', ["a.rpm", "b.repo"]) kolla = build.KollaWorker(self.conf) self.assertEqual(2, len(kolla.rpm_setup)) def test_build_distro_package_manager(self): """check distro_package_manager conf value is taken""" self.conf.set_override('distro_package_manager', 'foo') kolla = build.KollaWorker(self.conf) self.assertEqual('foo', kolla.distro_package_manager) def test_build_distro_package_manager_debian(self): """check distro_package_manager apt for debian""" self.conf.set_override('base', 'debian') kolla = build.KollaWorker(self.conf) self.assertEqual('apt', kolla.distro_package_manager) def test_build_distro_package_manager_ubuntu(self): """check distro_package_manager apt for ubuntu""" self.conf.set_override('base', 'ubuntu') kolla = build.KollaWorker(self.conf) self.assertEqual('apt', kolla.distro_package_manager) def test_base_package_type(self): """check base_package_type conf value is taken""" self.conf.set_override('base_package_type', 'pip') kolla = build.KollaWorker(self.conf) self.assertEqual('pip', kolla.base_package_type) def test_base_package_type_debian(self): """check base_package_type deb for debian""" self.conf.set_override('base', 'debian') kolla = build.KollaWorker(self.conf) self.assertEqual('deb', kolla.base_package_type) def test_pre_defined_exist_profile(self): # default profile include the fake image: image-base self.conf.set_override('profile', ['default']) kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.filter_images() self.assertEqual(1, len(self._get_matched_images(kolla.images))) def test_pre_defined_exist_profile_not_include(self): # infra profile do not include the fake image: image-base self.conf.set_override('profile', ['infra']) kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.filter_images() self.assertEqual(0, len(self._get_matched_images(kolla.images))) def test_pre_defined_not_exist_profile(self): # NOTE(jeffrey4l): not exist profile will raise ValueError self.conf.set_override('profile', ['not_exist']) kolla = build.KollaWorker(self.conf) kolla.images = self.images self.assertRaises(ValueError, kolla.filter_images) @mock.patch('json.dump') def test_list_dependencies(self, dump_mock): self.conf.set_override('profile', ['all']) kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.filter_images() kolla.list_dependencies() dump_mock.assert_called_once_with(mock.ANY, sys.stdout, indent=2) def test_summary(self): kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.image_statuses_good['good'] = utils.Status.BUILT kolla.image_statuses_bad['bad'] = utils.Status.ERROR kolla.image_statuses_allowed_to_fail['bad2'] = utils.Status.ERROR kolla.image_statuses_unmatched['unmatched'] = utils.Status.UNMATCHED results = kolla.summary() self.assertEqual('error', results['failed'][0]['status']) # bad self.assertEqual('error', results['failed'][1]['status']) # bad2 @mock.patch('json.dump') def test_summary_json_format(self, dump_mock): self.conf.set_override('format', 'json') kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.image_statuses_good['good'] = build.Status.BUILT kolla.image_statuses_bad['bad'] = build.Status.ERROR kolla.image_statuses_allowed_to_fail['bad2'] = build.Status.ERROR kolla.image_statuses_unmatched['unmatched'] = build.Status.UNMATCHED results = kolla.summary() dump_mock.assert_called_once_with(results, sys.stdout) @mock.patch('json.dump') def test_summary_json_format_file(self, dump_mock): tmpdir = tempfile.mkdtemp() file_path = os.path.join(tmpdir, 'summary.json') try: self.conf.set_override('format', 'json') self.conf.set_override('summary_json_file', file_path) kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.image_statuses_good['good'] = build.Status.BUILT kolla.image_statuses_bad['bad'] = build.Status.ERROR kolla.image_statuses_allowed_to_fail['bad2'] = build.Status.ERROR kolla.image_statuses_unmatched['unmatched'] = ( build.Status.UNMATCHED) results = kolla.summary() dump_mock.assert_called_once_with(results, mock.ANY, indent=4) self.assertEqual(dump_mock.call_args[0][1].name, file_path) finally: os.remove(file_path) os.rmdir(tmpdir) @mock.patch('builtins.open') def test_summary_json_format_file_error(self, open_mock): open_mock.side_effect = OSError self.conf.set_override('format', 'json') self.conf.set_override('summary_json_file', 'fake-file') kolla = build.KollaWorker(self.conf) kolla.images = self.images kolla.image_statuses_good['good'] = build.Status.BUILT kolla.image_statuses_bad['bad'] = build.Status.ERROR kolla.image_statuses_allowed_to_fail['bad2'] = build.Status.ERROR kolla.image_statuses_unmatched['unmatched'] = ( build.Status.UNMATCHED) self.assertRaises(SystemExit, kolla.summary) @mock.patch('shutil.copytree') def test_work_dir(self, copytree_mock): self.conf.set_override('work_dir', 'tmp/foo') kolla = build.KollaWorker(self.conf) kolla.setup_working_dir() self.assertEqual('tmp/foo/docker', kolla.working_dir) class BuildKitCheckTest(base.TestCase): @mock.patch('subprocess.check_output') def test_check_docker_buildx_success(self, mock_check_output): mock_check_output.return_value = b'github.com/docker/buildx v0.21.0' common_utils.check_docker_buildx() mock_check_output.assert_called_once_with( ['docker', 'buildx', 'version'], stderr=mock.ANY) @mock.patch('subprocess.check_output') def test_check_docker_buildx_not_installed(self, mock_check_output): import subprocess as sp # nosec mock_check_output.side_effect = sp.CalledProcessError(1, 'docker') self.assertRaises(sp.CalledProcessError, common_utils.check_docker_buildx) @mock.patch('subprocess.check_output') def test_check_docker_buildx_docker_missing(self, mock_check_output): mock_check_output.side_effect = OSError(2, 'No such file or directory') self.assertRaises(OSError, common_utils.check_docker_buildx) class MainTest(base.TestCase): def setUp(self): super(MainTest, self).setUp() self.conf.set_override('buildkit', False) # Mock subprocess so that _build_buildkit and check_docker_buildx # succeed when the full build pipeline is exercised (buildkit=True is # the default for a fresh conf). mock_proc = mock.MagicMock() mock_proc.__enter__.return_value = mock_proc mock_proc.stdout = [] mock_proc.returncode = 0 popen_patcher = mock.patch('subprocess.Popen', return_value=mock_proc) popen_patcher.start() self.addCleanup(popen_patcher.stop) check_output_patcher = mock.patch( 'subprocess.check_output', return_value=b'github.com/docker/buildx v0.21.0') check_output_patcher.start() self.addCleanup(check_output_patcher.stop) @mock.patch.object(build, 'run_build') def test_images_built(self, mock_run_build): image_statuses = ({}, {'img': 'built'}, {}, {}, {}, {}) mock_run_build.return_value = image_statuses result = build_cmd.main() self.assertEqual(0, result) @mock.patch.object(build, 'run_build') def test_images_unmatched(self, mock_run_build): image_statuses = ({}, {}, {'img': 'unmatched'}, {}, {}, {}) mock_run_build.return_value = image_statuses result = build_cmd.main() self.assertEqual(0, result) @mock.patch.object(build, 'run_build') def test_no_images_built(self, mock_run_build): mock_run_build.return_value = None result = build_cmd.main() self.assertEqual(0, result) @mock.patch.object(build, 'run_build') def test_bad_images(self, mock_run_build): image_statuses = ({'img': 'error'}, {}, {}, {}, {}, {}) mock_run_build.return_value = image_statuses result = build_cmd.main() self.assertEqual(1, result) @mock.patch('sys.argv') @mock.patch(engine_client) def test_run_build(self, mock_client, mock_sys): result = build.run_build() self.assertTrue(result) @mock.patch('sys.argv') @mock.patch('kolla.common.utils.check_docker_buildx') @mock.patch(engine_client) def test_run_build_exits_when_buildx_missing( self, mock_client, mock_check_buildx, mock_sys): import subprocess as sp # nosec mock_check_buildx.side_effect = sp.CalledProcessError(1, 'docker') self.assertRaises(SystemExit, build.run_build) @mock.patch.object(build, 'run_build') def test_skipped_images(self, mock_run_build): image_statuses = ({}, {}, {}, {'img': 'skipped'}, {}, {}) mock_run_build.return_value = image_statuses result = build_cmd.main() self.assertEqual(0, result) @mock.patch.object(build, 'run_build') def test_unbuildable_images(self, mock_run_build): image_statuses = ({}, {}, {}, {}, {'img': 'unbuildable'}, {}) mock_run_build.return_value = image_statuses result = build_cmd.main() self.assertEqual(0, result) @mock.patch('requests.get') @mock.patch(engine_client) def test_build_etcd_cross_arch(self, mock_docker, mock_requests_get): """Test building etcd with --base-arch aarch64. This ensures we fetch the arm64 tarball instead of amd64. """ mock_response = mock.Mock() # Force error (404) so we can inspect the requested URL mock_response.status_code = 404 mock_requests_get.return_value = mock_response argv = ['kolla-build', '--base-arch', 'aarch64', 'etcd'] with mock.patch.object(sys, 'argv', argv): build_cmd.main() self.assertTrue( mock_requests_get.called, "requests.get was never called; 'etcd' may not have been matched." ) found_url = None for call_args in mock_requests_get.call_args_list: url = call_args[0][0] # Look for a string like 'etcd-vX.Y.Z-linux-...' if 'etcd-v' in url and 'linux-' in url: found_url = url break self.assertIsNotNone( found_url, "No GET request found for the etcd tarball " "(expected 'etcd-v' and 'linux-' in the URL)." ) self.assertIn( 'arm64', found_url, "Expected 'arm64' in etcd URL (aarch64), got: %s" % found_url ) self.assertNotIn( 'amd64', found_url, "Should not be 'amd64' for aarch64 build, got: %s" % found_url )