+         for source_name in ("module", "comps", "json"):

+     for source_name in ("module", "comps", "json"):

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

+ 

+ 

+ # This program is free software; you can redistribute it and/or modify

+ # it under the terms of the GNU General Public License as published by

+ # the Free Software Foundation; version 2 of the License.

+ #

+ # This program is distributed in the hope that it will be useful,

+ # but WITHOUT ANY WARRANTY; without even the implied warranty of

+ # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

+ # GNU Library General Public License for more details.

+ #

+ # You should have received a copy of the GNU General Public License

+ # along with this program; if not, see <https://gnu.org/licenses/>.

+ 

+ 

+ """

+ Get a package list based on modulemd metadata loaded in pkgset phase. Each

+ modulemd file contains a list of exact RPM NEVRAs that should be include, so

+ just go over all modules in a given variant and join all lists together.

+ """

+ 

+ 

+ import pungi.arch

+ import pungi.phases.gather.source

+ 

+ 

+ class GatherSourceModule(pungi.phases.gather.source.GatherSourceBase):

+     enabled = True

+ 

+     def __call__(self, arch, variant):

+         groups = set()

+         packages = set()

+ 

+         # Check if this variant contains some modules

+         if variant is None or variant.modules is None:

+             return packages, groups

+ 

+         compatible_arches = pungi.arch.get_compatible_arches(arch, multilib=True)

+ 

+         for nsvc, mmd in variant.arch_mmds[arch].items():

+             available_rpms = sum(

+                 (

+                     variant.nsvc_to_pkgset[nsvc].rpms_by_arch.get(a, [])

+                     for a in compatible_arches

+                 ),

+                 [],

+             )

+             to_include = set(mmd.get_rpm_artifacts().get())

+             for rpm_obj in available_rpms:

+                 if rpm_obj.nevra in to_include:

+                     packages.add((rpm_obj, None))

+ 

+         return packages, groups

+     def add_fake_module(self, nsvc, rpm_nvrs=None, with_artifacts=False, mmd_arch=None):

+         if mmd_arch:

+             self.arch_mmds.setdefault(mmd_arch, {})[mmd.dup_nsvc()] = mmd

+                          [mock.call(compose.conf['gather_method'])] * 3)

+                 mock.call(compose, 'x86_64', compose.variants['Server'], 'module', pkg_set),

+                        prepopulate=set())] * 3

+                          [mock.call(compose.conf['gather_method'])] * 3)

+                 mock.call(compose, 'x86_64', compose.variants['Server'], 'module', pkg_set),

+                        fulltree_excludes=set(), prepopulate=set())] * 3

+             'gather_method': {'^Server$': {'comps': 'deps', 'module': 'nodeps', 'json': 'deps'}},

+                          [mock.call('nodeps'), mock.call('deps'), mock.call('deps')])

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

+ 

+ try:

+     import unittest2 as unittest

+ except ImportError:

+     import unittest

+ 

+ import mock

+ import os

+ import sys

+ 

+ sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

+ 

+ from pungi.phases.gather.sources.source_module import GatherSourceModule

+ from tests import helpers

+ from pungi import Modulemd

+ 

+ 

+ @unittest.skipUnless(Modulemd is not None, "Skipped test, no module support.")

+ class TestGatherSourceModule(helpers.PungiTestCase):

+     def setUp(self):

+         super(TestGatherSourceModule, self).setUp()

+         self.compose = helpers.DummyCompose(self.topdir, {})

+ 

+     def _add_pkg(self, arch):

+         mock_rpm = mock.Mock(

+             version="1.0.0",

+             release="1",

+             epoch=0,

+             excludearch=None,

+             exclusivearch=None,

+             sourcerpm="pkg-1.0.0-1",

+             nevra="pkg-0:1.0.0-1.%s" % arch,

+             arch=arch,

+         )

+         mock_rpm.name = "pkg"

+         self.compose.variants["Server"].nsvc_to_pkgset[

+             "testmodule:master:1:2017"

+         ].rpms_by_arch[arch] = [mock_rpm]

+ 

+     def test_without_modules(self):

+         source = GatherSourceModule(self.compose)

+         packages, groups = source("x86_64", self.compose.variants["Server"])

+         self.assertItemsEqual(packages, [])

+         self.assertItemsEqual(groups, [])

+ 

+     def test_include_two_packages(self):

+         self.compose.variants["Server"].add_fake_module(

+             "testmodule:master:1:2017",

+             rpm_nvrs=["pkg-0:1.0.0-1.x86_64", "pkg-0:1.0.0-1.i686"],

+             with_artifacts=True,

+             mmd_arch="x86_64",

+         )

+ 

+         self._add_pkg("x86_64")

+         self._add_pkg("i686")

+ 

+         source = GatherSourceModule(self.compose)

+         packages, groups = source("x86_64", self.compose.variants["Server"])

+         self.assertItemsEqual(

+             [(rpm[0].nevra, rpm[1]) for rpm in packages],

+             [("pkg-0:1.0.0-1.x86_64", None), ("pkg-0:1.0.0-1.i686", None)],

+         )

+         self.assertItemsEqual(groups, [])

+ 

+     def test_does_not_include_unlisted(self):

+         self.compose.variants["Server"].add_fake_module(

+             "testmodule:master:1:2017",

+             rpm_nvrs=[],

+             with_artifacts=True,

+             mmd_arch="x86_64",

+         )

+ 

+         self._add_pkg("x86_64")

+ 

+         source = GatherSourceModule(self.compose)

+         packages, groups = source("x86_64", self.compose.variants["Server"])

+         self.assertItemsEqual(packages, [])

+         self.assertItemsEqual(groups, [])