# -*- coding: utf-8 -*- from unittest import mock import os from tests import helpers from pungi import checks from pungi.phases import image_container class ImageContainerPhaseTest(helpers.PungiTestCase): @mock.patch("pungi.phases.image_container.ThreadPool") def test_run(self, ThreadPool): cfg = helpers.IterableMock() compose = helpers.DummyCompose( self.topdir, {"image_container": {"^Everything$": cfg}} ) pool = ThreadPool.return_value phase = image_container.ImageContainerPhase(compose) phase.run() self.assertEqual(len(pool.add.call_args_list), 1) self.assertEqual( pool.queue_put.call_args_list, [mock.call((compose, compose.variants["Everything"], cfg))], ) @mock.patch("pungi.phases.image_container.ThreadPool") def test_skip_without_config(self, ThreadPool): compose = helpers.DummyCompose(self.topdir, {}) compose.just_phases = None compose.skip_phases = [] phase = image_container.ImageContainerPhase(compose) self.assertTrue(phase.skip()) class ImageContainerConfigTest(helpers.PungiTestCase): def assertConfigMissing(self, cfg, key): conf = helpers.load_config( helpers.PKGSET_REPOS, **{"image_container": {"^Server$": cfg}} ) errors, warnings = checks.validate(conf, offline=True) self.assertIn( "Failed validation in image_container.^Server$: %r is not valid under any of the given schemas" # noqa: E501 % cfg, errors, ) self.assertIn(" Possible reason: %r is a required property" % key, errors) self.assertEqual([], warnings) def test_correct(self): conf = helpers.load_config( helpers.PKGSET_REPOS, **{ "image_container": { "^Server$": [ { "url": "http://example.com/repo.git#HEAD", "target": "container-candidate", "git_branch": "main", "image_spec": {"type": "qcow2"}, } ] } } ) errors, warnings = checks.validate(conf, offline=True) self.assertEqual([], errors) self.assertEqual([], warnings) def test_missing_url(self): self.assertConfigMissing( { "target": "container-candidate", "git_branch": "main", "image_spec": {"type": "qcow2"}, }, "url", ) def test_missing_target(self): self.assertConfigMissing( { "url": "http://example.com/repo.git#HEAD", "git_branch": "main", "image_spec": {"type": "qcow2"}, }, "target", ) def test_missing_git_branch(self): self.assertConfigMissing( { "url": "http://example.com/repo.git#HEAD", "target": "container-candidate", "image_spec": {"type": "qcow2"}, }, "git_branch", ) def test_missing_image_spec(self): self.assertConfigMissing( { "url": "http://example.com/repo.git#HEAD", "target": "container-candidate", "git_branch": "main", }, "image_spec", ) class ImageContainerThreadTest(helpers.PungiTestCase): def setUp(self): super(ImageContainerThreadTest, self).setUp() self.pool = mock.Mock() self.repofile_path = "work/global/tmp-Server/image-container-Server-1.repo" self.t = image_container.ImageContainerThread(self.pool) self.compose = helpers.DummyCompose( self.topdir, { "koji_profile": "koji", "koji_cache": "/tmp", "translate_paths": [(self.topdir, "http://root")], }, ) self.cfg = { "url": "git://example.com/repo?#BEEFCAFE", "target": "f24-docker-candidate", "git_branch": "f24-docker", "image_spec": {"type": "qcow2"}, } self.compose.im.images["Server"] = { "x86_64": [ mock.Mock(path="Server/x86_64/iso/image.iso", type="iso"), mock.Mock(path="Server/x86_64/images/image.qcow2", type="qcow2"), ] } def _setupMock(self, KojiWrapper): self.wrapper = KojiWrapper.return_value self.wrapper.koji_proxy.buildContainer.return_value = 12345 self.wrapper.watch_task.return_value = 0 def assertRepoFile(self): repofile = os.path.join(self.topdir, self.repofile_path) with open(repofile) as f: repo_content = list(f) self.assertIn("[image-to-include]\n", repo_content) self.assertIn( "baseurl=http://root/compose/Server/$basearch/images/image.qcow2\n", repo_content, ) self.assertIn("enabled=0\n", repo_content) def assertKojiCalls(self, cfg, scratch=False): opts = { "git_branch": cfg["git_branch"], "yum_repourls": ["http://root/" + self.repofile_path], } if scratch: opts["scratch"] = True self.assertEqual( self.wrapper.mock_calls, [ mock.call.login(), mock.call.koji_proxy.buildContainer( cfg["url"], cfg["target"], opts, priority=None, ), mock.call.save_task_id(12345), mock.call.watch_task( 12345, os.path.join( self.topdir, "logs/global/image_container/Server-1-watch-task.log", ), ), ], ) @mock.patch("pungi.phases.image_container.add_metadata") @mock.patch("pungi.phases.image_container.kojiwrapper.KojiWrapper") def test_success(self, KojiWrapper, add_metadata): self._setupMock(KojiWrapper) self.t.process( (self.compose, self.compose.variants["Server"], self.cfg.copy()), 1 ) self.assertRepoFile() self.assertKojiCalls(self.cfg) self.assertEqual( add_metadata.call_args_list, [mock.call(self.compose.variants["Server"], 12345, self.compose, False)], ) @mock.patch("pungi.phases.image_container.add_metadata") @mock.patch("pungi.phases.image_container.kojiwrapper.KojiWrapper") def test_scratch_build(self, KojiWrapper, add_metadata): self.cfg["scratch"] = True self._setupMock(KojiWrapper) self.t.process( (self.compose, self.compose.variants["Server"], self.cfg.copy()), 1 ) self.assertRepoFile() self.assertKojiCalls(self.cfg, scratch=True) self.assertEqual( add_metadata.call_args_list, [mock.call(self.compose.variants["Server"], 12345, self.compose, True)], ) @mock.patch("pungi.phases.image_container.add_metadata") @mock.patch("pungi.phases.image_container.kojiwrapper.KojiWrapper") def test_task_fail(self, KojiWrapper, add_metadata): self._setupMock(KojiWrapper) self.wrapper.watch_task.return_value = 1 with self.assertRaises(RuntimeError) as ctx: self.t.process( (self.compose, self.compose.variants["Server"], self.cfg.copy()), 1 ) self.assertRegex(str(ctx.exception), r"task 12345 failed: see .+ for details") self.assertRepoFile() self.assertKojiCalls(self.cfg) self.assertEqual(add_metadata.call_args_list, []) @mock.patch("pungi.phases.image_container.add_metadata") @mock.patch("pungi.phases.image_container.kojiwrapper.KojiWrapper") def test_task_fail_failable(self, KojiWrapper, add_metadata): self.cfg["failable"] = "*" self._setupMock(KojiWrapper) self.wrapper.watch_task.return_value = 1 self.t.process( (self.compose, self.compose.variants["Server"], self.cfg.copy()), 1 ) self.assertRepoFile() self.assertKojiCalls(self.cfg) self.assertEqual(add_metadata.call_args_list, []) @mock.patch("pungi.phases.image_container.add_metadata") @mock.patch("pungi.phases.image_container.kojiwrapper.KojiWrapper") def test_non_unique_spec(self, KojiWrapper, add_metadata): self.cfg["image_spec"] = {"path": ".*/image\\..*"} self._setupMock(KojiWrapper) with self.assertRaises(RuntimeError) as ctx: self.t.process( (self.compose, self.compose.variants["Server"], self.cfg.copy()), 1 ) self.assertRegex( str(ctx.exception), "2 images matched specification. Only one was expected." ) self.assertEqual(self.wrapper.mock_calls, []) self.assertEqual(add_metadata.call_args_list, [])