almalinux/pungi
13
1
Fork 0
Code Issues Pull Requests Releases Activity

[createiso] Simplify code, test phase

Browse Source 
A couple arguments passed from phase to worker threads were not
duplicated. Only one copy is passed now.

A test case was added both for the phase itself and for worker thread as
well.

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
This commit is contained in:
Lubomír Sedlář 2016-04-06 10:41:07 +02:00
parent df400002d8
commit 3e1a6edecc
3 changed files with 530 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
pungi/phases/createiso.py
View File

@ -30,7 +30,8 @@ from pungi.wrappers.iso import IsoWrapper
from pungi.wrappers.createrepo import CreaterepoWrapper
from pungi.wrappers.kojiwrapper import KojiWrapper
from pungi.phases.base import PhaseBase
from pungi.util import makedirs, get_volid, get_arch_variant_data, failable
from pungi.util import (makedirs, get_volid, get_arch_variant_data, failable,
get_file_size, get_mtime)
from pungi.media_split import MediaSplitter
from pungi.compose_metadata.discinfo import read_discinfo, write_discinfo
@ -87,7 +88,7 @@ class CreateisoPhase(PhaseBase):
if not self._find_rpms(os_tree):
self.compose.log_warning("No RPMs found for %s.%s, skipping ISO"
% (variant, arch))
% (variant.uid, arch))
continue
split_iso_data = split_iso(self.compose, arch, variant)
@ -100,8 +101,6 @@ class CreateisoPhase(PhaseBase):
arch, variant, disc_type=disc_type, disc_num=disc_num)
iso_path = self.compose.paths.compose.iso_path(
arch, variant, filename, symlink_to=symlink_isos_to)
relative_iso_path = self.compose.paths.compose.iso_path(
arch, variant, filename, create_dir=False, relative=True)
if os.path.isfile(iso_path):
self.compose.log_warning("Skipping mkisofs, image already exists: %s" % iso_path)
continue
@ -114,11 +113,7 @@ class CreateisoPhase(PhaseBase):
bootable = self._is_bootable(variant, arch)
cmd = {
"arch": arch,
"variant": variant,
"iso_path": iso_path,
"relative_iso_path": relative_iso_path,
"build_arch": arch,
"bootable": bootable,
"cmd": [],
"label": "", # currently not used
@ -140,10 +135,9 @@ class CreateisoPhase(PhaseBase):
]
if bootable:
cmd['cmd'].extend([
'--bootable',
'--buildinstall-method={}'.format(self.compose.conf['buildinstall_method']),
])
cmd['cmd'].append(
'--buildinstall-method={}'.format(self.compose.conf['buildinstall_method'])
)
if self.compose.supported:
cmd['cmd'].append('--supported')
@ -157,7 +151,8 @@ class CreateisoPhase(PhaseBase):
commands.append((cmd, variant, arch))
self.compose.notifier.send('createiso-targets', deliverables=deliverables)
if self.compose.notifier:
self.compose.notifier.send('createiso-targets', deliverables=deliverables)
for (cmd, variant, arch) in commands:
self.pool.add(CreateIsoThread(self.pool))
@ -172,7 +167,7 @@ class CreateisoPhase(PhaseBase):
class CreateIsoThread(WorkerThread):
def fail(self, compose, cmd):
def fail(self, compose, cmd, variant, arch):
compose.log_error("CreateISO failed, removing ISO: %s" % cmd["iso_path"])
try:
# remove incomplete ISO
@ -180,35 +175,36 @@ class CreateIsoThread(WorkerThread):
# TODO: remove jigdo & template
except OSError:
pass
compose.notifier.send('createiso-imagefail',
file=cmd['iso_path'],
arch=cmd['arch'],
variant=str(cmd['variant']))
if compose.notifier:
compose.notifier.send('createiso-imagefail',
file=cmd['iso_path'],
arch=arch,
variant=str(variant))
def process(self, item, num):
compose, cmd, variant, arch = item
with failable(compose, variant, arch, 'iso', 'Creating ISO'):
self.worker(compose, cmd, num)
self.worker(compose, cmd, variant, arch, num)
def worker(self, compose, cmd, num):
def worker(self, compose, cmd, variant, arch, num):
mounts = [compose.topdir]
if "mount" in cmd:
mounts.append(cmd["mount"])
runroot = compose.conf.get("runroot", False)
bootable = compose.conf.get("bootable", False)
bootable = cmd['bootable']
log_file = compose.paths.log.log_file(
cmd["arch"], "createiso-%s" % os.path.basename(cmd["iso_path"]))
arch, "createiso-%s" % os.path.basename(cmd["iso_path"]))
msg = "Creating ISO (arch: %s, variant: %s): %s" % (
cmd["arch"], cmd["variant"], os.path.basename(cmd["iso_path"]))
arch, variant, os.path.basename(cmd["iso_path"]))
self.pool.log_info("[BEGIN] %s" % msg)
if runroot:
# run in a koji build root
packages = ["coreutils", "genisoimage", "isomd5sum", "jigdo", "strace", "lsof"]
packages = ["coreutils", "genisoimage", "isomd5sum", "jigdo", "pungi"]
extra_packages = {
'lorax': ['lorax', 'pungi'],
'lorax': ['lorax'],
'buildinstall': ['anaconda'],
}
if bootable:
@ -223,16 +219,17 @@ class CreateIsoThread(WorkerThread):
tag_info = koji_proxy.getTag(runroot_tag)
tag_arches = tag_info["arches"].split(" ")
if not cmd["bootable"]:
build_arch = arch
if not bootable:
if "x86_64" in tag_arches:
# assign non-bootable images to x86_64 if possible
cmd["build_arch"] = "x86_64"
elif cmd["build_arch"] == "src":
build_arch = "x86_64"
elif build_arch == "src":
# pick random arch from available runroot tag arches
cmd["build_arch"] = random.choice(tag_arches)
build_arch = random.choice(tag_arches)
koji_cmd = koji_wrapper.get_runroot_cmd(
runroot_tag, cmd["build_arch"], cmd["cmd"],
runroot_tag, build_arch, cmd["cmd"],
channel=runroot_channel, use_shell=True, task_id=True,
packages=packages, mounts=mounts)
@ -242,7 +239,7 @@ class CreateIsoThread(WorkerThread):
output = koji_wrapper.run_runroot_cmd(koji_cmd, log_file=log_file)
if output["retcode"] != 0:
self.fail(compose, cmd)
self.fail(compose, cmd, variant, arch)
raise RuntimeError("Runroot task failed: %s. See %s for more details."
% (output["task_id"], log_file))
@ -251,37 +248,38 @@ class CreateIsoThread(WorkerThread):
try:
run(cmd["cmd"], show_cmd=True, logfile=log_file)
except:
self.fail(compose, cmd)
self.fail(compose, cmd, variant, arch)
raise
iso = IsoWrapper()
img = Image(compose.im)
img.path = cmd["relative_iso_path"]
img.mtime = int(os.stat(cmd["iso_path"]).st_mtime)
img.size = os.path.getsize(cmd["iso_path"])
img.arch = cmd["arch"]
img.path = cmd["iso_path"].replace(compose.paths.compose.topdir(), '').lstrip('/')
img.mtime = get_mtime(cmd["iso_path"])
img.size = get_file_size(cmd["iso_path"])
img.arch = arch
# XXX: HARDCODED
img.type = "dvd"
img.format = "iso"
img.disc_number = cmd["disc_num"]
img.disc_count = cmd["disc_count"]
img.bootable = cmd["bootable"]
img.subvariant = str(cmd['variant'])
img.subvariant = variant.uid
img.implant_md5 = iso.get_implanted_md5(cmd["iso_path"])
try:
img.volume_id = iso.get_volume_id(cmd["iso_path"])
except RuntimeError:
pass
compose.im.add(cmd["variant"].uid, cmd["arch"], img)
compose.im.add(variant.uid, arch, img)
# TODO: supported_iso_bit
# add: boot.iso
self.pool.log_info("[DONE ] %s" % msg)
compose.notifier.send('createiso-imagedone',
file=cmd['iso_path'],
arch=cmd['arch'],
variant=str(cmd['variant']))
if compose.notifier:
compose.notifier.send('createiso-imagedone',
file=cmd['iso_path'],
arch=arch,
variant=str(variant))
def split_iso(compose, arch, variant):

3
tests/helpers.py
View File

@ -63,8 +63,9 @@ class DummyCompose(object):
self.im = mock.Mock(images={'Client': {'i386': [self.image]}})
self.old_composes = []
self.config_dir = '/home/releng/config'
self.notifier = None
def get_variants(self, arch=None, types=None):
def get_variants(self, arch=None, types=None, recursive=None):
return [v for v in self.variants.values() if not arch or arch in v.arches]
def can_fail(self, variant, arch, deliverable):

488
tests/test_createiso_phase.py Executable file
View File

@ -0,0 +1,488 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
import mock
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from tests import helpers
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)
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',
'release_is_layered': False,
'createiso_skip': [
]
})
pool = ThreadPool.return_value
phase = createiso.CreateisoPhase(compose)
phase.run()
self.assertEqual(len(pool.add.call_args_list), 0)
self.assertEqual(pool.queue_put.call_args_list, [])
self.assertItemsEqual(
compose.log_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.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):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'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)
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'])])
self.assertEqual(len(pool.add.call_args_list), 1)
self.maxDiff = None
self.assertItemsEqual(
pool.queue_put.call_args_list,
[mock.call((
compose,
{
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(self.topdir),
'bootable': False,
'cmd': ['pungi-createiso',
'--output-dir={}/compose/Server/x86_64/iso'.format(self.topdir),
'--iso-name=image-name', '--volid=test-1.0 Server.x86_64',
'--graft-points=dummy-graft-points',
'--arch=x86_64', '--supported',
'--jigdo-dir={}/compose/Server/x86_64/jigdo'.format(self.topdir),
'--os-tree={}/compose/Server/x86_64/os'.format(self.topdir)],
'label': '',
'disc_num': 1,
'disc_count': 1,
},
compose.variants['Server'],
'x86_64'
))]
)
@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):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'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)
phase.run()
self.assertItemsEqual(
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)])
self.assertItemsEqual(
split_iso.call_args_list,
[mock.call(compose, 'x86_64', compose.variants['Server']),
mock.call(compose, 'src', compose.variants['Server'])])
self.assertEqual(len(pool.add.call_args_list), 2)
self.maxDiff = None
self.assertItemsEqual(
pool.queue_put.call_args_list,
[mock.call((compose,
{'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(self.topdir),
'bootable': True,
'cmd': ['pungi-createiso',
'--output-dir={}/compose/Server/x86_64/iso'.format(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',
'--jigdo-dir={}/compose/Server/x86_64/jigdo'.format(self.topdir),
'--os-tree={}/compose/Server/x86_64/os'.format(self.topdir)],
'label': '',
'disc_num': 1,
'disc_count': 1},
compose.variants['Server'],
'x86_64')),
mock.call((compose,
{'iso_path': '{}/compose/Server/source/iso/image-name'.format(self.topdir),
'bootable': False,
'cmd': ['pungi-createiso',
'--output-dir={}/compose/Server/source/iso'.format(self.topdir),
'--iso-name=image-name', '--volid=test-1.0 Server.src',
'--graft-points=dummy-graft-points',
'--arch=src', '--supported',
'--jigdo-dir={}/compose/Server/source/jigdo'.format(self.topdir),
'--os-tree={}/compose/Server/source/tree'.format(self.topdir)],
'label': '',
'disc_num': 1,
'disc_count': 1},
compose.variants['Server'],
'src'))]
)
class CreateisoThreadTest(helpers.PungiTestCase):
@mock.patch('pungi.phases.createiso.IsoWrapper')
@mock.patch('pungi.phases.createiso.get_mtime')
@mock.patch('pungi.phases.createiso.get_file_size')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_in_runroot(self, KojiWrapper, get_file_size, get_mtime, IsoWrapper):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': True,
'runroot_tag': 'f25-build',
'koji_profile': 'koji',
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(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=['{}'.format(self.topdir)],
packages=['coreutils', 'genisoimage', 'isomd5sum',
'jigdo', 'pungi'],
task_id=True, use_shell=True)])
self.assertEqual(
run_runroot.call_args_list,
[mock.call(get_runroot_cmd.return_value,
log_file='{}/logs/x86_64/createiso-image-name.x86_64.log'.format(self.topdir))])
self.assertEqual(IsoWrapper.return_value.get_implanted_md5.call_args_list,
[mock.call(cmd['iso_path'])])
self.assertEqual(IsoWrapper.return_value.get_volume_id.call_args_list,
[mock.call(cmd['iso_path'])])
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.IsoWrapper')
@mock.patch('pungi.phases.createiso.get_mtime')
@mock.patch('pungi.phases.createiso.get_file_size')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_bootable(self, KojiWrapper, get_file_size, get_mtime, IsoWrapper):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': True,
'bootable': True,
'buildinstall_method': 'lorax',
'runroot_tag': 'f25-build',
'koji_profile': 'koji',
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(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)
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=['{}'.format(self.topdir)],
packages=['coreutils', 'genisoimage', 'isomd5sum',
'jigdo', 'pungi', 'lorax'],
task_id=True, use_shell=True)])
self.assertEqual(
run_runroot.call_args_list,
[mock.call(get_runroot_cmd.return_value,
log_file='{}/logs/x86_64/createiso-image-name.x86_64.log'.format(self.topdir))])
self.assertEqual(IsoWrapper.return_value.get_implanted_md5.call_args_list,
[mock.call(cmd['iso_path'])])
self.assertEqual(IsoWrapper.return_value.get_volume_id.call_args_list,
[mock.call(cmd['iso_path'])])
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.IsoWrapper')
@mock.patch('pungi.phases.createiso.get_mtime')
@mock.patch('pungi.phases.createiso.get_file_size')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_in_runroot_crash(self, KojiWrapper, get_file_size, get_mtime, IsoWrapper):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': True,
'runroot_tag': 'f25-build',
'koji_profile': 'koji',
'failable_deliverables': [
('^.*$', {'*': 'iso'})
]
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(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
t = createiso.CreateIsoThread(mock.Mock())
with mock.patch('time.sleep'):
t.process((compose, cmd, compose.variants['Server'], 'x86_64'), 1)
compose.log_info.assert_has_calls([
mock.call('[FAIL] Creating ISO (variant Server, arch x86_64) failed, but going on anyway.'),
mock.call('BOOM')
])
@mock.patch('pungi.phases.createiso.IsoWrapper')
@mock.patch('pungi.phases.createiso.get_mtime')
@mock.patch('pungi.phases.createiso.get_file_size')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_in_runroot_fail(self, KojiWrapper, get_file_size, get_mtime, IsoWrapper):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': True,
'runroot_tag': 'f25-build',
'koji_profile': 'koji',
'failable_deliverables': [
('^.*$', {'*': 'iso'})
]
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(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',
}
t = createiso.CreateIsoThread(mock.Mock())
with mock.patch('time.sleep'):
t.process((compose, cmd, compose.variants['Server'], 'x86_64'), 1)
compose.log_info.assert_has_calls([
mock.call('[FAIL] Creating ISO (variant Server, arch x86_64) failed, but going on anyway.'),
mock.call('Runroot task failed: 1234. See {} for more details.'.format(
self.topdir + '/logs/x86_64/createiso-image-name.x86_64.log'))
])
@mock.patch('pungi.phases.createiso.IsoWrapper')
@mock.patch('pungi.phases.createiso.get_mtime')
@mock.patch('pungi.phases.createiso.get_file_size')
@mock.patch('pungi.phases.createiso.run')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_locally(self, KojiWrapper, run, get_file_size, get_mtime, IsoWrapper):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': False,
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(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='{}/logs/x86_64/createiso-image-name.x86_64.log'.format(self.topdir))])
self.assertEqual(IsoWrapper.return_value.get_implanted_md5.call_args_list,
[mock.call(cmd['iso_path'])])
self.assertEqual(IsoWrapper.return_value.get_volume_id.call_args_list,
[mock.call(cmd['iso_path'])])
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.run')
@mock.patch('pungi.phases.createiso.KojiWrapper')
def test_process_locally_crash(self, KojiWrapper, run):
compose = helpers.DummyCompose(self.topdir, {
'release_short': 'test',
'release_version': '1.0',
'release_is_layered': False,
'runroot': False,
'failable_deliverables': [
('^.*$', {'*': 'iso'})
]
})
cmd = {
'iso_path': '{}/compose/Server/x86_64/iso/image-name'.format(self.topdir),
'bootable': False,
'cmd': mock.Mock(),
'label': '',
'disc_num': 1,
'disc_count': 1,
}
run.side_effect = helpers.boom
t = createiso.CreateIsoThread(mock.Mock())
with mock.patch('time.sleep'):
t.process((compose, cmd, compose.variants['Server'], 'x86_64'), 1)
compose.log_info.assert_has_calls([
mock.call('[FAIL] Creating ISO (variant Server, arch x86_64) failed, but going on anyway.'),
mock.call('BOOM')
])
if __name__ == '__main__':
unittest.main()