[test] Add checks for created images

The performed checks:
 * If format is ISO, the file must have correct magic string
 * If it's bootable, there must be MBR or GPT

When a check fails on any failable deliverable, it will be logged and
the file removed from metadata (it will still remain on the disk). This
required a change to write the images.json file later (after test
phase).

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>

bin/pungi-koji

@@ -360,12 +360,13 @@ def run_compose(compose):
     image_checksum_phase.stop()
 
     pungi.metadata.write_compose_info(compose)
-    compose.im.dump(compose.paths.compose.metadata("images.json"))
 
     # TEST phase
     test_phase.start()
     test_phase.stop()
 
+    compose.im.dump(compose.paths.compose.metadata("images.json"))
+
     # create a latest symlink
     compose_dir = os.path.basename(compose.topdir)
     symlink_name = "latest-%s-%s" % (compose.conf["release_short"], ".".join(compose.conf["release_version"].split(".")[:-1]))

pungi/phases/buildinstall.py

@@ -370,6 +370,7 @@ def link_boot_iso(compose, arch, variant):
     img.bootable = True
     img.subvariant = variant.name
     img.implant_md5 = implant_md5
+    setattr(img, 'deliverable', 'buildinstall')
     try:
         img.volume_id = iso.get_volume_id(new_boot_iso_path)
     except RuntimeError:

pungi/phases/createiso.py

@@ -266,6 +266,7 @@ class CreateIsoThread(WorkerThread):
         img.bootable = cmd["bootable"]
         img.subvariant = variant.uid
         img.implant_md5 = iso.get_implanted_md5(cmd["iso_path"])
+        setattr(img, 'deliverable', 'iso')
         try:
             img.volume_id = iso.get_volume_id(cmd["iso_path"])
         except RuntimeError:

pungi/phases/image_build.py

@@ -251,6 +251,7 @@ class CreateImageBuildThread(WorkerThread):
             img.disc_count = 1
             img.bootable = False
             img.subvariant = subvariant
+            setattr(img, 'deliverable', 'image-build')
             compose.im.add(variant=variant.uid, arch=image_info['arch'], image=img)
 
         self.pool.log_info("[DONE ] %s" % msg)

pungi/phases/live_images.py

@@ -290,6 +290,7 @@ class CreateLiveImageThread(WorkerThread):
         img.disc_count = 1
         img.bootable = True
         img.subvariant = subvariant
+        setattr(img, 'deliverable', 'live')
         compose.im.add(variant=variant.uid, arch=arch, image=img)
 
     def _is_image(self, path):

pungi/phases/livemedia_phase.py

@@ -204,6 +204,7 @@ class LiveMediaThread(WorkerThread):
             img.disc_count = 1
             img.bootable = True
             img.subvariant = subvariant
+            setattr(img, 'deliverable', 'live-media')
             compose.im.add(variant=variant.uid, arch=image_info['arch'], image=img)
 
         self.pool.log_info('[DONE ] %s' % msg)

pungi/phases/ostree_installer.py

@@ -108,6 +108,7 @@ class OstreeInstallerThread(WorkerThread):
         img.bootable = True
         img.subvariant = variant.name
         img.implant_md5 = implant_md5
+        setattr(img, 'deliverable', 'ostree-installer')
         try:
             img.volume_id = iso_wrapper.get_volume_id(full_iso_path)
         except RuntimeError:

pungi/phases/test.py

@@ -16,6 +16,7 @@
 
 
 import tempfile
+import os
 
 from kobo.shortcuts import run
 
@@ -23,7 +24,7 @@ from pungi.wrappers.repoclosure import RepoclosureWrapper
 from pungi.arch import get_valid_arches
 from pungi.phases.base import PhaseBase
 from pungi.phases.gather import get_lookaside_repos
-from pungi.util import rmtree, is_arch_multilib
+from pungi.util import rmtree, is_arch_multilib, failable
 
 
 class TestPhase(PhaseBase):
@@ -31,6 +32,7 @@ class TestPhase(PhaseBase):
 
     def run(self):
         run_repoclosure(self.compose)
+        check_image_sanity(self.compose)
 
 
 def run_repoclosure(compose):
@@ -99,3 +101,53 @@ def run_repoclosure(compose):
         rmtree(tmp_dir)
 
     compose.log_info("[DONE ] %s" % msg)
+
+
+def check_image_sanity(compose):
+    """
+    Go through all images in manifest and make basic sanity tests on them. If
+    any check fails for a failable deliverable, it will be removed from
+    manifest and logged. Otherwise the compose will be aborted.
+    """
+    im = compose.im
+    for variant_uid in im.images:
+        variant = compose.variants[variant_uid]
+        for arch in im.images[variant_uid]:
+            images = im.images[variant_uid][arch]
+            im.images[variant_uid][arch] = [img for img in images
+                                            if check(compose, variant, arch, img)]
+
+
+def check(compose, variant, arch, image):
+    result = True
+    path = os.path.join(compose.paths.compose.topdir(), image.path)
+    deliverable = getattr(image, 'deliverable')
+    with failable(compose, variant, arch, deliverable, subvariant=image.subvariant):
+        with open(path) as f:
+            if image.format == 'iso' and not is_iso(f):
+                result = False
+                raise RuntimeError('%s does not look like an ISO file' % path)
+            if image.bootable and not has_mbr(f) and not has_gpt(f):
+                result = False
+                raise RuntimeError(
+                    '%s is supposed to be bootable, but does not have MBR nor GPT' % path)
+    # If exception is raised above, failable may catch it
+    return result
+
+
+def _check_magic(f, offset, bytes):
+    """Check that the file has correct magic number at correct offset."""
+    f.seek(offset)
+    return f.read(len(bytes)) == bytes
+
+
+def is_iso(f):
+    return _check_magic(f, 0x8001, 'CD001')
+
+
+def has_mbr(f):
+    return _check_magic(f, 0x1fe, '\x55\xAA')
+
+
+def has_gpt(f):
+    return _check_magic(f, 0x200, 'EFI PART')

tests/test_test_phase.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+
+import unittest
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import pungi.phases.test as test_phase
+from tests.helpers import DummyCompose, PungiTestCase, touch
+
+
+FAILABLE_CONFIG = {
+    'failable_deliverables': [
+        ('^.+$', {'*': ['iso']}),
+    ]
+}
+
+UNBOOTABLE_ISO = ('\0' * 0x8001) + 'CD001' + ('\0' * 100)
+ISO_WITH_MBR = ('\0' * 0x1fe) + '\x55\xAA' + ('\0' * 0x7e01) + 'CD001' + ('\0' * 100)
+ISO_WITH_GPT = ('\0' * 0x200) + 'EFI PART' + ('\0' * 0x7df9) + 'CD001' + ('\0' * 100)
+ISO_WITH_MBR_AND_GPT = ('\0' * 0x1fe) + '\x55\xAAEFI PART' + ('\0' * 0x7df9) + 'CD001' + ('\0' * 100)
+
+
+class TestCheckImageSanity(PungiTestCase):
+
+    def test_missing_file_reports_error(self):
+        compose = DummyCompose(self.topdir, {})
+
+        with self.assertRaises(IOError):
+            test_phase.check_image_sanity(compose)
+
+    def test_missing_file_doesnt_report_if_failable(self):
+        compose = DummyCompose(self.topdir, FAILABLE_CONFIG)
+        compose.image.deliverable = 'iso'
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Failable deliverable must not raise')
+
+    def test_correct_iso_does_not_raise(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = False
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), UNBOOTABLE_ISO)
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Correct unbootable image must not raise')
+
+    def test_incorrect_iso_raises(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = False
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), 'Hey there')
+
+        with self.assertRaises(RuntimeError) as ctx:
+            test_phase.check_image_sanity(compose)
+
+        self.assertIn('does not look like an ISO file', str(ctx.exception))
+
+    def test_bootable_iso_without_mbr_or_gpt_raises(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = True
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), UNBOOTABLE_ISO)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            test_phase.check_image_sanity(compose)
+
+        self.assertIn('is supposed to be bootable, but does not have MBR nor GPT',
+                      str(ctx.exception))
+
+    def test_failable_bootable_iso_without_mbr_gpt_doesnt_raise(self):
+        compose = DummyCompose(self.topdir, FAILABLE_CONFIG)
+        compose.image.format = 'iso'
+        compose.image.bootable = True
+        compose.image.deliverable = 'iso'
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), UNBOOTABLE_ISO)
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Failable deliverable must not raise')
+
+    def test_bootable_iso_with_mbr_does_not_raise(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = True
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), ISO_WITH_MBR)
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Bootable image with MBR must not raise')
+
+    def test_bootable_iso_with_gpt_does_not_raise(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = True
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), ISO_WITH_GPT)
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Bootable image with GPT must not raise')
+
+    def test_bootable_iso_with_mbr_and_gpt_does_not_raise(self):
+        compose = DummyCompose(self.topdir, {})
+        compose.image.format = 'iso'
+        compose.image.bootable = True
+        touch(os.path.join(self.topdir, 'compose', compose.image.path), ISO_WITH_MBR_AND_GPT)
+
+        try:
+            test_phase.check_image_sanity(compose)
+        except:
+            self.fail('Bootable image with MBR and GPT must not raise')
+
+
+if __name__ == "__main__":
+    unittest.main()