Compute checksums in ImageChecksumPhase

The phase goes through all images declared in image manifest, computes
their checksums, stores them in appropriate files and updates the
manifest so that it includes the actual checksums.

The documentation contains details about new configuration options.

The test suite now needs Python's mock package.

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
This commit is contained in:
Lubomír Sedlář 2015-11-20 13:14:24 +01:00
parent 660c8bc2b3
commit 129e654690
4 changed files with 239 additions and 19 deletions

View File

@ -522,6 +522,19 @@ Example
}),
]
Media Checksums Settings
========================
**media_checksums**
(*list*) -- list of checksum types to compute, allowed values are ``md5``,
``sha1`` and ``sha256``
**media_checksum_one_file**
(*bool*) -- when ``True``, only one ``CHECKSUM`` file will be created per
directory; this option requires ``media_checksums`` to only specify one
type
Translate Paths Settings
========================

View File

@ -154,9 +154,6 @@ class CreateisoPhase(PhaseBase):
isomd5sum_cmd = " ".join([pipes.quote(i) for i in isomd5sum_cmd])
cmd["cmd"].append(isomd5sum_cmd)
# compute md5sum, sha1sum, sha256sum
cmd["cmd"].extend(iso.get_checksum_cmds(iso_name))
# create iso manifest
cmd["cmd"].append(iso.get_manifest_cmd(iso_name))

View File

@ -1,30 +1,114 @@
# -*- coding: utf-8 -*-
import os
from kobo import shortcuts
from .base import PhaseBase
MULTIPLE_CHECKSUMS_ERROR = (
'Config option "media_checksum_one_file" requires only one checksum'
' to be configured in "media_checksums".'
)
class ImageChecksumPhase(PhaseBase):
"""Go through images generated in ImageBuild phase and generate their
checksums.
"""Go through images specified in image manifest and generate their
checksums. The manifest will be updated with the checksums.
"""
name = 'image_checksum'
config_options = (
{
"name": "media_checksums",
"expected_types": [list],
"optional": True,
},
{
"name": "media_checksum_one_file",
"expected_types": [bool],
"optional": True,
}
)
def __init__(self, compose):
super(ImageChecksumPhase, self).__init__(compose)
self.checksums = self.compose.conf.get('media_checksums', ['md5', 'sha1', 'sha256'])
self.one_file = self.compose.conf.get('media_checksum_one_file', False)
def validate(self):
errors = []
try:
super(ImageChecksumPhase, self).validate()
except ValueError as exc:
errors = exc.message.split('\n')
if self.one_file and len(self.checksums) != 1:
errors.append(MULTIPLE_CHECKSUMS_ERROR)
if errors:
raise ValueError('\n'.join(errors))
def _get_images(self):
"""Returns a mapping from directories to sets of ``Image``s.
The paths to dirs are absolute.
"""
top_dir = self.compose.paths.compose.topdir()
images = {}
for variant in self.compose.im.images:
for arch in self.compose.im.images[variant]:
for image in self.compose.im.images[variant][arch]:
path = os.path.dirname(os.path.join(top_dir, image.path))
images.setdefault(path, set()).add(image)
return images
def run(self):
compose = self.compose
# merge checksum files
for variant in compose.get_variants(types=["variant", "layered-product"]):
for arch in variant.arches + ["src"]:
iso_dir = compose.paths.compose.iso_dir(arch, variant, create_dir=False)
if not iso_dir or not os.path.exists(iso_dir):
for path, images in self._get_images().iteritems():
checksums = {}
for image in images:
filename = os.path.basename(image.path)
full_path = os.path.join(path, filename)
if not os.path.exists(full_path):
continue
for checksum_type in ("md5", "sha1", "sha256"):
checksum_upper = "%sSUM" % checksum_type.upper()
checksums = sorted([i for i in os.listdir(iso_dir) if i.endswith(".%s" % checksum_upper)])
fo = open(os.path.join(iso_dir, checksum_upper), "w")
for i in checksums:
data = open(os.path.join(iso_dir, i), "r").read()
fo.write(data)
fo.close()
digests = shortcuts.compute_file_checksums(full_path, self.checksums)
for checksum, digest in digests.iteritems():
checksums.setdefault(checksum, {})[filename] = digest
image.add_checksum(None, checksum, digest)
if not self.one_file:
dump_individual(full_path, digest, checksum)
if not checksums:
continue
if self.one_file:
dump_checksums(path, checksums[self.checksums[0]])
else:
for checksum in self.checksums:
dump_checksums(path, checksums[checksum], '%sSUM' % checksum.upper())
def dump_checksums(dir, checksums, filename='CHECKSUM'):
"""Create file with checksums.
:param dir: where to put the file
:param checksums: mapping from filenames to checksums
:param filename: what to call the file
"""
with open(os.path.join(dir, filename), 'w') as f:
for file, checksum in checksums.iteritems():
f.write('{} *{}\n'.format(checksum, file))
def dump_individual(path, checksum, ext):
"""Create a file with a single checksum, saved into a file with an extra
extension.
:param path: path to the checksummed file
:param checksum: the actual digest value
:param ext: what extension to add to the checksum file
"""
with open('%s.%sSUM' % (path, ext.upper()), 'w') as f:
f.write('{} *{}\n'.format(checksum, os.path.basename(path)))

126
tests/test_imagechecksumphase.py Executable file
View File

@ -0,0 +1,126 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import unittest
import mock
import os
import sys
import tempfile
import shutil
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from pungi.phases.image_checksum import (ImageChecksumPhase,
dump_checksums,
dump_individual)
class _DummyCompose(object):
def __init__(self, config):
self.conf = config
self.paths = mock.Mock(
compose=mock.Mock(
topdir=mock.Mock(return_value='/a/b')
)
)
self.image = mock.Mock(
path='Client/i386/iso/image.iso',
)
self.im = mock.Mock(images={'Client': {'i386': [self.image]}})
class TestImageChecksumPhase(unittest.TestCase):
def test_config_skip_individual_with_multiple_algorithms(self):
compose = _DummyCompose({
'media_checksums': ['md5', 'sha1'],
'media_checksum_one_file': True
})
phase = ImageChecksumPhase(compose)
with self.assertRaises(ValueError) as err:
phase.validate()
self.assertIn('media_checksum_one_file', err.message)
@mock.patch('os.path.exists')
@mock.patch('kobo.shortcuts.compute_file_checksums')
@mock.patch('pungi.phases.image_checksum.dump_checksums')
def test_checksum_one_file(self, dump, cc, exists):
compose = _DummyCompose({
'media_checksums': ['sha256'],
'media_checksum_one_file': True,
})
phase = ImageChecksumPhase(compose)
exists.return_value = True
cc.return_value = {'sha256': 'cafebabe'}
phase.run()
dump.assert_called_once_with('/a/b/Client/i386/iso', {'image.iso': 'cafebabe'})
cc.assert_called_once_with('/a/b/Client/i386/iso/image.iso', ['sha256'])
compose.image.add_checksum.assert_called_once_with(None, 'sha256', 'cafebabe')
@mock.patch('os.path.exists')
@mock.patch('kobo.shortcuts.compute_file_checksums')
@mock.patch('pungi.phases.image_checksum.dump_checksums')
@mock.patch('pungi.phases.image_checksum.dump_individual')
def test_checksum_save_individuals(self, indiv_dump, dump, cc, exists):
compose = _DummyCompose({
'media_checksums': ['md5', 'sha256'],
})
phase = ImageChecksumPhase(compose)
exists.return_value = True
cc.return_value = {'md5': 'cafebabe', 'sha256': 'deadbeef'}
phase.run()
indiv_dump.assert_has_calls(
[mock.call('/a/b/Client/i386/iso/image.iso', 'cafebabe', 'md5'),
mock.call('/a/b/Client/i386/iso/image.iso', 'deadbeef', 'sha256')],
any_order=True
)
dump.assert_has_calls(
[mock.call('/a/b/Client/i386/iso', {'image.iso': 'cafebabe'}, 'MD5SUM'),
mock.call('/a/b/Client/i386/iso', {'image.iso': 'deadbeef'}, 'SHA256SUM')],
any_order=True
)
cc.assert_called_once_with('/a/b/Client/i386/iso/image.iso', ['md5', 'sha256'])
compose.image.add_checksum.assert_has_calls([mock.call(None, 'sha256', 'deadbeef'),
mock.call(None, 'md5', 'cafebabe')],
any_order=True)
class TestChecksums(unittest.TestCase):
def setUp(self):
_, name = tempfile.mkstemp()
self.tmp_dir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.tmp_dir)
def test_dump_checksums(self):
dump_checksums(self.tmp_dir, {'file1.iso': 'abcdef', 'file2.iso': 'cafebabe'})
with open(os.path.join(self.tmp_dir, 'CHECKSUM'), 'r') as f:
data = f.read().rstrip().split('\n')
expected = [
'abcdef *file1.iso',
'cafebabe *file2.iso',
]
self.assertItemsEqual(expected, data)
def test_dump_individual(self):
base_path = os.path.join(self.tmp_dir, 'file.iso')
dump_individual(base_path, 'cafebabe', 'md5')
with open(base_path + '.MD5SUM', 'r') as f:
data = f.read()
self.assertEqual('cafebabe *file.iso\n', data)
if __name__ == "__main__":
unittest.main()