215 lines
7.4 KiB
Python
Executable File
215 lines
7.4 KiB
Python
Executable File
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
|
|
from unittest import mock
|
|
import six
|
|
|
|
from pungi.module_util import Modulemd
|
|
from pungi.phases.pkgset import common
|
|
from tests import helpers
|
|
|
|
|
|
class MockCreateRepo(object):
|
|
def __init__(self, createrepo_c):
|
|
self.createrepo_c = createrepo_c
|
|
|
|
def get_createrepo_cmd(self, path_prefix, outputdir, pkglist, **kwargs):
|
|
return (path_prefix, outputdir, pkglist)
|
|
|
|
|
|
@mock.patch("pungi.phases.init.run_in_threads", new=helpers.fake_run_in_threads)
|
|
@mock.patch("pungi.phases.pkgset.common.CreaterepoWrapper", new=MockCreateRepo)
|
|
@mock.patch("pungi.phases.pkgset.common.run")
|
|
class TestMaterializedPkgsetCreate(helpers.PungiTestCase):
|
|
def setUp(self):
|
|
super(TestMaterializedPkgsetCreate, self).setUp()
|
|
self.compose = helpers.DummyCompose(self.topdir, {})
|
|
self.prefix = "/prefix"
|
|
self.pkgset = self._make_pkgset("foo")
|
|
self.subsets = {}
|
|
|
|
def _mk_call(self, arch, name):
|
|
pkglist = "%s.%s.conf" % (arch, name)
|
|
logfile = "arch_repo.%s.%s.log" % (name, arch)
|
|
return mock.call(
|
|
(
|
|
self.prefix,
|
|
os.path.join(self.topdir, "work", arch, "repo", name),
|
|
os.path.join(self.topdir, "work", arch, "package_list", pkglist),
|
|
),
|
|
logfile=os.path.join(self.topdir, "logs", arch, logfile),
|
|
show_cmd=True,
|
|
)
|
|
|
|
def _make_pkgset(self, name):
|
|
pkgset = mock.Mock()
|
|
pkgset.name = name
|
|
pkgset.reuse = None
|
|
|
|
def mock_subset(primary, arch_list, **kwargs):
|
|
self.subsets[primary] = mock.Mock()
|
|
return self.subsets[primary]
|
|
|
|
pkgset.subset.side_effect = mock_subset
|
|
return pkgset
|
|
|
|
def _mk_paths(self, name, arches):
|
|
paths = {"global": os.path.join(self.topdir, "work/global/repo", name)}
|
|
for arch in arches:
|
|
paths[arch] = os.path.join(self.topdir, "work", arch, "repo", name)
|
|
return paths
|
|
|
|
def test_run(self, mock_run):
|
|
result = common.MaterializedPackageSet.create(
|
|
self.compose, self.pkgset, self.prefix
|
|
)
|
|
|
|
six.assertCountEqual(
|
|
self, result.package_sets.keys(), ["global", "amd64", "x86_64"]
|
|
)
|
|
self.assertEqual(result["global"], self.pkgset)
|
|
self.assertEqual(result["x86_64"], self.subsets["x86_64"])
|
|
self.assertEqual(result["amd64"], self.subsets["amd64"])
|
|
|
|
self.pkgset.subset.assert_any_call(
|
|
"x86_64",
|
|
["x86_64", "noarch", "src"],
|
|
exclusive_noarch=True,
|
|
inherit_to_noarch=True,
|
|
)
|
|
self.pkgset.subset.assert_any_call(
|
|
"amd64",
|
|
["amd64", "x86_64", "noarch", "src"],
|
|
exclusive_noarch=True,
|
|
inherit_to_noarch=True,
|
|
)
|
|
|
|
for arch, pkgset in result.package_sets.items():
|
|
pkgset.save_file_list.assed_any_call(
|
|
os.path.join(
|
|
self.topdir, "work", arch, "package_list", arch + ".foo.conf"
|
|
),
|
|
remove_path_prefix=self.prefix,
|
|
)
|
|
|
|
self.assertEqual(result.paths, self._mk_paths("foo", ["amd64", "x86_64"]))
|
|
|
|
mock_run.assert_has_calls(
|
|
[self._mk_call(arch, "foo") for arch in ["global", "amd64", "x86_64"]],
|
|
any_order=True,
|
|
)
|
|
|
|
@helpers.unittest.skipUnless(Modulemd, "Skipping tests, no module support")
|
|
@mock.patch("pungi.phases.pkgset.common.collect_module_defaults")
|
|
@mock.patch("pungi.phases.pkgset.common.collect_module_obsoletes")
|
|
@mock.patch("pungi.phases.pkgset.common.add_modular_metadata")
|
|
def test_run_with_modulemd(self, amm, cmo, cmd, mock_run):
|
|
# Test Index for cmo
|
|
mod_index = Modulemd.ModuleIndex.new()
|
|
mmdobs = Modulemd.Obsoletes.new(
|
|
1, 10993435, "mod_name", "mod_stream", "testmsg"
|
|
)
|
|
mmdobs.set_obsoleted_by("mod_name", "mod_name_2")
|
|
mod_index.add_obsoletes(mmdobs)
|
|
cmo.return_value = mod_index
|
|
|
|
mmd = {
|
|
"x86_64": [
|
|
Modulemd.ModuleStream.new(
|
|
Modulemd.ModuleStreamVersionEnum.TWO, "mod_name", "stream_name"
|
|
)
|
|
]
|
|
}
|
|
common.MaterializedPackageSet.create(
|
|
self.compose, self.pkgset, self.prefix, mmd=mmd
|
|
)
|
|
cmd.assert_called_once_with(
|
|
os.path.join(self.topdir, "work/global/module_defaults"),
|
|
{"mod_name"},
|
|
overrides_dir=None,
|
|
)
|
|
|
|
cmo.assert_called_once()
|
|
cmd.assert_called_once()
|
|
amm.assert_called_once()
|
|
|
|
self.assertEqual(
|
|
amm.mock_calls[0][1][1], os.path.join(self.topdir, "work/x86_64/repo/foo")
|
|
)
|
|
self.assertIsInstance(amm.mock_calls[0][1][2], Modulemd.ModuleIndex)
|
|
self.assertIsNotNone(amm.mock_calls[0][1][2].get_module("mod_name"))
|
|
# Check if proper Index is used by add_modular_metadata
|
|
self.assertIsNotNone(
|
|
amm.mock_calls[0][1][2].get_module("mod_name").get_obsoletes()
|
|
)
|
|
self.assertEqual(
|
|
amm.mock_calls[0][1][3],
|
|
os.path.join(self.topdir, "logs/x86_64/arch_repo_modulemd.foo.x86_64.log"),
|
|
)
|
|
|
|
|
|
class TestCreateArchRepos(helpers.PungiTestCase):
|
|
def setUp(self):
|
|
super(TestCreateArchRepos, self).setUp()
|
|
self.compose = helpers.DummyCompose(self.topdir, {})
|
|
self.prefix = "/prefix"
|
|
self.paths = {}
|
|
self.pkgset = mock.Mock()
|
|
self.pkgset.reuse = None
|
|
self.pkgset.name = "foo"
|
|
|
|
@mock.patch("pungi.phases.pkgset.common._create_arch_repo")
|
|
def test_call_create_arch_repo(self, mock_create):
|
|
common.create_arch_repos(
|
|
self.compose, self.prefix, self.paths, self.pkgset, None
|
|
)
|
|
mock_create.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
mock.ANY,
|
|
(self.compose, "amd64", self.prefix, self.paths, self.pkgset, None),
|
|
1,
|
|
),
|
|
mock.call(
|
|
mock.ANY,
|
|
(
|
|
self.compose,
|
|
"x86_64",
|
|
self.prefix,
|
|
self.paths,
|
|
self.pkgset,
|
|
None,
|
|
),
|
|
2,
|
|
),
|
|
]
|
|
)
|
|
|
|
@mock.patch("pungi.phases.pkgset.common.os.path.isdir", return_value=True)
|
|
@mock.patch("pungi.phases.pkgset.common.copy_all")
|
|
def test_reuse_arch_repo(self, mock_copy_all, mock_isdir):
|
|
self.pkgset.reuse = "/path/to/old/global/repo"
|
|
old_repo = "/path/to/old/repo"
|
|
self.compose.paths.old_compose_path = mock.Mock(return_value=old_repo)
|
|
common.create_arch_repos(
|
|
self.compose, self.prefix, self.paths, self.pkgset, None
|
|
)
|
|
mock_copy_all.assert_has_calls(
|
|
[
|
|
mock.call(
|
|
old_repo, os.path.join(self.compose.topdir, "work/amd64/repo/foo")
|
|
),
|
|
mock.call(
|
|
old_repo, os.path.join(self.compose.topdir, "work/x86_64/repo/foo")
|
|
),
|
|
],
|
|
any_order=True,
|
|
)
|
|
self.compose.log_info.assert_has_calls(
|
|
[
|
|
mock.call("[BEGIN] %s", "Copying repodata for reuse: %s" % old_repo),
|
|
mock.call("[DONE ] %s", "Copying repodata for reuse: %s" % old_repo),
|
|
]
|
|
)
|