almalinux/pungi
14
1
Fork 0
Code Issues Pull Requests Releases Activity
6080b45178
pungi/tests/test_gather_method_hybrid.py

585 lines
19 KiB
Python
Raw Normal View History

gather: Add a hybrid depsolver backend This patch adds a new gather method called `hybrid`, which uses a `fus` binary, which must exist somewhere on the `$PATH`. It will call it multiple times to add multilib packages. The solver can handle packages, modules and comps groups as input. However comps groups are never passed in. Pungi will expand it to a list of packages to avoid issues with comps handling in fus. It ignores optional packages, and if the group mentions a package that does not exist, nothing else from the group is included. Multilib is also handled outside of fus. Pungi will run it, parse the packages from output, determines multilib packages and adds them as input. Then it runs the solver again. This is done until nothing new is added. Usually two passes should be enough. Source packages and debuginfo are added as a final step. All debuginfo packages from any included source are added. If the source or debuginfo package is included in any lookaside repo, it will be skipped. The tool expects to get a platform stream that should be provided for modules to depend on. Pungi looks into the modules and gets the platform from there. If there are more requests, an error is raised. There is some missing functionality and options that are ignored. Particularly these are: * gather_fulltree * gather_selfhosting * greedy_method Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
2018-06-22 06:58:54 +00:00
# -*- coding: utf-8 -*-
from collections import namedtuple
import copy
import mock
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from pungi.phases.gather.methods import method_hybrid as hybrid
from tests import helpers
MockPkg = namedtuple(
"MockPkg", ["name", "version", "release", "epoch", "sourcerpm", "file_path", "arch"]
)
class NamedMock(mock.Mock):
def __init__(self, name=None, **kwargs):
super(NamedMock, self).__init__(**kwargs)
self.name = name
class TestMethodHybrid(helpers.PungiTestCase):
@mock.patch("pungi.phases.gather.get_lookaside_repos")
@mock.patch("pungi.phases.gather.methods.method_hybrid.expand_groups")
@mock.patch("pungi.phases.gather.methods.method_hybrid.expand_packages")
@mock.patch("pungi.phases.gather.methods.method_hybrid.create_module_repo")
def test_call_method(self, cmr, ep, eg, glr):
compose = helpers.DummyCompose(self.topdir, {})
cmr.return_value = (mock.Mock(), mock.Mock())
m = hybrid.GatherMethodHybrid(compose)
m.run_solver = mock.Mock(return_value=(mock.Mock(), mock.Mock()))
pkg = MockPkg(
name="pkg",
version="1",
release="2",
arch="x86_64",
epoch=3,
sourcerpm=None,
file_path=None,
)
eg.return_value = ["foo", "bar"]
package_sets = {"x86_64": mock.Mock(rpms_by_arch={"x86_64": [pkg]})}
arch = "x86_64"
variant = compose.variants["Server"]
res = m(arch, variant, package_sets, set(["pkg"]), ["standard"])
self.assertEqual(res, ep.return_value)
self.assertEqual(cmr.call_args_list, [mock.call(compose, variant, arch)])
self.assertEqual(
m.run_solver.call_args_list,
[mock.call(variant, arch, set(["pkg", "foo", "bar"]), *cmr.return_value)],
)
self.assertEqual(
ep.call_args_list,
[
mock.call(
{"pkg-3:1-2.x86_64": pkg},
{},
glr.return_value,
m.run_solver.return_value[0],
m.run_solver.return_value[1],
)
],
)
self.assertEqual(
eg.call_args_list,
[mock.call(compose, arch, variant, ["standard"], set_pkg_arch=False)],
)
class MockModule(object):
def __init__(
self, name, platform=None, stream=None, version=None, context=None, rpms=None
):
self.name = name
self.platform = platform
self.stream = stream
self.version = version
self.context = context
self.rpms = rpms or ["pkg-1.0-1.x86_64"]
def get_name(self):
return self.name
def peek_name(self):
return self.name
def peek_stream(self):
return self.stream
def peek_version(self):
return self.version
def peek_context(self):
return self.context
def peek_dependencies(self):
return [
mock.Mock(
peek_requires=mock.Mock(
return_value={
"platform": mock.Mock(
dup=mock.Mock(return_value=[self.platform])
)
}
)
)
]
def copy(self):
return self
def set_arch(self, arch):
pass
def get_rpm_artifacts(self):
return mock.Mock(dup=mock.Mock(return_value=self.rpms))
class HelperMixin(object):
def _repo(self, name):
return os.path.join(self.compose.topdir, "work/x86_64/%s" % name)
@mock.patch("pungi.phases.gather.methods.method_hybrid.Modulemd")
@mock.patch("pungi.phases.gather.methods.method_hybrid.run")
class TestCreateModuleRepo(HelperMixin, helpers.PungiTestCase):
def setUp(self):
super(TestCreateModuleRepo, self).setUp()
self.compose = helpers.DummyCompose(self.topdir, {})
self.variant = self.compose.variants["Server"]
def test_no_modules(self, run, Modulemd):
plat, pkgs = hybrid.create_module_repo(self.compose, self.variant, "x86_64")
self.assertIsNone(plat)
self.assertItemsEqual(pkgs, [])
self.assertEqual(run.call_args_list, [])
self.assertEqual(Modulemd.mock_calls, [])
def test_more_than_one_platform(self, run, Modulemd):
self.variant.arch_mmds["x86_64"] = {
"mod:1": MockModule("mod", platform="f29"),
"mod:2": MockModule("mod", platform="f30"),
}
with self.assertRaises(RuntimeError) as ctx:
hybrid.create_module_repo(self.compose, self.variant, "x86_64")
self.assertIn("conflicting requests for platform", str(ctx.exception))
self.assertEqual(run.call_args_list, [])
self.assertEqual(Modulemd.mock_calls, [])
@mock.patch("pungi.phases.gather.methods.method_hybrid.iter_module_defaults")
def test_creating_repo_with_module_and_default(self, imd, run, Modulemd):
mod = MockModule("mod", platform="f29")
self.variant.arch_mmds["x86_64"] = {"mod:1": mod}
default = mock.Mock(peek_module_name=mock.Mock(return_value="mod"))
imd.return_value = [default]
plat, pkgs = hybrid.create_module_repo(self.compose, self.variant, "x86_64")
self.assertEqual(plat, "f29")
self.assertItemsEqual(pkgs, ["pkg-1.0-1.x86_64"])
self.assertEqual(
Modulemd.mock_calls, [mock.call.dump([mod, default], mock.ANY)]
)
create, modify = run.call_args_list
self.assertEqual(
create[0][0][:2], ["createrepo_c", self._repo("module_repo_Server")]
)
self.assertEqual(
modify[0][0][:4],
[
"modifyrepo_c",
Modulemd.mock_calls[0][1][1],
self._repo("module_repo_Server/repodata"),
"--mdtype=modules",
],
)
class ModifiedMagicMock(mock.MagicMock):
"""Like MagicMock, but remembers original values or mutable arguments."""
def _mock_call(_mock_self, *args, **kwargs):
return super(ModifiedMagicMock, _mock_self)._mock_call(
*copy.deepcopy(args), **copy.deepcopy(kwargs)
)
@mock.patch("pungi.wrappers.fus.parse_output")
@mock.patch("pungi.wrappers.fus.get_cmd", new_callable=ModifiedMagicMock)
@mock.patch("pungi.phases.gather.methods.method_hybrid.run")
class TestRunSolver(HelperMixin, helpers.PungiTestCase):
def setUp(self):
super(TestRunSolver, self).setUp()
self.compose = helpers.DummyCompose(self.topdir, {})
self.phase = hybrid.GatherMethodHybrid(self.compose)
self.phase.multilib_methods = []
self.logfile1 = os.path.join(
self.compose.topdir, "logs/x86_64/hybrid-depsolver-Server-iter-1.x86_64.log"
)
self.logfile2 = os.path.join(
self.compose.topdir, "logs/x86_64/hybrid-depsolver-Server-iter-2.x86_64.log"
)
def test_with_modules(self, run, gc, po):
self.compose.has_comps = None
self.compose.variants["Server"].arch_mmds["x86_64"] = {
"mod:master": mock.Mock(
peek_name=mock.Mock(return_value="mod"),
peek_stream=mock.Mock(return_value="master"),
)
}
po.return_value = (mock.Mock(), mock.Mock())
res = self.phase.run_solver(
self.compose.variants["Server"],
"x86_64",
[],
platform="pl",
modular_rpms=[],
)
self.assertEqual(res, po.return_value)
self.assertEqual(po.call_args_list, [mock.call(self.logfile1)])
self.assertEqual(
run.call_args_list,
[mock.call(gc.return_value, logfile=self.logfile1, show_cmd=True)],
)
self.assertEqual(
gc.call_args_list,
[
mock.call(
"x86_64",
[self._repo("repo"), self._repo("module_repo_Server")],
[],
[],
["mod:master"],
platform="pl",
)
],
)
def test_with_comps(self, run, gc, po):
po.return_value = (mock.Mock(), mock.Mock())
res = self.phase.run_solver(
self.compose.variants["Server"],
"x86_64",
[("pkg", None)],
platform=None,
modular_rpms=[],
)
self.assertEqual(res, po.return_value)
self.assertEqual(po.call_args_list, [mock.call(self.logfile1)])
self.assertEqual(
run.call_args_list,
[mock.call(gc.return_value, logfile=self.logfile1, show_cmd=True)],
)
self.assertEqual(
gc.call_args_list,
[mock.call("x86_64", [self._repo("repo")], [], ["pkg"], [], platform=None)],
)
@mock.patch("pungi.phases.gather.methods.method_hybrid.cr")
def test_multilib_devel(self, cr, run, gc, po):
self.phase.arch = "x86_64"
self.phase.multilib_methods = ["devel"]
self.phase.multilib = mock.Mock()
self.phase.multilib.is_multilib.side_effect = (
lambda pkg: pkg.name == "pkg-devel"
)
self.phase.valid_arches = ["x86_64", "i686", "noarch"]
cr.Metadata.return_value.keys.return_value = []
self.phase.package_maps = {
"x86_64": {
"pkg-devel-1.0-1.x86_64": NamedMock(name="pkg-devel"),
"pkg-devel-1.0-1.i686": NamedMock(name="pkg-devel"),
"foo-1.0-1.x86_64": NamedMock(name="foo"),
}
}
self.phase.packages = self.phase.package_maps["x86_64"]
final = [
("pkg-devel-1.0-1", "x86_64"),
("foo-1.0-1", "x86_64"),
("pkg-devel-1.0-1", "i686"),
]
po.side_effect = [
[[("pkg-devel-1.0-1", "x86_64"), ("foo-1.0-1", "x86_64")], set()],
[final, set()],
]
res = self.phase.run_solver(
self.compose.variants["Server"],
"x86_64",
[("pkg-devel", None), ("foo", None)],
platform=None,
modular_rpms=[],
)
self.assertEqual(res, (final, set()))
self.assertEqual(
po.call_args_list, [mock.call(self.logfile1), mock.call(self.logfile2)]
)
self.assertEqual(
run.call_args_list,
[
mock.call(gc.return_value, logfile=self.logfile1, show_cmd=True),
mock.call(gc.return_value, logfile=self.logfile2, show_cmd=True),
],
)
self.assertEqual(
gc.call_args_list,
[
mock.call(
"x86_64",
[self._repo("repo")],
[],
["pkg-devel", "foo"],
[],
platform=None,
),
mock.call(
"x86_64",
[self._repo("repo")],
[],
["pkg-devel", "foo", "pkg-devel.i686"],
[],
platform=None,
),
],
)
@mock.patch("pungi.phases.gather.methods.method_hybrid.cr")
def test_multilib_runtime(self, cr, run, gc, po):
packages = {
"abc": NamedMock(
name="foo",
epoch=None,
version="1.0",
release="1",
arch="x86_64",
provides=[("/usr/lib/libfoo.1.so.1", None, None)],
),
"def": NamedMock(
name="foo",
epoch=None,
version="1.0",
release="1",
arch="i686",
provides=[("/usr/lib/libfoo.1.so.1", None, None)],
),
"ghi": NamedMock(
name="pkg-devel",
epoch=None,
version="1.0",
release="1",
arch="x86_64",
provides=[],
),
}
cr.Metadata.return_value.keys.return_value = packages.keys()
cr.Metadata.return_value.get.side_effect = lambda key: packages[key]
self.phase.multilib_methods = ["runtime"]
self.phase.multilib = mock.Mock()
self.phase.multilib.is_multilib.side_effect = lambda pkg: pkg.name == "foo"
self.phase.valid_arches = ["x86_64", "i686", "noarch"]
self.phase.arch = "x86_64"
self.phase.package_maps = {
"x86_64": {
"pkg-devel-1.0-1.x86_64": mock.Mock(),
"pkg-devel-1.0-1.i686": mock.Mock(),
"foo-1.0-1.x86_64": mock.Mock(),
"foo-1.0-1.i686": mock.Mock(),
}
}
final = [
("pkg-devel-1.0-1", "x86_64"),
("foo-1.0-1", "x86_64"),
("foo-1.0-1", "i686"),
]
po.side_effect = [
([("pkg-devel-1.0-1", "x86_64"), ("foo-1.0-1", "x86_64")], set()),
(final, set()),
]
res = self.phase.run_solver(
self.compose.variants["Server"],
"x86_64",
[("pkg-devel", None), ("foo", None)],
platform=None,
modular_rpms=[],
)
self.assertEqual(res, (final, set()))
self.assertEqual(
po.call_args_list, [mock.call(self.logfile1), mock.call(self.logfile2)]
)
self.assertEqual(
run.call_args_list,
[
mock.call(gc.return_value, logfile=self.logfile1, show_cmd=True),
mock.call(gc.return_value, logfile=self.logfile2, show_cmd=True),
],
)
self.assertEqual(
gc.call_args_list,
[
mock.call(
"x86_64",
[self._repo("repo")],
[],
["pkg-devel", "foo"],
[],
platform=None,
),
mock.call(
"x86_64",
[self._repo("repo")],
[],
["pkg-devel", "foo", "foo.i686"],
[],
platform=None,
),
],
)
class TestExpandPackages(helpers.PungiTestCase):
def _mk_packages(self, src=None, debug_arch=None):
pkg = MockPkg(
name="pkg",
version="1",
release="2",
arch="x86_64",
epoch=3,
sourcerpm="pkg-1-2.src",
file_path="/tmp/pkg.rpm",
)
nevra_to_pkg = {"pkg-3:1-2.x86_64": pkg}
if src or debug_arch:
nevra_to_pkg["pkg-3:1-2.src"] = pkg._replace(
arch="src", file_path="/tmp/spkg.rpm"
)
if debug_arch:
nevra_to_pkg["pkg-debuginfo-3:1-2.%s" % debug_arch] = pkg._replace(
name="pkg-debuginfo", arch=debug_arch, file_path="/tmp/d1.rpm"
)
return nevra_to_pkg
def test_single_package(self):
nevra_to_pkg = self._mk_packages()
res = hybrid.expand_packages(
nevra_to_pkg, {}, [], [("pkg-3:1-2", "x86_64")], []
)
self.assertEqual(
res,
{
"rpm": [{"path": "/tmp/pkg.rpm", "flags": []}],
"srpm": [],
"debuginfo": [],
},
)
def test_include_src_and_debuginfo(self):
nevra_to_pkg = self._mk_packages(debug_arch="x86_64")
res = hybrid.expand_packages(
nevra_to_pkg, {}, [], [("pkg-3:1-2", "x86_64")], []
)
self.assertEqual(
res,
{
"rpm": [{"path": "/tmp/pkg.rpm", "flags": []}],
"srpm": [{"path": "/tmp/spkg.rpm", "flags": []}],
"debuginfo": [{"path": "/tmp/d1.rpm", "flags": []}],
},
)
def test_skip_debuginfo_for_different_arch(self):
nevra_to_pkg = self._mk_packages(debug_arch="i686")
res = hybrid.expand_packages(
nevra_to_pkg, {}, [], [("pkg-3:1-2", "x86_64")], []
)
self.assertEqual(
res,
{
"rpm": [{"path": "/tmp/pkg.rpm", "flags": []}],
"srpm": [{"path": "/tmp/spkg.rpm", "flags": []}],
"debuginfo": [],
},
)
@mock.patch("pungi.phases.gather.methods.method_hybrid.cr")
def test_skip_lookaside_source_and_debuginfo(self, cr):
nevra_to_pkg = self._mk_packages(debug_arch="x86_64")
lookasides = [mock.Mock()]
repo = {
"abc": NamedMock(
name="pkg",
arch="src",
location_base="file:///tmp/",
location_href="spkg.rpm",
),
"def": NamedMock(
name="pkg-debuginfo",
arch="x86_64",
location_base="file:///tmp/",
location_href="d1.rpm",
),
}
cr.Metadata.return_value.keys.return_value = repo.keys()
cr.Metadata.return_value.get.side_effect = lambda key: repo[key]
res = hybrid.expand_packages(
nevra_to_pkg, {}, lookasides, [("pkg-3:1-2", "x86_64")], []
)
self.assertEqual(
res,
{
"rpm": [{"path": "/tmp/pkg.rpm", "flags": []}],
"srpm": [],
"debuginfo": [],
},
)
@mock.patch("pungi.phases.gather.methods.method_hybrid.cr")
def test_skip_lookaside_packages(self, cr):
nevra_to_pkg = self._mk_packages(debug_arch="x86_64")
lookasides = [mock.Mock()]
repo = {
"abc": NamedMock(
name="pkg",
arch="x86_64",
location_base="file:///tmp/",
location_href="pkg.rpm",
)
}
cr.Metadata.return_value.keys.return_value = repo.keys()
cr.Metadata.return_value.get.side_effect = lambda key: repo[key]
res = hybrid.expand_packages(
nevra_to_pkg, {}, lookasides, [("pkg-3:1-2", "x86_64")], []
)
self.assertEqual(res, {"rpm": [], "srpm": [], "debuginfo": []})
def test_expand_module_packages(self):
nevra_to_pkg = self._mk_packages(src=True)
mod = MockModule(
"foo",
stream="1.0",
version="201807131350",
context="deadcafe",
rpms=["pkg-3:1-2.x86_64"],
)
res = hybrid.expand_packages(
nevra_to_pkg, {"foo-1.0": mod}, [], [], ["foo:1.0:201807131350:deadcafe"]
)
self.assertEqual(
res,
{
"rpm": [{"flags": [], "path": "/tmp/pkg.rpm"}],
"srpm": [{"flags": [], "path": "/tmp/spkg.rpm"}],
"debuginfo": [],
},
)
Reference in New Issue Copy Permalink