path: root/kolla/tests/test_build.py

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import 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
        )