pungi/tests/test_koji_wrapper.py

#!/usr/bin/env python2
# -*- coding: utf-8 -*-

import mock
import unittest

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from pungi.wrappers.kojiwrapper import KojiWrapper


class KojiWrapperTest(unittest.TestCase):

    def setUp(self):
        self.koji_profile = mock.Mock()
        with mock.patch('pungi.wrappers.kojiwrapper.koji') as koji:
            koji.get_profile_module = mock.Mock(
                return_value=mock.Mock(
                    pathinfo=mock.Mock(
                        work=mock.Mock(return_value='/koji'),
                        taskrelpath=mock.Mock(side_effect=lambda id: 'task/' + str(id)),
                        imagebuild=mock.Mock(side_effect=lambda id: '/koji/imagebuild/' + str(id)),
                    )
                )
            )
            self.koji_profile = koji.get_profile_module.return_value
            self.koji = KojiWrapper('koji')

    @mock.patch('pungi.wrappers.kojiwrapper.open')
    def test_get_image_build_cmd_without_required_data(self, mock_open):
        with self.assertRaises(AssertionError):
            self.koji.get_image_build_cmd(
                {
                    'name': 'test-name',
                },
                '/tmp/file'
            )

    @mock.patch('pungi.wrappers.kojiwrapper.open')
    def test_get_image_build_cmd_correct(self, mock_open):
        cmd = self.koji.get_image_build_cmd(
            {
                'name': 'test-name',
                'version': '1',
                'target': 'test-target',
                'install_tree': '/tmp/test/install_tree',
                'arches': 'x86_64',
                'format': 'docker,qcow2',
                'kickstart': 'test-kickstart',
                'ksurl': 'git://example.com/ks.git',
                'distro': 'test-distro',
            },
            '/tmp/file'
        )

        self.assertEqual(cmd[0], 'koji')
        self.assertEqual(cmd[1], 'image-build')
        self.assertItemsEqual(cmd[2:],
                              ['--config=/tmp/file', '--wait'])

        output = mock_open.return_value
        self.assertEqual(mock.call('[image-build]\n'), output.write.mock_calls[0])
        self.assertItemsEqual(output.write.mock_calls[1:],
                              [mock.call('name = test-name\n'),
                               mock.call('version = 1\n'),
                               mock.call('target = test-target\n'),
                               mock.call('install_tree = /tmp/test/install_tree\n'),
                               mock.call('arches = x86_64\n'),
                               mock.call('format = docker,qcow2\n'),
                               mock.call('kickstart = test-kickstart\n'),
                               mock.call('ksurl = git://example.com/ks.git\n'),
                               mock.call('distro = test-distro\n'),
                               mock.call('\n')])

    def test_get_image_paths(self):

        # The data for this tests is obtained from the actual Koji build. It
        # includes lots of fields that are not used, but for the sake of
        # completeness is fully preserved.

        getTaskChildren_data = {
            12387273: [
                {
                    'arch': 'i386',
                    'awaited': False,
                    'channel_id': 12,
                    'completion_time': '2016-01-03 05:34:08.374262',
                    'completion_ts': 1451799248.37426,
                    'create_time': '2016-01-03 05:15:20.311599',
                    'create_ts': 1451798120.3116,
                    'host_id': 158,
                    'id': 12387276,
                    'label': 'i386',
                    'method': 'createImage',
                    'owner': 131,
                    'parent': 12387273,
                    'priority': 19,
                    'request': [
                        'Fedora-Cloud-Base',
                        '23',
                        '20160103',
                        'i386',
                        {
                            'build_tag': 299,
                            'build_tag_name': 'f23-build',
                            'dest_tag': 294,
                            'dest_tag_name': 'f23-updates-candidate',
                            'id': 144,
                            'name': 'f23-candidate'
                        },
                        299,
                        {
                            'create_event': 14011966,
                            'create_ts': 1451761803.33528,
                            'creation_time': '2016-01-02 19:10:03.335283',
                            'id': 563977,
                            'state': 1
                        },
                        'http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/i386/os/',
                        {
                            'disk_size': '3',
                            'distro': 'Fedora-20',
                            'format': ['qcow2', 'raw-xz'],
                            'kickstart': 'work/cli-image/1451798116.800155.wYJWTVHw/fedora-cloud-base-2878aa0.ks',
                            'release': '20160103',
                            'repo': ['http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/$arch/os/',
                                     'http://infrastructure.fedoraproject.org/pub/fedora/linux/updates/23/$arch/'],
                            'scratch': True
                        }
                    ],
                    'start_time': '2016-01-03 05:15:29.828081',
                    'start_ts': 1451798129.82808,
                    'state': 2,
                    'waiting': None,
                    'weight': 2.0
                }, {
                    'arch': 'x86_64',
                    'awaited': False,
                    'channel_id': 12,
                    'completion_time': '2016-01-03 05:33:20.066366',
                    'completion_ts': 1451799200.06637,
                    'create_time': '2016-01-03 05:15:20.754201',
                    'create_ts': 1451798120.7542,
                    'host_id': 156,
                    'id': 12387277,
                    'label': 'x86_64',
                    'method': 'createImage',
                    'owner': 131,
                    'parent': 12387273,
                    'priority': 19,
                    'request': [
                        'Fedora-Cloud-Base',
                        '23',
                        '20160103',
                        'x86_64',
                        {
                            'build_tag': 299,
                            'build_tag_name': 'f23-build',
                            'dest_tag': 294,
                            'dest_tag_name': 'f23-updates-candidate',
                            'id': 144,
                            'name': 'f23-candidate'
                        },
                        299,
                        {
                            'create_event': 14011966,
                            'create_ts': 1451761803.33528,
                            'creation_time': '2016-01-02 19:10:03.335283',
                            'id': 563977,
                            'state': 1
                        },
                        'http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/x86_64/os/',
                        {
                            'disk_size': '3',
                            'distro': 'Fedora-20',
                            'format': ['qcow2', 'raw-xz'],
                            'kickstart': 'work/cli-image/1451798116.800155.wYJWTVHw/fedora-cloud-base-2878aa0.ks',
                            'release': '20160103',
                            'repo': ['http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/$arch/os/',
                                     'http://infrastructure.fedoraproject.org/pub/fedora/linux/updates/23/$arch/'],
                            'scratch': True
                        }
                    ],
                    'start_time': '2016-01-03 05:15:35.196043',
                    'start_ts': 1451798135.19604,
                    'state': 2,
                    'waiting': None,
                    'weight': 2.0
                }
            ]
        }

        getTaskResult_data = {
            12387276: {
                'arch': 'i386',
                'files': ['tdl-i386.xml',
                          'fedora-cloud-base-2878aa0.ks',
                          'koji-f23-build-12387276-base.ks',
                          'libvirt-qcow2-i386.xml',
                          'Fedora-Cloud-Base-23-20160103.i386.qcow2',
                          'libvirt-raw-xz-i386.xml',
                          'Fedora-Cloud-Base-23-20160103.i386.raw.xz'],
                'logs': ['oz-i386.log'],
                'name': 'Fedora-Cloud-Base',
                'release': '20160103',
                'rpmlist': [],
                'task_id': 12387276,
                'version': '23'
            },
            12387277: {
                'arch': 'x86_64',
                'files': ['tdl-x86_64.xml',
                          'fedora-cloud-base-2878aa0.ks',
                          'koji-f23-build-12387277-base.ks',
                          'libvirt-qcow2-x86_64.xml',
                          'Fedora-Cloud-Base-23-20160103.x86_64.qcow2',
                          'libvirt-raw-xz-x86_64.xml',
                          'Fedora-Cloud-Base-23-20160103.x86_64.raw.xz'],
                'logs': ['oz-x86_64.log'],
                'name': 'Fedora-Cloud-Base',
                'release': '20160103',
                'rpmlist': [],
                'task_id': 12387277,
                'version': '23'
            }

        }

        self.koji.koji_proxy = mock.Mock(
            getTaskChildren=mock.Mock(side_effect=lambda task_id, request: getTaskChildren_data.get(task_id)),
            getTaskResult=mock.Mock(side_effect=lambda task_id: getTaskResult_data.get(task_id))
        )
        result = self.koji.get_image_paths(12387273)
        self.assertItemsEqual(result.keys(), ['i386', 'x86_64'])
        self.maxDiff = None
        self.assertItemsEqual(result['i386'],
                              ['/koji/task/12387276/tdl-i386.xml',
                               '/koji/task/12387276/fedora-cloud-base-2878aa0.ks',
                               '/koji/task/12387276/koji-f23-build-12387276-base.ks',
                               '/koji/task/12387276/libvirt-qcow2-i386.xml',
                               '/koji/task/12387276/Fedora-Cloud-Base-23-20160103.i386.qcow2',
                               '/koji/task/12387276/libvirt-raw-xz-i386.xml',
                               '/koji/task/12387276/Fedora-Cloud-Base-23-20160103.i386.raw.xz'])
        self.assertItemsEqual(result['x86_64'],
                              ['/koji/task/12387277/tdl-x86_64.xml',
                               '/koji/task/12387277/fedora-cloud-base-2878aa0.ks',
                               '/koji/task/12387277/koji-f23-build-12387277-base.ks',
                               '/koji/task/12387277/libvirt-qcow2-x86_64.xml',
                               '/koji/task/12387277/Fedora-Cloud-Base-23-20160103.x86_64.qcow2',
                               '/koji/task/12387277/libvirt-raw-xz-x86_64.xml',
                               '/koji/task/12387277/Fedora-Cloud-Base-23-20160103.x86_64.raw.xz'])


class LiveMediaTestCase(unittest.TestCase):
    def setUp(self):
        self.koji_profile = mock.Mock()
        with mock.patch('pungi.wrappers.kojiwrapper.koji') as koji:
            koji.get_profile_module = mock.Mock(
                return_value=mock.Mock(
                    pathinfo=mock.Mock(
                        work=mock.Mock(return_value='/koji'),
                        taskrelpath=mock.Mock(side_effect=lambda id: 'task/' + str(id)),
                        imagebuild=mock.Mock(side_effect=lambda id: '/koji/imagebuild/' + str(id)),
                    )
                )
            )
            self.koji_profile = koji.get_profile_module.return_value
            self.koji = KojiWrapper('koji')

    def test_get_live_media_cmd_minimal(self):
        opts = {
            'name': 'name', 'version': '1', 'target': 'tgt', 'arch': 'x,y,z',
            'ksfile': 'kickstart', 'install_tree': '/mnt/os'
        }
        cmd = self.koji.get_live_media_cmd(opts)
        self.assertEqual(cmd,
                         ['koji', 'spin-livemedia', 'name', '1', 'tgt', 'x,y,z', 'kickstart',
                          '--install-tree=/mnt/os', '--wait'])

    def test_get_live_media_cmd_full(self):
        opts = {
            'name': 'name', 'version': '1', 'target': 'tgt', 'arch': 'x,y,z',
            'ksfile': 'kickstart', 'install_tree': '/mnt/os', 'scratch': True,
            'repo': ['repo-1', 'repo-2'], 'skip_tag': True,
        }
        cmd = self.koji.get_live_media_cmd(opts)
        self.assertEqual(cmd[:8],
                         ['koji', 'spin-livemedia', 'name', '1', 'tgt', 'x,y,z', 'kickstart',
                          '--install-tree=/mnt/os'])
        self.assertItemsEqual(cmd[8:],
                              ['--repo=repo-1', '--repo=repo-2', '--skip-tag', '--scratch', '--wait'])


class RunrootKojiWrapperTest(unittest.TestCase):

    def setUp(self):
        self.koji_profile = mock.Mock()
        with mock.patch('pungi.wrappers.kojiwrapper.koji') as koji:
            koji.get_profile_module = mock.Mock(
                return_value=mock.Mock(
                    pathinfo=mock.Mock(
                        work=mock.Mock(return_value='/koji'),
                        taskrelpath=mock.Mock(side_effect=lambda id: 'task/' + str(id)),
                        imagebuild=mock.Mock(side_effect=lambda id: '/koji/imagebuild/' + str(id)),
                    )
                )
            )
            self.koji_profile = koji.get_profile_module.return_value
            self.koji = KojiWrapper('koji')

    def test_get_cmd_minimal(self):
        cmd = self.koji.get_runroot_cmd('tgt', 's390x', 'date', use_shell=False, task_id=False)
        self.assertEqual(len(cmd), 6)
        self.assertEqual(cmd[0], 'koji')
        self.assertEqual(cmd[1], 'runroot')
        self.assertEqual(cmd[-3], 'tgt')
        self.assertEqual(cmd[-2], 's390x')
        self.assertEqual(cmd[-1], 'rm -f /var/lib/rpm/__db*; rm -rf /var/cache/yum/*; set -x; date')
        self.assertItemsEqual(cmd[2:-3],
                              ['--channel-override=runroot-local'])

    def test_get_cmd_full(self):
        cmd = self.koji.get_runroot_cmd('tgt', 's390x', ['/bin/echo', '&'],
                                        quiet=True, channel='chan',
                                        packages=['strace', 'lorax'],
                                        mounts=['/tmp'], weight=1000)
        self.assertEqual(len(cmd), 13)
        self.assertEqual(cmd[0], 'koji')
        self.assertEqual(cmd[1], 'runroot')
        self.assertEqual(cmd[-3], 'tgt')
        self.assertEqual(cmd[-2], 's390x')
        self.assertEqual(cmd[-1], 'rm -f /var/lib/rpm/__db*; rm -rf /var/cache/yum/*; set -x; /bin/echo \'&\'')
        self.assertItemsEqual(cmd[2:-3],
                              ['--channel-override=chan', '--quiet', '--use-shell',
                               '--task-id', '--weight=1000', '--package=strace',
                               '--package=lorax', '--mount=/tmp'])

    @mock.patch('pungi.wrappers.kojiwrapper.run')
    def test_run_runroot_cmd_no_task_id(self, run):
        cmd = ['koji', 'runroot']
        output = 'Output ...'
        run.return_value = (0, output)

        result = self.koji.run_runroot_cmd(cmd)
        self.assertDictEqual(result, {'retcode': 0, 'output': output, 'task_id': None})

    @mock.patch('pungi.wrappers.kojiwrapper.run')
    def test_run_runroot_cmd_with_task_id(self, run):
        cmd = ['koji', 'runroot', '--task-id']
        output = 'Output ...\n'
        run.return_value = (0, '1234\n' + output)

        result = self.koji.run_runroot_cmd(cmd)
        self.assertDictEqual(result, {'retcode': 0, 'output': output, 'task_id': 1234})

    @mock.patch('pungi.wrappers.kojiwrapper.run')
    def test_run_runroot_cmd_with_task_id_and_fail(self, run):
        cmd = ['koji', 'runroot', '--task-id']
        output = 'You are not authorized to run this\n'
        run.return_value = (1, output)

        result = self.koji.run_runroot_cmd(cmd)
        self.assertDictEqual(result, {'retcode': 1, 'output': output, 'task_id': None})


if __name__ == "__main__":
    unittest.main()