Add image-container phase
This phase runs after image-build and osbuild and can embed an image into a container. JIRA: RHELCMP-3820 Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
This commit is contained in:
Lubomír Sedlář
parent
61e90fd7e0
commit
40133074b3
8 changed files with 502 additions and 11 deletions
262
tests/test_image_container_phase.py
Normal file
262
tests/test_image_container_phase.py
Normal file
|
|
@ -0,0 +1,262 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
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",
|
||||
"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"},
|
||||
"arch_override": "x86_64",
|
||||
}
|
||||
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/x86_64/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"],
|
||||
"arch_override": cfg["arch_override"],
|
||||
"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.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, [])
|
||||
Loading…
Add table
Add a link
Reference in a new issue