# 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 io 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_local_directory_archive_reproducible(self, mock_client): src_dir = self.useFixture(fixtures.TempDir()).path sub_dir = os.path.join(src_dir, 'sub') os.mkdir(sub_dir) file_path = os.path.join(sub_dir, 'test.txt') with open(file_path, 'w') as f: f.write('Hello') source = {'name': 'fake-image-base', 'type': 'local', 'enabled': True, 'source': src_dir} push_queue = mock.Mock() archives = [] for mtime in (1000000000, 2000000000): os.utime(file_path, (mtime, mtime)) os.utime(sub_dir, (mtime, mtime)) os.utime(src_dir, (mtime, mtime)) image = FAKE_IMAGE.copy() image.path = self.useFixture(fixtures.TempDir()).path builder = tasks.BuildTask(self.conf, image, push_queue) dest_archive = builder.process_source(image, source) self.assertIsNotNone(dest_archive) with open(dest_archive, 'rb') as f: archives.append(f.read()) self.assertEqual(archives[0], archives[1]) with tarfile.open(fileobj=io.BytesIO(archives[0])) as tar: for member in tar.getmembers(): self.assertEqual(0, member.mtime) self.assertEqual(0, member.uid) self.assertEqual(0, member.gid) self.assertEqual('root', member.uname) self.assertEqual('root', member.gname) @mock.patch.dict(os.environ, clear=True) @mock.patch(engine_client) def test_empty_plugins_archive_reproducible(self, mock_client): self.dc = mock_client push_queue = mock.Mock() archives = [] for _ in range(2): image = FAKE_IMAGE.copy() image.path = self.useFixture(fixtures.TempDir()).path builder = tasks.BuildTask(self.conf, image, push_queue) builder.run() self.assertTrue(builder.success) with open(os.path.join(image.path, 'plugins-archive'), 'rb') as f: archives.append(f.read()) self.assertEqual(archives[0], archives[1]) with tarfile.open(fileobj=io.BytesIO(archives[0])) as tar: members = tar.getmembers() self.assertEqual(['plugins'], [m.name for m in members]) self.assertEqual(0, members[0].mtime) @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 = ['debian', 'rocky', '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) @mock.patch(engine_client) def test_create_dockerfiles_pip_args_lowercase_from(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') self.conf.set_override('pip_index_url', 'https://example.org/simple') 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('ARG PIP_INDEX_URL', content) @mock.patch(engine_client) def test_build_image_list_parent_name_lowercase_from(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') child_dir = os.path.join(tmpdir, 'child') os.makedirs(child_dir) with open(os.path.join(child_dir, 'Dockerfile.j2'), 'w') as f: f.write('from {{ namespace }}/{{ image_prefix }}base:{{ tag }}\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() kolla.build_image_list() expected_parent = '{}/{}base:{}'.format( kolla.namespace, kolla.image_prefix, kolla.tag) child = [i for i in kolla.images if i.name == 'child'][0] self.assertEqual(expected_parent, child.parent_name) 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 )