pungi/tests/test_compose.py

# -*- coding: utf-8 -*-

import logging
import mock

try:
    import unittest2 as unittest
except ImportError:
    import unittest
import os
import six
import tempfile
import shutil

from pungi.compose import Compose


class ConfigWrapper(dict):
    def __init__(self, *args, **kwargs):
        super(ConfigWrapper, self).__init__(*args, **kwargs)
        self._open_file = "%s/fixtures/config.conf" % os.path.abspath(
            os.path.dirname(__file__)
        )


class ComposeTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp_dir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self.tmp_dir)

    @mock.patch("pungi.compose.ComposeInfo")
    def test_setup_logger(self, ci):
        conf = {}
        logger = logging.getLogger("test_setup_logger")
        compose = Compose(conf, self.tmp_dir, logger=logger)
        self.assertEqual(len(logger.handlers), 2)

        pungi_log = logger.handlers[0].stream.name
        exclude_arch_log = logger.handlers[1].stream.name
        self.assertEqual(os.path.basename(pungi_log), "pungi.global.log")
        self.assertEqual(
            os.path.basename(exclude_arch_log), "excluding-arch.global.log"
        )

        msg = "test log"
        compose.log_info(msg)
        with open(pungi_log) as f:
            self.assertTrue(msg in f.read())
        with open(exclude_arch_log) as f:
            self.assertTrue(msg not in f.read())

        msg = "Populating package set for arch: x86_64"
        compose.log_info(msg)
        with open(exclude_arch_log) as f:
            self.assertTrue(msg in f.read())

    @mock.patch("pungi.compose.ComposeInfo")
    def test_can_fail(self, ci):
        conf = {
            "failable_deliverables": [
                (
                    "^.*$",
                    {"*": ["buildinstall"], "i386": ["buildinstall", "live", "iso"]},
                ),
            ]
        }
        compose = Compose(conf, self.tmp_dir)
        variant = mock.Mock(uid="Server")

        self.assertTrue(compose.can_fail(variant, "x86_64", "buildinstall"))
        self.assertFalse(compose.can_fail(variant, "x86_64", "live"))
        self.assertTrue(compose.can_fail(variant, "i386", "live"))

        self.assertFalse(compose.can_fail(None, "x86_64", "live"))
        self.assertTrue(compose.can_fail(None, "i386", "live"))

        self.assertTrue(compose.can_fail(variant, "*", "buildinstall"))
        self.assertFalse(compose.can_fail(variant, "*", "live"))

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_image_name(self, ci):
        conf = {}
        variant = mock.Mock(uid="Server", type="variant")
        ci.return_value.compose.respin = 2
        ci.return_value.compose.id = "compose_id"
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "RC-1.0"
        ci.return_value.compose.label_major_version = "1"

        ci.return_value.release.version = "3.0"
        ci.return_value.release.short = "rel_short"

        compose = Compose(conf, self.tmp_dir)

        keys = [
            "arch",
            "compose_id",
            "date",
            "disc_num",
            "disc_type",
            "label",
            "label_major_version",
            "release_short",
            "respin",
            "suffix",
            "type",
            "type_suffix",
            "variant",
            "version",
        ]
        format = "-".join(["%(" + k + ")s" for k in keys])
        name = compose.get_image_name(
            "x86_64",
            variant,
            format=format,
            disc_num=7,
            disc_type="live",
            suffix=".iso",
        )

        self.assertEqual(
            name,
            "-".join(
                [
                    "x86_64",
                    "compose_id",
                    "20160107",
                    "7",
                    "live",
                    "RC-1.0",
                    "1",
                    "rel_short",
                    "2",
                    ".iso",
                    "nightly",
                    ".n",
                    "Server",
                    "3.0",
                ]
            ),
        )

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_image_name_variant_mapping(self, ci):
        conf = {"image_name_format": {"^Server$": "whatever"}}
        variant = mock.Mock(uid="Server", type="variant")

        compose = Compose(conf, self.tmp_dir)

        name = compose.get_image_name(
            "x86_64", variant, disc_num=7, disc_type="live", suffix=".iso"
        )

        self.assertEqual(name, "whatever")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_image_name_variant_mapping_no_match(self, ci):
        conf = {"image_name_format": {"^Client$": "whatever"}}
        variant = mock.Mock(uid="Server", type="variant")
        ci.return_value.compose.id = "compose_id"

        compose = Compose(conf, self.tmp_dir)
        name = compose.get_image_name(
            "x86_64", variant, disc_num=7, disc_type="live", suffix=".iso"
        )

        self.assertEqual(name, "compose_id-Server-x86_64-live7.iso")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_image_name_layered_product(self, ci):
        conf = {}
        variant = mock.Mock(uid="Server-LP", type="layered-product")
        variant.parent = mock.Mock(uid="Server")
        ci.return_value.compose.respin = 2
        ci.return_value.compose.id = "compose_id"
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "RC-1.0"
        ci.return_value.compose.label_major_version = "1"

        ci.return_value.release.version = "3.0"
        ci.return_value.release.short = "rel_short"

        ci.return_value["Server-LP"].compose_id = "Gluster 1.0"

        compose = Compose(conf, self.tmp_dir)

        format = "{compose_id} {variant}"
        name = compose.get_image_name(
            "x86_64",
            variant,
            format=format,
            disc_num=7,
            disc_type="live",
            suffix=".iso",
        )

        self.assertEqual(name, "Gluster 1.0 Server")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_image_name_type_netinst(self, ci):
        conf = {}
        variant = mock.Mock(uid="Server", type="variant")
        ci.return_value.compose.respin = 2
        ci.return_value.compose.id = "compose_id"
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "RC-1.0"
        ci.return_value.compose.label_major_version = "1"

        ci.return_value.release.version = "3.0"
        ci.return_value.release.short = "rel_short"

        compose = Compose(conf, self.tmp_dir)

        keys = [
            "arch",
            "compose_id",
            "date",
            "disc_num",
            "disc_type",
            "label",
            "label_major_version",
            "release_short",
            "respin",
            "suffix",
            "type",
            "type_suffix",
            "variant",
            "version",
        ]
        format = "-".join(["%(" + k + ")s" for k in keys])
        name = compose.get_image_name(
            "x86_64",
            variant,
            format=format,
            disc_num=7,
            disc_type="netinst",
            suffix=".iso",
        )

        self.assertEqual(
            name,
            "-".join(
                [
                    "x86_64",
                    "compose_id",
                    "20160107",
                    "7",
                    "netinst",
                    "RC-1.0",
                    "1",
                    "rel_short",
                    "2",
                    ".iso",
                    "nightly",
                    ".n",
                    "Server",
                    "3.0",
                ]
            ),
        )

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_release(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = None

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_release, "20160107.n.2")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_release_production(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "production"
        ci.return_value.compose.type_suffix = ""
        ci.return_value.compose.label = None

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_release, "20160107.2")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_release_from_label(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "production"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "Alpha-1.2"

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_release, "1.2")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_version_without_label(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = None
        ci.return_value.release.version = "25"

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_version, "25")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_version_with_label(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "Alpha-1.2"
        ci.return_value.release.version = "25"

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_version, "25_Alpha")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_image_version_with_label_rc(self, ci):
        conf = {}
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "nightly"
        ci.return_value.compose.type_suffix = ".n"
        ci.return_value.compose.label = "RC-1.2"
        ci.return_value.release.version = "25"

        compose = Compose(conf, self.tmp_dir)

        self.assertEqual(compose.image_version, "25")

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_variant_arches_without_filter(self, ci):
        ci.return_value.compose.id = "composeid"

        conf = ConfigWrapper(
            variants_file={"scm": "file", "repo": None, "file": "variants.xml"},
            release_name="Test",
            release_version="1.0",
            release_short="test",
            release_type="ga",
            release_internal=False,
        )

        compose = Compose(conf, self.tmp_dir)
        compose.read_variants()

        self.assertEqual(
            sorted(v.uid for v in compose.variants.values()),
            ["Client", "Crashy", "Live", "Server"],
        )
        self.assertEqual(
            sorted(v.uid for v in compose.variants["Server"].variants.values()),
            ["Server-Gluster", "Server-ResilientStorage", "Server-optional"],
        )
        six.assertCountEqual(
            self, compose.variants["Client"].arches, ["i386", "x86_64"]
        )
        self.assertEqual(compose.variants["Crashy"].arches, ["ppc64le"])
        self.assertEqual(compose.variants["Live"].arches, ["x86_64"])
        six.assertCountEqual(
            self, compose.variants["Server"].arches, ["s390x", "x86_64"]
        )
        self.assertEqual(
            compose.variants["Server"].variants["Gluster"].arches, ["x86_64"]
        )
        self.assertEqual(
            compose.variants["Server"].variants["ResilientStorage"].arches, ["x86_64"]
        )
        six.assertCountEqual(
            self,
            compose.variants["Server"].variants["optional"].arches,
            ["s390x", "x86_64"],
        )

        self.assertEqual(
            [v.uid for v in compose.get_variants()],
            [
                "Client",
                "Crashy",
                "Live",
                "Server",
                "Server-Gluster",
                "Server-ResilientStorage",
                "Server-optional",
            ],
        )
        self.assertEqual(compose.get_arches(), ["i386", "ppc64le", "s390x", "x86_64"])

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_variant_arches_with_arch_filter(self, ci):
        ci.return_value.compose.id = "composeid"

        conf = ConfigWrapper(
            variants_file={"scm": "file", "repo": None, "file": "variants.xml"},
            release_name="Test",
            release_version="1.0",
            release_short="test",
            release_type="ga",
            release_internal=False,
            tree_arches=["x86_64"],
        )

        compose = Compose(conf, self.tmp_dir)
        compose.read_variants()

        self.assertEqual(
            sorted(v.uid for v in compose.variants.values()),
            ["Client", "Live", "Server"],
        )
        self.assertEqual(
            sorted(v.uid for v in compose.variants["Server"].variants.values()),
            ["Server-Gluster", "Server-ResilientStorage", "Server-optional"],
        )
        self.assertEqual(compose.variants["Client"].arches, ["x86_64"])
        self.assertEqual(compose.variants["Live"].arches, ["x86_64"])
        self.assertEqual(compose.variants["Server"].arches, ["x86_64"])
        self.assertEqual(
            compose.variants["Server"].variants["Gluster"].arches, ["x86_64"]
        )
        self.assertEqual(
            compose.variants["Server"].variants["ResilientStorage"].arches, ["x86_64"]
        )
        self.assertEqual(
            compose.variants["Server"].variants["optional"].arches, ["x86_64"]
        )

        self.assertEqual(compose.get_arches(), ["x86_64"])
        self.assertEqual(
            [v.uid for v in compose.get_variants()],
            [
                "Client",
                "Live",
                "Server",
                "Server-Gluster",
                "Server-ResilientStorage",
                "Server-optional",
            ],
        )

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_variant_arches_with_variant_filter(self, ci):
        ci.return_value.compose.id = "composeid"
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "production"
        ci.return_value.compose.type_suffix = ".n"

        conf = ConfigWrapper(
            variants_file={"scm": "file", "repo": None, "file": "variants.xml"},
            release_name="Test",
            release_version="1.0",
            release_short="test",
            release_type="ga",
            release_internal=False,
            tree_variants=["Server", "Client", "Server-Gluster"],
        )

        compose = Compose(conf, self.tmp_dir)
        compose.read_variants()

        self.assertEqual(
            sorted(v.uid for v in compose.variants.values()), ["Client", "Server"]
        )
        six.assertCountEqual(
            self, compose.variants["Client"].arches, ["i386", "x86_64"]
        )
        six.assertCountEqual(
            self, compose.variants["Server"].arches, ["s390x", "x86_64"]
        )
        self.assertEqual(
            compose.variants["Server"].variants["Gluster"].arches, ["x86_64"]
        )

        self.assertEqual(compose.get_arches(), ["i386", "s390x", "x86_64"])
        self.assertEqual(
            [v.uid for v in compose.get_variants()],
            ["Client", "Server", "Server-Gluster"],
        )

    @mock.patch("pungi.compose.ComposeInfo")
    def test_get_variant_arches_with_both_filters(self, ci):
        ci.return_value.compose.id = "composeid"
        ci.return_value.compose.respin = 2
        ci.return_value.compose.date = "20160107"
        ci.return_value.compose.type = "production"
        ci.return_value.compose.type_suffix = ".n"

        logger = mock.Mock()
        logger.handlers = []

        conf = ConfigWrapper(
            variants_file={"scm": "file", "repo": None, "file": "variants.xml"},
            release_name="Test",
            release_version="1.0",
            release_short="test",
            release_type="ga",
            release_internal=False,
            tree_variants=["Server", "Client", "Server-optional"],
            tree_arches=["x86_64"],
        )

        compose = Compose(conf, self.tmp_dir, logger=logger)
        compose.read_variants()

        self.assertEqual(
            sorted(v.uid for v in compose.variants.values()), ["Client", "Server"]
        )
        self.assertEqual(compose.variants["Client"].arches, ["x86_64"])
        self.assertEqual(compose.variants["Server"].arches, ["x86_64"])
        self.assertEqual(
            compose.variants["Server"].variants["optional"].arches, ["x86_64"]
        )

        self.assertEqual(compose.get_arches(), ["x86_64"])
        self.assertEqual(
            [v.uid for v in compose.get_variants()],
            ["Client", "Server", "Server-optional"],
        )

        six.assertCountEqual(
            self,
            logger.info.call_args_list,
            [
                mock.call("Excluding variant Live: filtered by configuration."),
                mock.call("Excluding variant Crashy: all its arches are filtered."),
                mock.call(
                    "Excluding variant Server-ResilientStorage: filtered by configuration."  # noqa: E501
                ),
                mock.call(
                    "Excluding variant Server-Gluster: filtered by configuration."
                ),
            ],
        )

    @mock.patch("pungi.compose.ComposeInfo")
    def test_mkdtemp(self, ci):
        ci.return_value.compose.id = "composeid"
        conf = ConfigWrapper(
            variants_file={"scm": "file", "repo": None, "file": "variants.xml"},
            release_name="Test",
            release_version="1.0",
            release_short="test",
            release_type="ga",
            release_internal=False,
            tree_variants=["Server", "Client", "Server-optional"],
            tree_arches=["x86_64"],
        )
        compose = Compose(conf, self.tmp_dir)
        d = compose.mkdtemp()
        self.assertTrue(os.path.isdir(d))
        d = compose.mkdtemp(prefix="tweak_buildinstall")
        self.assertTrue(os.path.isdir(d))


class StatusTest(unittest.TestCase):
    def setUp(self):
        self.tmp_dir = tempfile.mkdtemp()
        self.logger = mock.Mock()
        self.logger.handlers = []
        with mock.patch("pungi.compose.ComposeInfo"):
            self.compose = Compose({}, self.tmp_dir, logger=self.logger)

    def tearDown(self):
        shutil.rmtree(self.tmp_dir)

    def test_get_status_non_existing(self):
        status = self.compose.get_status()
        self.assertIsNone(status)

    def test_get_status_existing(self):
        with open(os.path.join(self.tmp_dir, "STATUS"), "w") as f:
            f.write("FOOBAR")

        self.assertEqual(self.compose.get_status(), "FOOBAR")

    def test_get_status_is_dir(self):
        os.mkdir(os.path.join(self.tmp_dir, "STATUS"))

        self.assertIsNone(self.compose.get_status())

    def test_write_status(self):
        self.compose.write_status("DOOMED")

        with open(os.path.join(self.tmp_dir, "STATUS"), "r") as f:
            self.assertEqual(f.read(), "DOOMED\n")

    def test_write_non_standard_status(self):
        self.compose.write_status("FOOBAR")

        self.assertEqual(self.logger.log.call_count, 1)
        with open(os.path.join(self.tmp_dir, "STATUS"), "r") as f:
            self.assertEqual(f.read(), "FOOBAR\n")

    def test_write_status_on_finished(self):
        self.compose.write_status("FINISHED")

        with self.assertRaises(RuntimeError):
            self.compose.write_status("NOT REALLY")

    def test_write_status_with_failed_deliverables(self):
        self.compose.conf = {
            "failable_deliverables": [("^.+$", {"*": ["live", "build-image"]})]
        }

        variant = mock.Mock(uid="Server")
        self.compose.fail_deliverable(variant, "x86_64", "live")
        self.compose.fail_deliverable(None, "*", "build-image")

        self.compose.write_status("FINISHED")

        self.logger.log.assert_has_calls(
            [
                mock.call(
                    20, "Failed build-image on variant <>, arch <*>, subvariant <None>."
                ),
                mock.call(
                    20,
                    "Failed live on variant <Server>, arch <x86_64>, subvariant <None>.",  # noqa: E501
                ),
            ],
            any_order=True,
        )

        with open(os.path.join(self.tmp_dir, "STATUS"), "r") as f:
            self.assertEqual(f.read(), "FINISHED_INCOMPLETE\n")

    def test_calls_notifier(self):
        self.compose.notifier = mock.Mock()
        self.compose.write_status("FINISHED")

        self.assertTrue(self.compose.notifier.send.call_count, 1)

    def test_no_database_with_dnf_backend(self):
        self.compose.conf["gather_backend"] = "dnf"
        self.assertFalse(self.compose.should_create_yum_database)

    def test_no_database_with_dnf_backend_config_override(self):
        self.compose.conf["gather_backend"] = "dnf"
        self.compose.conf["createrepo_database"] = True
        self.assertTrue(self.compose.should_create_yum_database)

    def test_no_database_with_yum_backend(self):
        self.compose.conf["gather_backend"] = "yum"
        self.assertTrue(self.compose.should_create_yum_database)

    def test_no_database_with_yum_backend_config_override(self):
        self.compose.conf["gather_backend"] = "yum"
        self.compose.conf["createrepo_database"] = False
        self.assertFalse(self.compose.should_create_yum_database)