import os try: from unittest import mock except ImportError: import mock from pungi.phases.kiwibuild import KiwiBuildPhase, RunKiwiBuildThread from tests.helpers import DummyCompose, PungiTestCase MINIMAL_CONF = { "description_scm": "https://example.com/kiwi.git", "description_path": "Fedora.kiwi", "kiwi_profile": "Cloud-Base-Generic", } def _merge(a, b): """This would be a | b on 3.9 and later, or {**a, **b} or 3.5 and later.""" c = a.copy() c.update(b) return c @mock.patch("pungi.phases.kiwibuild.ThreadPool") class TestKiwiBuildPhase(PungiTestCase): def test_minimal(self, ThreadPool): cfg = _merge({"target": "f40"}, MINIMAL_CONF) compose = DummyCompose(self.topdir, {"kiwibuild": {"^Server$": [cfg]}}) self.assertValidConfig(compose.conf) phase = KiwiBuildPhase(compose) phase.run() phase.pool.add.assert_called() assert phase.pool.queue_put.call_args_list == [ mock.call( ( compose, compose.variants["Server"], cfg, ["amd64", "x86_64"], { "release": None, "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": None, "type_attr": None, "bundle_name_format": None, }, [self.topdir + "/compose/Server/$arch/os"], [], # failable arches ) ) ] def test_full(self, ThreadPool): cfg = _merge( { "target": "f40", "release": "1234", "arches": ["x86_64"], "repos": ["https://example.com/repo/", "Client"], "failable": ["*"], "subvariant": "Test", "type": "custom", "type_attr": ["foo", "bar"], "bundle_name_format": "fmt", }, MINIMAL_CONF, ) compose = DummyCompose(self.topdir, {"kiwibuild": {"^Server$": [cfg]}}) self.assertValidConfig(compose.conf) phase = KiwiBuildPhase(compose) phase.run() phase.pool.add.assert_called() assert phase.pool.queue_put.call_args_list == [ mock.call( ( compose, compose.variants["Server"], cfg, ["x86_64"], { "release": "1234", "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": "custom", "type_attr": ["foo", "bar"], "bundle_name_format": "fmt", }, [ "https://example.com/repo/", self.topdir + "/compose/Client/$arch/os", self.topdir + "/compose/Server/$arch/os", ], ["x86_64"], ) ) ] def test_failable(self, ThreadPool): cfg = _merge({"target": "f40", "failable": ["x86_64"]}, MINIMAL_CONF) compose = DummyCompose(self.topdir, {"kiwibuild": {"^Server$": [cfg]}}) self.assertValidConfig(compose.conf) phase = KiwiBuildPhase(compose) phase.run() phase.pool.add.assert_called() assert phase.pool.queue_put.call_args_list == [ mock.call( ( compose, compose.variants["Server"], cfg, ["amd64", "x86_64"], { "release": None, "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": None, "type_attr": None, "bundle_name_format": None, }, [self.topdir + "/compose/Server/$arch/os"], ["x86_64"], # failable arches ) ) ] def test_with_phase_opts(self, ThreadPool): cfg = {"kiwi_profile": "Generic"} compose = DummyCompose( self.topdir, { "kiwibuild": {"^Server$": [cfg]}, "kiwibuild_target": "f40", "kiwibuild_release": "1234", "kiwibuild_description_scm": "foo", "kiwibuild_description_path": "bar", "kiwibuild_type": "custom", "kiwibuild_type_attr": ["foo", "bar"], "kiwibuild_bundle_name_format": "fmt", }, ) self.assertValidConfig(compose.conf) phase = KiwiBuildPhase(compose) phase.run() phase.pool.add.assert_called() assert phase.pool.queue_put.call_args_list == [ mock.call( ( compose, compose.variants["Server"], cfg, ["amd64", "x86_64"], { "release": "1234", "target": "f40", "descscm": "foo", "descpath": "bar", "type": "custom", "type_attr": ["foo", "bar"], "bundle_name_format": "fmt", }, [self.topdir + "/compose/Server/$arch/os"], [], # failable arches ) ) ] def test_with_global_opts(self, ThreadPool): cfg = MINIMAL_CONF compose = DummyCompose( self.topdir, { "kiwibuild": {"^Server$": [cfg]}, "global_target": "f40", "global_release": "1234", }, ) self.assertValidConfig(compose.conf) phase = KiwiBuildPhase(compose) phase.run() phase.pool.add.assert_called() assert phase.pool.queue_put.call_args_list == [ mock.call( ( compose, compose.variants["Server"], cfg, ["amd64", "x86_64"], { "release": "1234", "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": None, "type_attr": None, "bundle_name_format": None, }, [self.topdir + "/compose/Server/$arch/os"], [], # failable arches ) ) ] @mock.patch("pungi.phases.kiwibuild.Linker") @mock.patch("pungi.util.get_mtime") @mock.patch("pungi.util.get_file_size") @mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper") class TestKiwiBuildThread(PungiTestCase): def _img_path(self, arch, filename=None): path = self.topdir + "/compose/Server/%s/images" % arch if filename: path += "/" + filename return path def test_process(self, KojiWrapper, get_file_size, get_mtime, Linker): img_name = "FCBG.{arch}-Rawhide-1.6.vagrant.libvirt.box" self.repo = self.topdir + "/compose/Server/$arch/os" compose = DummyCompose( self.topdir, { "koji_profile": "koji", "kiwibuild_bundle_format": "%N-%P-40_Beta-%I.%A.%T", }, ) config = _merge({"subvariant": "Test"}, MINIMAL_CONF) pool = mock.Mock() get_image_paths = KojiWrapper.return_value.get_image_paths get_image_paths.return_value = { "x86_64": [ "/koji/task/1234/FCBG.x86_64-Rawhide-1.6.packages", "/koji/task/1234/%s" % img_name.format(arch="x86_64"), ], "amd64": [ "/koji/task/1234/FCBG.amd64-Rawhide-1.6.packages", "/koji/task/1234/%s" % img_name.format(arch="amd64"), ], } KojiWrapper.return_value.koji_proxy.kiwiBuild.return_value = 1234 KojiWrapper.return_value.watch_task.return_value = 0 t = RunKiwiBuildThread(pool) get_file_size.return_value = 1024 get_mtime.return_value = 13579 t.process( ( compose, compose.variants["Server"], config, ["amd64", "x86_64"], { "release": "1.6", "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": "t", "type_attr": ["ta"], "bundle_name_format": "fmt", }, [self.repo], [], ), 1, ) assert KojiWrapper.return_value.koji_proxy.kiwiBuild.mock_calls == [ mock.call( "f40", ["amd64", "x86_64"], MINIMAL_CONF["description_scm"], MINIMAL_CONF["description_path"], profile=MINIMAL_CONF["kiwi_profile"], release="1.6", repos=[self.repo], type="t", type_attr=["ta"], result_bundle_name_format="fmt", optional_arches=[], ) ] assert get_image_paths.mock_calls == [mock.call(1234)] assert os.path.isdir(self._img_path("x86_64")) assert os.path.isdir(self._img_path("amd64")) Linker.return_value.link.assert_has_calls( [ mock.call( "/koji/task/1234/FCBG.amd64-Rawhide-1.6.vagrant.libvirt.box", self._img_path("amd64", img_name.format(arch="amd64")), link_type="hardlink-or-copy", ), mock.call( "/koji/task/1234/FCBG.x86_64-Rawhide-1.6.vagrant.libvirt.box", self._img_path("x86_64", img_name.format(arch="x86_64")), link_type="hardlink-or-copy", ), ], any_order=True, ) assert len(compose.im.add.call_args_list) == 2 for call in compose.im.add.call_args_list: _, kwargs = call image = kwargs["image"] expected_path = "Server/{0.arch}/images/{1}".format( image, img_name.format(arch=image.arch) ) assert kwargs["variant"] == "Server" assert kwargs["arch"] in ("amd64", "x86_64") assert kwargs["arch"] == image.arch assert image.path == expected_path assert "vagrant-libvirt.box" == image.format assert "vagrant-libvirt" == image.type assert "Test" == image.subvariant def test_handle_koji_fail(self, KojiWrapper, get_file_size, get_mtime, Linker): self.repo = self.topdir + "/compose/Server/$arch/os" compose = DummyCompose(self.topdir, {"koji_profile": "koji"}) config = MINIMAL_CONF pool = mock.Mock() get_image_paths = KojiWrapper.return_value.get_image_paths KojiWrapper.return_value.koji_proxy.kiwiBuild.return_value = 1234 KojiWrapper.return_value.watch_task.return_value = 1 t = RunKiwiBuildThread(pool) try: t.process( ( compose, compose.variants["Server"], config, ["amd64", "x86_64"], { "release": "1.6", "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": None, "type_attr": None, "bundle_name_format": None, }, [self.repo], [], ), 1, ) assert False, "Exception should have been raised" except RuntimeError: pass assert len(KojiWrapper.return_value.koji_proxy.kiwiBuild.mock_calls) == 1 assert get_image_paths.mock_calls == [] assert Linker.return_value.link.mock_calls == [] assert len(compose.im.add.call_args_list) == 0 def test_handle_fail_on_optional_arch( self, KojiWrapper, get_file_size, get_mtime, Linker ): self.repo = self.topdir + "/compose/Server/$arch/os" compose = DummyCompose(self.topdir, {"koji_profile": "koji"}) config = MINIMAL_CONF pool = mock.Mock() get_image_paths = KojiWrapper.return_value.get_image_paths get_image_paths.return_value = { "x86_64": [ "/koji/task/1234/FCBG.x86_64-Rawhide-1.6.packages", "/koji/task/1234/FCBG.x86_64-Rawhide-1.6.qcow2", ], } KojiWrapper.return_value.koji_proxy.kiwiBuild.return_value = 1234 KojiWrapper.return_value.watch_task.return_value = 0 t = RunKiwiBuildThread(pool) get_file_size.return_value = 1024 get_mtime.return_value = 13579 t.process( ( compose, compose.variants["Server"], config, ["amd64", "x86_64"], { "release": "1.6", "target": "f40", "descscm": MINIMAL_CONF["description_scm"], "descpath": MINIMAL_CONF["description_path"], "type": None, "type_attr": None, "bundle_name_format": None, }, [self.repo], ["amd64"], ), 1, ) assert KojiWrapper.return_value.koji_proxy.kiwiBuild.mock_calls == [ mock.call( "f40", ["amd64", "x86_64"], MINIMAL_CONF["description_scm"], MINIMAL_CONF["description_path"], profile=MINIMAL_CONF["kiwi_profile"], release="1.6", repos=[self.repo], type=None, type_attr=None, result_bundle_name_format=None, optional_arches=["amd64"], ) ] assert get_image_paths.mock_calls == [mock.call(1234)] assert os.path.isdir(self._img_path("x86_64")) assert not os.path.isdir(self._img_path("amd64")) assert Linker.return_value.link.mock_calls == [ mock.call( "/koji/task/1234/FCBG.x86_64-Rawhide-1.6.qcow2", self._img_path("x86_64", "FCBG.x86_64-Rawhide-1.6.qcow2"), link_type="hardlink-or-copy", ), ] assert len(compose.im.add.call_args_list) == 1 _, kwargs = compose.im.add.call_args_list[0] image = kwargs["image"] expected_path = "Server/x86_64/images/FCBG.x86_64-Rawhide-1.6.qcow2" assert kwargs["variant"] == "Server" assert kwargs["arch"] == "x86_64" assert kwargs["arch"] == image.arch assert image.path == expected_path assert "qcow2" == image.format assert "qcow2" == image.type assert "Server" == image.subvariant