# -*- coding: utf-8 -*- import logging import contextlib try: from unittest import mock except ImportError: import mock import six import productmd import os from tests import helpers from pungi.createiso import CreateIsoOpts from pungi.phases import createiso class CreateisoPhaseTest(helpers.PungiTestCase): @mock.patch("pungi.phases.createiso.ThreadPool") def test_skip_all(self, ThreadPool): compose = helpers.DummyCompose( self.topdir, {"createiso_skip": [("^.*$", {"*": True, "src": True})]} ) pool = ThreadPool.return_value phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = mock.Mock() phase.run() self.assertEqual(len(pool.add.call_args_list), 0) self.assertEqual(pool.queue_put.call_args_list, []) @mock.patch("pungi.phases.createiso.ThreadPool") def test_nothing_happens_without_rpms(self, ThreadPool): compose = helpers.DummyCompose( self.topdir, {"release_short": "test", "release_version": "1.0", "createiso_skip": []}, ) pool = ThreadPool.return_value phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = mock.Mock() phase.run() self.assertEqual(len(pool.add.call_args_list), 0) self.assertEqual(pool.queue_put.call_args_list, []) six.assertCountEqual( self, phase.logger.warning.call_args_list, [ mock.call("No RPMs found for Everything.x86_64, skipping ISO"), mock.call("No RPMs found for Everything.amd64, skipping ISO"), mock.call("No RPMs found for Everything.src, skipping ISO"), mock.call("No RPMs found for Client.amd64, skipping ISO"), mock.call("No RPMs found for Client.src, skipping ISO"), mock.call("No RPMs found for Server.x86_64, skipping ISO"), mock.call("No RPMs found for Server.amd64, skipping ISO"), mock.call("No RPMs found for Server.src, skipping ISO"), ], ) @mock.patch("pungi.createiso.write_script") @mock.patch("pungi.phases.createiso.prepare_iso") @mock.patch("pungi.phases.createiso.split_iso") @mock.patch("pungi.phases.createiso.ThreadPool") def test_start_one_worker(self, ThreadPool, split_iso, prepare_iso, write_script): compose = helpers.DummyCompose( self.topdir, {"release_short": "test", "release_version": "1.0", "createiso_skip": []}, ) helpers.touch( os.path.join( compose.paths.compose.os_tree("x86_64", compose.variants["Server"]), "dummy.rpm", ) ) disc_data = mock.Mock() split_iso.return_value = [disc_data] prepare_iso.return_value = "dummy-graft-points" pool = ThreadPool.return_value phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = mock.Mock() phase.run() self.assertEqual( prepare_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], disc_count=1, disc_num=1, split_iso_data=disc_data, ) ], ) self.assertEqual( split_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], no_split=False, logger=phase.logger, ) ], ) self.assertEqual(len(pool.add.call_args_list), 1) self.maxDiff = None self.assertEqual( [x[0][0] for x in write_script.call_args_list], [ CreateIsoOpts( output_dir="%s/compose/Server/x86_64/iso" % self.topdir, iso_name="image-name", volid="test-1.0 Server.x86_64", graft_points="dummy-graft-points", arch="x86_64", supported=True, jigdo_dir=None, os_tree=None, hfs_compat=True, use_xorrisofs=False, script_dir="%s/work/x86_64/tmp-Server" % self.topdir, ) ], ) self.assertEqual( pool.queue_put.call_args_list, [ mock.call( ( compose, { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": [ "bash", self.topdir + "/work/x86_64/tmp-Server/createiso-image-name.sh", ], "label": "", "disc_num": 1, "disc_count": 1, }, compose.variants["Server"], "x86_64", ) ) ], ) @mock.patch("pungi.createiso.write_script") @mock.patch("pungi.phases.createiso.prepare_iso") @mock.patch("pungi.phases.createiso.split_iso") @mock.patch("pungi.phases.createiso.ThreadPool") def test_bootable(self, ThreadPool, split_iso, prepare_iso, write_script): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "buildinstall_method": "lorax", "bootable": True, "createiso_skip": [], }, ) helpers.touch( os.path.join( compose.paths.compose.os_tree("x86_64", compose.variants["Server"]), "dummy.rpm", ) ) helpers.touch( os.path.join( compose.paths.compose.os_tree("src", compose.variants["Server"]), "dummy.rpm", ) ) disc_data = mock.Mock() split_iso.return_value = [disc_data] prepare_iso.return_value = "dummy-graft-points" pool = ThreadPool.return_value phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = mock.Mock() phase.run() six.assertCountEqual( self, prepare_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], disc_count=1, disc_num=1, split_iso_data=disc_data, ), mock.call( compose, "src", compose.variants["Server"], disc_count=1, disc_num=1, split_iso_data=disc_data, ), ], ) six.assertCountEqual( self, split_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], no_split=True, logger=phase.logger, ), mock.call( compose, "src", compose.variants["Server"], no_split=False, logger=phase.logger, ), ], ) self.assertEqual(len(pool.add.call_args_list), 2) self.maxDiff = None six.assertCountEqual( self, [x[0][0] for x in write_script.call_args_list], [ CreateIsoOpts( output_dir="%s/compose/Server/x86_64/iso" % self.topdir, boot_iso=( "%s/compose/Server/x86_64/os/images/boot.iso" % self.topdir ), iso_name="image-name", volid="test-1.0 Server.x86_64", graft_points="dummy-graft-points", arch="x86_64", buildinstall_method="lorax", supported=True, jigdo_dir=None, os_tree=None, hfs_compat=True, use_xorrisofs=False, script_dir="%s/work/x86_64/tmp-Server" % self.topdir, ), CreateIsoOpts( output_dir="%s/compose/Server/source/iso" % self.topdir, iso_name="image-name", volid="test-1.0 Server.src", graft_points="dummy-graft-points", arch="src", supported=True, jigdo_dir=None, os_tree=None, hfs_compat=True, use_xorrisofs=False, script_dir="%s/work/src/tmp-Server" % self.topdir, ), ], ) six.assertCountEqual( self, pool.queue_put.call_args_list, [ mock.call( ( compose, { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": True, "cmd": [ "bash", self.topdir + "/work/x86_64/tmp-Server/createiso-image-name.sh", ], "label": "", "disc_num": 1, "disc_count": 1, }, compose.variants["Server"], "x86_64", ) ), mock.call( ( compose, { "iso_path": "%s/compose/Server/source/iso/image-name" % self.topdir, "bootable": False, "cmd": [ "bash", self.topdir + "/work/src/tmp-Server/createiso-image-name.sh", ], "label": "", "disc_num": 1, "disc_count": 1, }, compose.variants["Server"], "src", ) ), ], ) @mock.patch("pungi.createiso.write_script") @mock.patch("pungi.phases.createiso.prepare_iso") @mock.patch("pungi.phases.createiso.split_iso") @mock.patch("pungi.phases.createiso.ThreadPool") def test_bootable_but_failed( self, ThreadPool, split_iso, prepare_iso, write_script ): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "buildinstall_method": "lorax", "bootable": True, "createiso_skip": [], }, ) helpers.touch( os.path.join( compose.paths.compose.os_tree("x86_64", compose.variants["Server"]), "dummy.rpm", ) ) helpers.touch( os.path.join( compose.paths.compose.os_tree("src", compose.variants["Server"]), "dummy.rpm", ) ) disc_data = mock.Mock() split_iso.return_value = [disc_data] prepare_iso.return_value = "dummy-graft-points" pool = ThreadPool.return_value mock_bi = mock.Mock(succeeded=lambda v, a: False) phase = createiso.CreateisoPhase(compose, mock_bi) phase.logger = mock.Mock() phase.run() self.assertEqual( prepare_iso.call_args_list, [ mock.call( compose, "src", compose.variants["Server"], disc_count=1, disc_num=1, split_iso_data=disc_data, ) ], ) self.assertEqual( split_iso.call_args_list, [ mock.call( compose, "src", compose.variants["Server"], no_split=False, logger=phase.logger, ) ], ) self.assertEqual(len(pool.add.call_args_list), 1) self.maxDiff = None self.assertEqual( [x[0][0] for x in write_script.call_args_list], [ CreateIsoOpts( output_dir="%s/compose/Server/source/iso" % self.topdir, iso_name="image-name", volid="test-1.0 Server.src", graft_points="dummy-graft-points", arch="src", supported=True, jigdo_dir=None, os_tree=None, hfs_compat=True, use_xorrisofs=False, script_dir="%s/work/src/tmp-Server" % self.topdir, ) ], ) self.assertEqual( pool.queue_put.call_args_list, [ mock.call( ( compose, { "iso_path": "%s/compose/Server/source/iso/image-name" % self.topdir, "bootable": False, "cmd": [ "bash", self.topdir + "/work/src/tmp-Server/createiso-image-name.sh", ], "label": "", "disc_num": 1, "disc_count": 1, }, compose.variants["Server"], "src", ) ) ], ) @mock.patch("pungi.createiso.write_script") @mock.patch("pungi.phases.createiso.prepare_iso") @mock.patch("pungi.phases.createiso.split_iso") @mock.patch("pungi.phases.createiso.ThreadPool") def test_bootable_product_but_not_variant( self, ThreadPool, split_iso, prepare_iso, write_script ): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "buildinstall_method": "lorax", "bootable": True, "createiso_skip": [], "buildinstall_skip": [("Server", {"*": True})], "iso_hfs_ppc64le_compatible": False, }, ) helpers.touch( os.path.join( compose.paths.compose.os_tree("x86_64", compose.variants["Server"]), "dummy.rpm", ) ) disc_data = mock.Mock() split_iso.return_value = [disc_data] prepare_iso.return_value = "dummy-graft-points" pool = ThreadPool.return_value mock_bi = mock.Mock(succeeded=lambda v, a: False) phase = createiso.CreateisoPhase(compose, mock_bi) phase.logger = mock.Mock() phase.run() self.maxDiff = None self.assertEqual( prepare_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], disc_count=1, disc_num=1, split_iso_data=disc_data, ) ], ) self.assertEqual( split_iso.call_args_list, [ mock.call( compose, "x86_64", compose.variants["Server"], no_split=False, logger=phase.logger, ) ], ) self.assertEqual(len(pool.add.call_args_list), 1) self.assertEqual( [x[0][0] for x in write_script.call_args_list], [ CreateIsoOpts( output_dir="%s/compose/Server/x86_64/iso" % self.topdir, iso_name="image-name", volid="test-1.0 Server.x86_64", graft_points="dummy-graft-points", arch="x86_64", supported=True, jigdo_dir=None, os_tree=None, hfs_compat=False, use_xorrisofs=False, script_dir="%s/work/x86_64/tmp-Server" % self.topdir, ) ], ) self.assertEqual( pool.queue_put.call_args_list, [ mock.call( ( compose, { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": [ "bash", self.topdir + "/work/x86_64/tmp-Server/createiso-image-name.sh", ], "label": "", "disc_num": 1, "disc_count": 1, }, compose.variants["Server"], "x86_64", ) ) ], ) class CreateisoThreadTest(helpers.PungiTestCase): @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_in_runroot(self, KojiWrapper, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } get_file_size.return_value = 1024 get_mtime.return_value = 13579 getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = {"arches": "x86_64"} get_runroot_cmd = KojiWrapper.return_value.get_runroot_cmd run_runroot = KojiWrapper.return_value.run_runroot_cmd run_runroot.return_value = { "retcode": 0, "output": "whatever", "task_id": 1234, } t = createiso.CreateIsoThread(mock.Mock()) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) self.assertEqual(getTag.call_args_list, [mock.call("f25-build")]) self.assertEqual( get_runroot_cmd.call_args_list, [ mock.call( "f25-build", "x86_64", cmd["cmd"], channel=None, mounts=[self.topdir], packages=["coreutils", "genisoimage", "isomd5sum"], use_shell=True, weight=None, ) ], ) self.assertEqual( run_runroot.call_args_list, [ mock.call( get_runroot_cmd.return_value, log_file="%s/logs/x86_64/createiso-image-name.x86_64.log" % self.topdir, ) ], ) self.assertEqual( iso.get_implanted_md5.call_args_list, [mock.call(cmd["iso_path"], logger=compose._logger)], ) self.assertEqual( iso.get_volume_id.call_args_list, [mock.call(cmd["iso_path"], False)] ) self.assertEqual(len(compose.im.add.call_args_list), 1) args, _ = compose.im.add.call_args_list[0] self.assertEqual(args[0], "Server") self.assertEqual(args[1], "x86_64") image = args[2] self.assertEqual(image.arch, "x86_64") self.assertEqual(image.path, "Server/x86_64/iso/image-name") self.assertEqual(image.format, "iso") self.assertEqual(image.type, "dvd") self.assertEqual(image.subvariant, "Server") @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_source_iso(self, KojiWrapper, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", "create_jigdo": False, "runroot_weights": {"createiso": 123}, }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } get_file_size.return_value = 1024 get_mtime.return_value = 13579 getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = {"arches": "x86_64"} get_runroot_cmd = KojiWrapper.return_value.get_runroot_cmd run_runroot = KojiWrapper.return_value.run_runroot_cmd run_runroot.return_value = { "retcode": 0, "output": "whatever", "task_id": 1234, } t = createiso.CreateIsoThread(mock.Mock()) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "src"), 1) self.assertEqual(getTag.call_args_list, [mock.call("f25-build")]) self.assertEqual( get_runroot_cmd.call_args_list, [ mock.call( "f25-build", "x86_64", cmd["cmd"], channel=None, mounts=[self.topdir], packages=["coreutils", "genisoimage", "isomd5sum"], use_shell=True, weight=123, ) ], ) self.assertEqual( run_runroot.call_args_list, [ mock.call( get_runroot_cmd.return_value, log_file="%s/logs/src/createiso-image-name.src.log" % self.topdir, ) ], ) self.assertEqual( iso.get_implanted_md5.call_args_list, [mock.call(cmd["iso_path"], logger=compose._logger)], ) self.assertEqual( iso.get_volume_id.call_args_list, [mock.call(cmd["iso_path"], False)] ) self.assertEqual(len(compose.im.add.call_args_list), 2) for args, _ in compose.im.add.call_args_list: self.assertEqual(args[0], "Server") self.assertIn(args[1], ["x86_64", "amd64"]) image = args[2] self.assertEqual(image.arch, "src") self.assertEqual(image.path, "Server/x86_64/iso/image-name") self.assertEqual(image.format, "iso") self.assertEqual(image.type, "dvd") self.assertEqual(image.subvariant, "Server") @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_bootable(self, KojiWrapper, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "bootable": True, "buildinstall_method": "lorax", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": True, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } get_file_size.return_value = 1024 get_mtime.return_value = 13579 getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = {"arches": "x86_64"} get_runroot_cmd = KojiWrapper.return_value.get_runroot_cmd run_runroot = KojiWrapper.return_value.run_runroot_cmd run_runroot.return_value = { "retcode": 0, "output": "whatever", "task_id": 1234, } t = createiso.CreateIsoThread(mock.Mock()) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) # There is no need to call getTag if `bootable` is True. self.assertEqual(getTag.call_args_list, []) self.assertEqual( get_runroot_cmd.call_args_list, [ mock.call( "f25-build", "x86_64", cmd["cmd"], channel=None, mounts=[self.topdir], packages=[ "coreutils", "genisoimage", "isomd5sum", "lorax", "which", ], use_shell=True, weight=None, ) ], ) self.assertEqual( run_runroot.call_args_list, [ mock.call( get_runroot_cmd.return_value, log_file="%s/logs/x86_64/createiso-image-name.x86_64.log" % self.topdir, ) ], ) self.assertEqual( iso.get_implanted_md5.call_args_list, [mock.call(cmd["iso_path"], logger=compose._logger)], ) self.assertEqual( iso.get_volume_id.call_args_list, [mock.call(cmd["iso_path"], False)] ) self.assertEqual(len(compose.im.add.call_args_list), 1) args, _ = compose.im.add.call_args_list[0] self.assertEqual(args[0], "Server") self.assertEqual(args[1], "x86_64") image = args[2] self.assertEqual(image.arch, "x86_64") self.assertEqual(image.path, "Server/x86_64/iso/image-name") self.assertEqual(image.format, "iso") self.assertEqual(image.type, "dvd") self.assertEqual(image.subvariant, "Server") @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_in_runroot_non_existing_tag( self, KojiWrapper, get_file_size, get_mtime, iso ): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = None t = createiso.CreateIsoThread(mock.Mock()) with self.assertRaises(RuntimeError) as ctx: with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) self.assertEqual('Tag "f25-build" does not exist.', str(ctx.exception)) @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_in_runroot_crash(self, KojiWrapper, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", "failable_deliverables": [("^.*$", {"*": "iso"})], }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = {"arches": "x86_64"} run_runroot = KojiWrapper.return_value.run_runroot_cmd run_runroot.side_effect = helpers.boom pool = mock.Mock() t = createiso.CreateIsoThread(pool) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) pool._logger.error.assert_has_calls( [ mock.call( "[FAIL] Iso (variant Server, arch x86_64) failed, but going on anyway." # noqa: E501 ), mock.call("BOOM"), ] ) @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_in_runroot_fail(self, KojiWrapper, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "runroot_tag": "f25-build", "koji_profile": "koji", "koji_cache": "/tmp", "failable_deliverables": [("^.*$", {"*": "iso"})], }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } getTag = KojiWrapper.return_value.koji_proxy.getTag getTag.return_value = {"arches": "x86_64"} run_runroot = KojiWrapper.return_value.run_runroot_cmd run_runroot.return_value = { "retcode": 1, "output": "Nope", "task_id": "1234", } pool = mock.Mock() t = createiso.CreateIsoThread(pool) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) pool._logger.error.assert_has_calls( [ mock.call( "[FAIL] Iso (variant Server, arch x86_64) failed, but going on anyway." # noqa: E501 ), mock.call( "Runroot task failed: 1234. See %s for more details." % (self.topdir + "/logs/x86_64/createiso-image-name.x86_64.log") ), ] ) @mock.patch("pungi.phases.createiso.iso") @mock.patch("pungi.phases.createiso.get_mtime") @mock.patch("pungi.phases.createiso.get_file_size") @mock.patch("pungi.runroot.run") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_locally(self, KojiWrapper, run, get_file_size, get_mtime, iso): compose = helpers.DummyCompose( self.topdir, {"release_short": "test", "release_version": "1.0"} ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } get_file_size.return_value = 1024 get_mtime.return_value = 13579 t = createiso.CreateIsoThread(mock.Mock()) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) self.assertEqual(KojiWrapper.return_value.mock_calls, []) self.assertEqual( run.call_args_list, [ mock.call( cmd["cmd"], show_cmd=True, logfile="%s/logs/x86_64/createiso-image-name.x86_64.log" % self.topdir, ) ], ) self.assertEqual( iso.get_implanted_md5.call_args_list, [mock.call(cmd["iso_path"], logger=compose._logger)], ) self.assertEqual( iso.get_volume_id.call_args_list, [mock.call(cmd["iso_path"], False)] ) self.assertEqual(len(compose.im.add.call_args_list), 1) args, _ = compose.im.add.call_args_list[0] self.assertEqual(args[0], "Server") self.assertEqual(args[1], "x86_64") image = args[2] self.assertEqual(image.arch, "x86_64") self.assertEqual(image.path, "Server/x86_64/iso/image-name") self.assertEqual(image.format, "iso") self.assertEqual(image.type, "dvd") self.assertEqual(image.subvariant, "Server") @mock.patch("pungi.runroot.run") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") def test_process_locally_crash(self, KojiWrapper, run): compose = helpers.DummyCompose( self.topdir, { "release_short": "test", "release_version": "1.0", "failable_deliverables": [("^.*$", {"*": "iso"})], }, ) cmd = { "iso_path": "%s/compose/Server/x86_64/iso/image-name" % self.topdir, "bootable": False, "cmd": mock.Mock(), "label": "", "disc_num": 1, "disc_count": 1, } run.side_effect = helpers.boom pool = mock.Mock() t = createiso.CreateIsoThread(pool) with mock.patch("time.sleep"): t.process((compose, cmd, compose.variants["Server"], "x86_64"), 1) pool._logger.error.assert_has_calls( [ mock.call( "[FAIL] Iso (variant Server, arch x86_64) failed, but going on anyway." # noqa: E501 ), mock.call("BOOM"), ] ) TREEINFO = """ [header] version = 1.0 [release] name = Dummy Product short = DP version = 1.0 [tree] arch = x86_64 platforms = x86_64 build_timestamp = 1464715102 variants = Server [variant-Server] id = Server uid = Server name = Server type = variant """ class DummySize(object): """ This is intended as a replacement for os.path.getsize that returns predefined sizes. The argument to __init__ should be a mapping from substring of filepath to size. """ def __init__(self, sizes): self.sizes = sizes def __call__(self, path): for fragment, size in self.sizes.items(): if fragment in path: return size return 0 class SplitIsoTest(helpers.PungiTestCase): def test_split_fits_on_single_disc(self): compose = helpers.DummyCompose(self.topdir, {}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch(os.path.join(self.topdir, "work/x86_64/Server/extra-files/GPL")) helpers.touch(os.path.join(self.topdir, "compose/Server/x86_64/os/GPL")) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/repodata/repomd.xml") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/b/bash.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/n/media.repo") ) with mock.patch( "os.path.getsize", DummySize( { "GPL": 20 * 2048, "bash": 150 * 2048, "media": 100 * 2048, "treeinfo": 10 * 2048, } ), ): data = createiso.split_iso(compose, "x86_64", compose.variants["Server"]) base_path = os.path.join(self.topdir, "compose/Server/x86_64/os") # GPL is sticky file, it should be first at all times. Files are # searched top-down, so nested ones are after top level ones. self.assertEqual( data, [ { "files": [ os.path.join(base_path, "GPL"), os.path.join(base_path, ".treeinfo"), os.path.join(base_path, "n/media.repo"), os.path.join(base_path, "Packages/b/bash.rpm"), ], "size": 573440, } ], ) def test_split_needs_two_discs(self): compose = helpers.DummyCompose(self.topdir, {}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch(os.path.join(self.topdir, "work/x86_64/Server/extra-files/GPL")) helpers.touch(os.path.join(self.topdir, "compose/Server/x86_64/os/GPL")) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/repodata/repomd.xml") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/b/bash.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/n/media.repo") ) M = 1024**2 G = 1024**3 with mock.patch( "os.path.getsize", DummySize( {"GPL": 20 * M, "bash": 3 * G, "media": 2 * G, "treeinfo": 10 * M} ), ): data = createiso.split_iso(compose, "x86_64", compose.variants["Server"]) base_path = os.path.join(self.topdir, "compose/Server/x86_64/os") # GPL is the only sticky file, it should be first at all times. # Files are searched top-down, so nested ones are after top level ones. self.assertEqual( data, [ { "files": [ os.path.join(base_path, "GPL"), os.path.join(base_path, ".treeinfo"), os.path.join(base_path, "n/media.repo"), ], "size": 2178940928, }, { "files": [ os.path.join(base_path, "GPL"), os.path.join(base_path, "Packages/b/bash.rpm"), ], "size": 3242196992, }, ], ) def test_no_split_when_requested(self): compose = helpers.DummyCompose(self.topdir, {}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch(os.path.join(self.topdir, "work/x86_64/Server/extra-files/GPL")) helpers.touch(os.path.join(self.topdir, "compose/Server/x86_64/os/GPL")) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/repodata/repomd.xml") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/b/bash.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/n/media.repo") ) M = 1024**2 G = 1024**3 with mock.patch( "os.path.getsize", DummySize( {"GPL": 20 * M, "bash": 3 * G, "media": 2 * G, "treeinfo": 10 * M} ), ): data = createiso.split_iso( compose, "x86_64", compose.variants["Server"], no_split=True ) base_path = os.path.join(self.topdir, "compose/Server/x86_64/os") # GPL is the only sticky file, it should be first at all times. # Files are searched top-down, so nested ones are after top level ones. self.assertEqual( data, [ { "files": [ os.path.join(base_path, "GPL"), os.path.join(base_path, ".treeinfo"), os.path.join(base_path, "n/media.repo"), os.path.join(base_path, "Packages/b/bash.rpm"), ], "size": 5400166400, } ], ) self.assertEqual( compose._logger.warning.call_args_list, [ mock.call( "ISO for Server.x86_64 does not fit on single media! " "It is 710652160 bytes too big. (Total size: 5400166400 B)" ) ], ) def test_keeps_reserve(self): compose = helpers.DummyCompose(self.topdir, {}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/spacer.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/x/pad.rpm") ) M = 1024**2 # treeinfo has size 0, spacer leaves 11M of free space, so with 10M # reserve the padding package should be on second disk with mock.patch( "os.path.getsize", DummySize({"spacer": 4688465664, "pad": 5 * M}) ): data = createiso.split_iso(compose, "x86_64", compose.variants["Server"]) base_path = os.path.join(self.topdir, "compose/Server/x86_64/os") self.assertEqual(len(data), 2) self.assertEqual( data[0]["files"], [ os.path.join(base_path, ".treeinfo"), os.path.join(base_path, "Packages/spacer.rpm"), ], ) self.assertEqual( data[1]["files"], [os.path.join(base_path, "Packages/x/pad.rpm")] ) def test_can_customize_reserve(self): compose = helpers.DummyCompose(self.topdir, {"split_iso_reserve": 1024**2}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/spacer.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/x/pad.rpm") ) M = 1024**2 with mock.patch( "os.path.getsize", DummySize({"spacer": 4688465664, "pad": 5 * M}) ): data = createiso.split_iso(compose, "x86_64", compose.variants["Server"]) self.assertEqual(len(data), 1) def test_can_change_iso_size(self): compose = helpers.DummyCompose(self.topdir, {"iso_size": "8G"}) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/.treeinfo"), TREEINFO ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/spacer.rpm") ) helpers.touch( os.path.join(self.topdir, "compose/Server/x86_64/os/Packages/x/pad.rpm") ) M = 1024**2 with mock.patch( "os.path.getsize", DummySize({"spacer": 4688465664, "pad": 5 * M}) ): data = createiso.split_iso(compose, "x86_64", compose.variants["Server"]) self.assertEqual(len(data), 1) class BreakHardlinksTest(helpers.PungiTestCase): def setUp(self): super(BreakHardlinksTest, self).setUp() self.src = os.path.join(self.topdir, "src") self.stage = os.path.join(self.topdir, "stage") def test_not_modify_dir(self): p = os.path.join(self.src, "dir") os.makedirs(p) d = {"dir": p} createiso.break_hardlinks(d, self.stage) self.assertEqual(d, {"dir": p}) def test_not_copy_file_with_one(self): f = os.path.join(self.src, "file") helpers.touch(f) d = {"f": f} createiso.break_hardlinks(d, self.stage) self.assertEqual(d, {"f": f}) def test_copy(self): f = os.path.join(self.src, "file") helpers.touch(f) os.link(f, os.path.join(self.topdir, "file")) d = {"f": f} createiso.break_hardlinks(d, self.stage) expected = self.stage + f self.assertEqual(d, {"f": expected}) self.assertTrue(os.path.exists(expected)) class TweakTreeinfo(helpers.PungiTestCase): def test_tweaking(self): input = os.path.join(helpers.FIXTURE_DIR, "original-treeinfo") expected = os.path.join(helpers.FIXTURE_DIR, "expected-treeinfo") output = os.path.join(self.topdir, "output") ti = createiso.load_and_tweak_treeinfo(input) ti.dump(output) self.assertFilesEqual(output, expected) class CreateisoTryReusePhaseTest(helpers.PungiTestCase): def setUp(self): super(CreateisoTryReusePhaseTest, self).setUp() self.logger = logging.getLogger() self.logger.setLevel(logging.DEBUG) self.logger.addHandler(logging.StreamHandler(os.devnull)) def test_disabled(self): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": False}) phase = createiso.CreateisoPhase(compose, mock.Mock()) self.assertFalse(phase.try_reuse(mock.Mock(), "Server", "x86_64", mock.Mock())) def test_buildinstall_changed(self): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger phase.bi = mock.Mock() phase.bi.reused.return_value = False cmd = {"disc_num": 1, "disc_count": 1} opts = CreateIsoOpts(buildinstall_method="lorax") self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) def test_no_old_config(self): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger cmd = {"disc_num": 1, "disc_count": 1} opts = CreateIsoOpts() self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) def test_old_config_changed(self): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) old_config = compose.conf.copy() old_config["release_version"] = "2" compose.load_old_compose_config.return_value = old_config phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger cmd = {"disc_num": 1, "disc_count": 1} opts = CreateIsoOpts() self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) def test_no_old_metadata(self): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) compose.load_old_compose_config.return_value = compose.conf.copy() phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger cmd = {"disc_num": 1, "disc_count": 1} opts = CreateIsoOpts() self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) @mock.patch("pungi.phases.createiso.read_json_file") def test_volume_id_differs(self, read_json_file): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) compose.load_old_compose_config.return_value = compose.conf.copy() phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger cmd = {"disc_num": 1, "disc_count": 1} opts = CreateIsoOpts(volid="new-volid") read_json_file.return_value = {"opts": {"volid": "old-volid"}} self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) @mock.patch("pungi.phases.createiso.read_json_file") def test_packages_differ(self, read_json_file): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) compose.load_old_compose_config.return_value = compose.conf.copy() phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger cmd = {"disc_num": 1, "disc_count": 1} new_graft_points = os.path.join(self.topdir, "new_graft_points") helpers.touch(new_graft_points, "Packages/f/foo-1-1.x86_64.rpm\n") opts = CreateIsoOpts(graft_points=new_graft_points, volid="volid") old_graft_points = os.path.join(self.topdir, "old_graft_points") helpers.touch(old_graft_points, "Packages/f/foo-1-2.x86_64.rpm\n") read_json_file.return_value = { "opts": {"graft_points": old_graft_points, "volid": "volid"} } self.assertFalse( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) @mock.patch("pungi.phases.createiso.read_json_file") def test_runs_perform_reuse(self, read_json_file): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) compose.load_old_compose_config.return_value = compose.conf.copy() phase = createiso.CreateisoPhase(compose, mock.Mock()) phase.logger = self.logger phase.perform_reuse = mock.Mock() cmd = {"disc_num": 1, "disc_count": 1} new_graft_points = os.path.join(self.topdir, "new_graft_points") helpers.touch(new_graft_points) opts = CreateIsoOpts(graft_points=new_graft_points, volid="volid") old_graft_points = os.path.join(self.topdir, "old_graft_points") helpers.touch(old_graft_points) dummy_iso_path = "dummy-iso-path" read_json_file.return_value = { "opts": { "graft_points": old_graft_points, "volid": "volid", }, "cmd": {"iso_path": dummy_iso_path}, } self.assertTrue( phase.try_reuse(cmd, compose.variants["Server"], "x86_64", opts) ) self.assertEqual( phase.perform_reuse.call_args_list, [ mock.call( cmd, compose.variants["Server"], "x86_64", opts, dummy_iso_path, ) ], ) @mock.patch("pungi.phases.createiso.OldFileLinker") @mock.patch("pungi.phases.createiso.add_iso_to_metadata") class CreateisoPerformReusePhaseTest(helpers.PungiTestCase): def test_success(self, add_iso_to_metadata, OldFileLinker): compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) phase = createiso.CreateisoPhase(compose, mock.Mock()) cmd = { "iso_path": "target/image.iso", "bootable": False, "disc_num": 1, "disc_count": 2, } opts = CreateIsoOpts() phase.perform_reuse( cmd, compose.variants["Server"], "x86_64", opts, "old/image.iso", ) self.assertEqual( add_iso_to_metadata.call_args_list, [ mock.call( compose, compose.variants["Server"], "x86_64", cmd["iso_path"], bootable=False, disc_count=2, disc_num=1, ), ], ) self.assertEqual( OldFileLinker.return_value.mock_calls, [ mock.call.link("old/image.iso", "target/image.iso"), mock.call.link("old/image.iso.manifest", "target/image.iso.manifest"), # The old log file doesn't exist in the test scenario. mock.call.link( None, os.path.join( self.topdir, "logs/x86_64/createiso-image.iso.x86_64.log" ), ), ], ) def test_failure(self, add_iso_to_metadata, OldFileLinker): OldFileLinker.return_value.link.side_effect = helpers.mk_boom() compose = helpers.DummyCompose(self.topdir, {"createiso_allow_reuse": True}) phase = createiso.CreateisoPhase(compose, mock.Mock()) cmd = { "iso_path": "target/image.iso", "bootable": False, "disc_num": 1, "disc_count": 2, } opts = CreateIsoOpts() with self.assertRaises(Exception): phase.perform_reuse( cmd, compose.variants["Server"], "x86_64", opts, "old/image.iso", ) self.assertEqual(add_iso_to_metadata.call_args_list, []) self.assertEqual( OldFileLinker.return_value.mock_calls, [ mock.call.link("old/image.iso", "target/image.iso"), mock.call.abort(), ], ) class ComposeConfGetIsoLevelTest(helpers.PungiTestCase): def test_global_config(self): compose = helpers.DummyCompose(self.topdir, {"iso_level": 3}) self.assertEqual( createiso.get_iso_level_config( compose, compose.variants["Server"], "x86_64" ), 3, ) def test_src_only_config(self): compose = helpers.DummyCompose( self.topdir, {"iso_level": [(".*", {"src": 4})]}, ) self.assertEqual( createiso.get_iso_level_config(compose, compose.variants["Server"], "src"), 4, ) def test_no_match(self): compose = helpers.DummyCompose( self.topdir, {"iso_level": [("^Server$", {"*": 4})]}, ) self.assertIsNone( createiso.get_iso_level_config( compose, compose.variants["Client"], "x86_64" ), ) def mk_mount(topdir, images): @contextlib.contextmanager def dummy_mount(path, logger): treeinfo = [ "[general]", "family = Test", "version = 1.0", "arch = x86_64", "variant = Server", "[checksums]", ] for image in images: helpers.touch(os.path.join(topdir, image.path), image.content) treeinfo.append("%s = sha256:%s" % (image.path, image.checksum)) helpers.touch(os.path.join(topdir, ".treeinfo"), "\n".join(treeinfo)) yield topdir return dummy_mount class _MockRun: """This class replaces kobo.shortcuts.run and validates that the correct two commands are called. The assertions can not be done after the tested function finishes because it will clean up the .treeinfo file that needs to be checked. """ def __init__(self): self.num_calls = 0 self.asserts = [self._assert_xorriso, self._assert_implantisomd5] def __call__(self, cmd, logfile): self.num_calls += 1 self.asserts.pop(0)(cmd) def _assert_xorriso(self, cmd): assert cmd[0] == "xorriso" ti = productmd.TreeInfo() input_iso = None for i, arg in enumerate(cmd): if arg == "-map": ti.load(cmd[i + 1]) if arg == "-outdev": self.temp_iso = cmd[i + 1] if arg == "-indev": input_iso = cmd[i + 1] assert self.input_iso == input_iso assert ti.checksums.checksums[self.image_relative_path] == self.image_checksum def _assert_implantisomd5(self, cmd): assert cmd[0] == "/usr/bin/implantisomd5" assert cmd[-1] == self.temp_iso class DummyImage: def __init__(self, path, content, checksum=None): self.path = path self.content = content self.checksum = checksum or helpers.hash_string("sha256", content) @mock.patch("os.rename") @mock.patch("pungi.phases.createiso.run", new_callable=_MockRun) class FixChecksumsTest(helpers.PungiTestCase): def test_checksum_matches(self, mock_run, mock_rename): compose = helpers.DummyCompose(self.topdir, {}) arch = "x86_64" iso_path = "DUMMY_ISO" with mock.patch( "pungi.wrappers.iso.mount", new=mk_mount(self.topdir, [DummyImage("images/eltorito.img", "eltorito")]), ): createiso.fix_treeinfo_checksums(compose, iso_path, arch) self.assertEqual(mock_run.num_calls, 0) self.assertEqual(mock_rename.call_args_list, []) def test_checksum_fix(self, mock_run, mock_rename): compose = helpers.DummyCompose(self.topdir, {}) arch = "x86_64" img = "images/eltorito.img" content = "eltorito" iso_path = "DUMMY_ISO" mock_run.input_iso = iso_path mock_run.image_relative_path = "images/eltorito.img" mock_run.image_checksum = ("sha256", helpers.hash_string("sha256", content)) with mock.patch( "pungi.wrappers.iso.mount", new=mk_mount(self.topdir, [DummyImage(img, content, "abc")]), ): createiso.fix_treeinfo_checksums(compose, iso_path, arch) # The new image was copied over the old one self.assertEqual( mock_rename.call_args_list, [mock.call(mock_run.temp_iso, iso_path)] )