almalinux/pungi
14
1
Fork 0
Code Issues Pull Requests Releases Activity
a6f6199910
pungi/tests/test_pkgset_source_koji.py
Lubomír Sedlář d8a553163f
Make python3-mock dependency optional 
https://fedoraproject.org/wiki/Changes/RemovePythonMockUsage

Prefer using unittest.mock to a standalone package. The separate
packages should only really be needed on Python 2.7 these days.

The test requirements file is updated to only require mock on old
Python, and the dependency is removed from setup.py to avoid issues
there.

Relates: https://src.fedoraproject.org/rpms/pungi/pull-request/9

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

(cherry picked from commit ff5a7e6377)
2024-08-30 13:38:43 +03:00

1231 lines
41 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
import json
import unittest
from unittest import mock
import os
import re
import six
from ddt import ddt, data, unpack
from typing import AnyStr, List, Set, Dict, Tuple
from tests.test_gather_method_hybrid import MockModule
from pungi.phases.pkgset.sources import source_koji, source_kojimock
from tests import helpers
from pungi.module_util import Modulemd
from pungi.util import read_single_module_stream_from_file
MMDS_DIR = os.path.join(helpers.FIXTURE_DIR, "mmds")
EVENT_INFO = {"id": 15681980, "ts": 1460956382.81936}
TAG_INFO = {
"maven_support": False,
"locked": False,
"name": "f25",
"extra": {"mock.package_manager": "dnf"},
"perm": None,
"id": 335,
"arches": None,
"maven_include_all": None,
"perm_id": None,
}
def _mk_module_build(r, t):
"""Create a dict as returned Koji buildinfo."""
return {
"name": "foo",
"version": "1",
"release": r,
"nvr": "foo-1-%s" % r,
"tag_name": t,
}
class TestGetKojiEvent(helpers.PungiTestCase):
def setUp(self):
super(TestGetKojiEvent, self).setUp()
self.compose = helpers.DummyCompose(self.topdir, {})
self.event_file = self.topdir + "/work/global/koji-event"
def test_use_preconfigured_event(self):
koji_wrapper = mock.Mock()
self.compose.koji_event = 123456
koji_wrapper.koji_proxy.getEvent.return_value = EVENT_INFO
event = source_koji.get_koji_event_info(self.compose, koji_wrapper)
self.assertEqual(event, EVENT_INFO)
six.assertCountEqual(
self, koji_wrapper.mock_calls, [mock.call.koji_proxy.getEvent(123456)]
)
with open(self.event_file) as f:
self.assertEqual(json.load(f), EVENT_INFO)
def test_gets_last_event(self):
self.compose.koji_event = None
koji_wrapper = mock.Mock()
koji_wrapper.koji_proxy.getLastEvent.return_value = EVENT_INFO
event = source_koji.get_koji_event_info(self.compose, koji_wrapper)
self.assertEqual(event, EVENT_INFO)
six.assertCountEqual(
self, koji_wrapper.mock_calls, [mock.call.koji_proxy.getLastEvent()]
)
with open(self.event_file) as f:
self.assertEqual(json.load(f), EVENT_INFO)
class TestPopulateGlobalPkgset(helpers.PungiTestCase):
def setUp(self):
super(TestPopulateGlobalPkgset, self).setUp()
self.compose = helpers.DummyCompose(
self.topdir,
{"pkgset_koji_tag": "f25", "sigkeys": ["foo", "bar"], "koji_cache": "/tmp"},
)
self.koji_wrapper = mock.Mock()
self.pkgset_path = os.path.join(
self.topdir, "work", "global", "pkgset_global.pickle"
)
self.koji_module_path = os.path.join(
self.topdir, "work", "global", "koji-module-Server.yaml"
)
@mock.patch("pungi.phases.pkgset.sources.source_koji.MaterializedPackageSet.create")
@mock.patch("pungi.phases.pkgset.pkgsets.KojiPackageSet")
def test_populate(self, KojiPackageSet, materialize):
materialize.side_effect = self.mock_materialize
KojiPackageSet.return_value.reuse = None
orig_pkgset = KojiPackageSet.return_value
pkgsets = source_koji.populate_global_pkgset(
self.compose, self.koji_wrapper, 123456
)
self.assertEqual(len(pkgsets), 1)
self.assertIs(pkgsets[0], orig_pkgset)
pkgsets[0].assert_has_calls(
[
mock.call.populate(
"f25",
event=123456,
inherit=True,
include_packages=set(),
),
],
)
def mock_materialize(self, compose, pkgset, prefix, mmd):
self.assertEqual(prefix, "/prefix")
self.assertEqual(compose, self.compose)
return pkgset
@mock.patch("pungi.phases.pkgset.sources.source_koji.MaterializedPackageSet.create")
@mock.patch("pungi.phases.pkgset.pkgsets.KojiPackageSet")
def test_populate_with_multiple_koji_tags(self, KojiPackageSet, materialize):
self.compose = helpers.DummyCompose(
self.topdir,
{
"pkgset_koji_tag": ["f25", "f25-extra"],
"sigkeys": ["foo", "bar"],
"koji_cache": "/tmp",
},
)
materialize.side_effect = self.mock_materialize
KojiPackageSet.return_value.reuse = None
pkgsets = source_koji.populate_global_pkgset(
self.compose, self.koji_wrapper, 123456
)
self.assertEqual(len(pkgsets), 2)
init_calls = KojiPackageSet.call_args_list
six.assertCountEqual(
self, [call[0][0] for call in init_calls], ["f25", "f25-extra"]
)
six.assertCountEqual(
self, [call[0][1] for call in init_calls], [self.koji_wrapper] * 2
)
six.assertCountEqual(
self, [call[0][2] for call in init_calls], [["foo", "bar"]] * 2
)
pkgsets[0].assert_has_calls(
[
mock.call.populate(
"f25", event=123456, inherit=True, include_packages=set()
),
]
)
pkgsets[1].assert_has_calls(
[
mock.call.populate(
"f25-extra", event=123456, inherit=True, include_packages=set()
),
]
)
@mock.patch("pungi.phases.pkgset.sources.source_koji.MaterializedPackageSet.create")
@mock.patch("pungi.phases.pkgset.pkgsets.KojiPackageSet.populate")
@mock.patch("pungi.phases.pkgset.pkgsets.KojiPackageSet.save_file_list")
def test_populate_packages_to_gather(self, save_file_list, popuplate, materialize):
self.compose = helpers.DummyCompose(
self.topdir,
{
"gather_method": "nodeps",
"pkgset_koji_tag": "f25",
"koji_cache": "/tmp",
"sigkeys": ["foo", "bar"],
"additional_packages": [(".*", {"*": ["pkg", "foo.x86_64"]})],
},
)
materialize.side_effect = self.mock_materialize
pkgsets = source_koji.populate_global_pkgset(
self.compose, self.koji_wrapper, 123456
)
self.assertEqual(len(pkgsets), 1)
six.assertCountEqual(self, pkgsets[0].packages, ["pkg", "foo"])
class TestGetPackageSetFromKoji(helpers.PungiTestCase):
def setUp(self):
super(TestGetPackageSetFromKoji, self).setUp()
self.compose = helpers.DummyCompose(
self.topdir, {"pkgset_koji_tag": "f25", "koji_cache": "/tmp"}
)
self.compose.koji_event = None
self.koji_wrapper = mock.Mock()
self.koji_wrapper.koji_proxy.getLastEvent.return_value = EVENT_INFO
self.koji_wrapper.koji_proxy.getTag.return_value = TAG_INFO
@mock.patch("pungi.phases.pkgset.sources.source_koji.populate_global_pkgset")
def test_get_package_sets(self, pgp):
pkgsets = source_koji.get_pkgset_from_koji(self.compose, self.koji_wrapper)
six.assertCountEqual(
self, self.koji_wrapper.koji_proxy.mock_calls, [mock.call.getLastEvent()]
)
self.assertEqual(pkgsets, pgp.return_value)
self.assertEqual(
pgp.call_args_list,
[mock.call(self.compose, self.koji_wrapper, EVENT_INFO)],
)
def test_get_koji_modules(self):
mock_build_ids = [
{"id": 1065873, "name": "testmodule2-master_dash-20180406051653.96c371af"}
]
mock_extra = {
"typeinfo": {
"module": {
"content_koji_tag": "module-b62270b82443edde",
"modulemd_str": mock.Mock(),
"name": "testmodule2",
"stream": "master",
"version": "20180406051653",
"context": "96c371af",
}
}
}
mock_build_md = [
{
"id": 1065873,
"epoch": None,
"extra": mock_extra,
"name": "testmodule2",
"nvr": "testmodule2-master_dash-20180406051653.2e6f5e0a",
"release": "20180406051653.2e6f5e0a",
"state": 1,
"version": "master_dash",
"completion_ts": 1433473124.0,
}
]
self.koji_wrapper.koji_proxy.search.return_value = mock_build_ids
self.koji_wrapper.koji_proxy.getBuild.return_value = mock_build_md[0]
event = {"id": 12345, "ts": 1533473124.0}
module_info_str = "testmodule2:master-dash:20180406051653:96c371af"
result = source_koji.get_koji_modules(
self.compose, self.koji_wrapper, event, module_info_str
)
assert isinstance(result, list)
assert len(result) == 1
module = result[0]
assert isinstance(module, dict)
self.assertIn("module_stream", module)
self.assertIn("module_version", module)
self.assertIn("module_context", module)
self.assertIn("tag", module)
expected_query = "testmodule2-master_dash-20180406051653.96c371af"
self.koji_wrapper.koji_proxy.search.assert_called_once_with(
expected_query, "build", "glob"
)
self.koji_wrapper.koji_proxy.getBuild.assert_called_once_with(
mock_build_ids[0]["id"]
)
def test_get_koji_modules_filter_by_event(self):
mock_build_ids = [
{"id": 1065873, "name": "testmodule2-master_dash-20180406051653.96c371af"}
]
mock_extra = {
"typeinfo": {
"module": {
"content_koji_tag": "module-b62270b82443edde",
"modulemd_str": mock.Mock(),
}
}
}
mock_build_md = [
{
"id": 1065873,
"epoch": None,
"extra": mock_extra,
"name": "testmodule2",
"nvr": "testmodule2-master_dash-20180406051653.2e6f5e0a",
"release": "20180406051653.2e6f5e0a",
"state": 1,
"version": "master_dash",
"completion_ts": 1633473124.0,
}
]
self.koji_wrapper.koji_proxy.search.return_value = mock_build_ids
self.koji_wrapper.koji_proxy.getBuild.return_value = mock_build_md[0]
event = {"id": 12345, "ts": 1533473124.0}
with self.assertRaises(ValueError) as ctx:
source_koji.get_koji_modules(
self.compose, self.koji_wrapper, event, "testmodule2:master-dash"
)
self.assertIn("No module build found", str(ctx.exception))
self.koji_wrapper.koji_proxy.search.assert_called_once_with(
"testmodule2-master_dash-*", "build", "glob"
)
self.koji_wrapper.koji_proxy.getBuild.assert_called_once_with(
mock_build_ids[0]["id"]
)
self.koji_wrapper.koji_proxy.listArchives.assert_not_called()
self.koji_wrapper.koji_proxy.listRPMs.assert_not_called()
def test_get_koji_modules_no_version(self):
mock_build_ids = [
{"id": 1065873, "name": "testmodule2-master-20180406051653.2e6f5e0a"},
{"id": 1065874, "name": "testmodule2-master-20180406051653.96c371af"},
]
mock_extra = [
{
"typeinfo": {
"module": {
"content_koji_tag": "module-b62270b82443edde",
"modulemd_str": mock.Mock(),
"name": "testmodule2",
"stream": "master",
"version": "20180406051653",
"context": "2e6f5e0a",
}
}
},
{
"typeinfo": {
"module": {
"content_koji_tag": "module-52e40b9cdd3c0f7d",
"modulemd_str": mock.Mock(),
"name": "testmodule2",
"stream": "master",
"version": "20180406051653",
"context": "96c371af",
}
}
},
]
mock_build_md = [
{
"id": 1065873,
"epoch": None,
"extra": mock_extra[0],
"name": "testmodule2",
"nvr": "testmodule2-master-20180406051653.2e6f5e0a",
"release": "20180406051653.2e6f5e0a",
"state": 1,
"version": "master",
"completion_ts": 1433473124.0,
},
{
"id": 1065874,
"epoch": None,
"extra": mock_extra[1],
"name": "testmodule2",
"nvr": "testmodule2-master-20180406051653.96c371af",
"release": "20180406051653.96c371af",
"state": 1,
"version": "master",
"completion_ts": 1433473124.0,
},
]
self.koji_wrapper.koji_proxy.search.return_value = mock_build_ids
self.koji_wrapper.koji_proxy.getBuild.side_effect = mock_build_md
event = {"id": 12345, "ts": 1533473124.0}
module_info_str = "testmodule2:master"
result = source_koji.get_koji_modules(
self.compose, self.koji_wrapper, event, module_info_str
)
assert isinstance(result, list)
assert len(result) == 2
module = result[0]
for module in result:
assert isinstance(module, dict)
self.assertIn("module_stream", module)
self.assertIn("module_version", module)
self.assertIn("module_context", module)
expected_query = "testmodule2-master-*"
self.koji_wrapper.koji_proxy.search.assert_called_once_with(
expected_query, "build", "glob"
)
expected_calls = [
mock.call(mock_build_ids[0]["id"]),
mock.call(mock_build_ids[1]["id"]),
]
self.koji_wrapper.koji_proxy.getBuild.mock_calls == expected_calls
def test_get_koji_modules_ignore_deleted(self):
mock_build_ids = [
{"id": 1065873, "name": "testmodule2-master_dash-20180406051653.96c371af"}
]
mock_extra = {
"typeinfo": {
"module": {
"content_koji_tag": "module-b62270b82443edde",
"modulemd_str": mock.Mock(),
"name": "testmodule2",
"stream": "master",
"version": "20180406051653",
"context": "96c371af",
}
}
}
mock_build_md = [
{
"id": 1065873,
"epoch": None,
"extra": mock_extra,
"name": "testmodule2",
"nvr": "testmodule2-master_dash-20180406051653.96c371af",
"release": "20180406051653.96c371af",
"state": 2,
"version": "master_dash",
"completion_ts": 1433473124.0,
}
]
self.koji_wrapper.koji_proxy.search.return_value = mock_build_ids
self.koji_wrapper.koji_proxy.getBuild.return_value = mock_build_md[0]
event = {"id": 12345, "ts": 1533473124.0}
with self.assertRaises(ValueError) as ctx:
source_koji.get_koji_modules(
self.compose, self.koji_wrapper, event, "testmodule2:master-dash"
)
self.compose.log_debug.assert_called_once_with(
"Module build %s has been deleted, ignoring it." % mock_build_ids[0]["name"]
)
self.assertIn("No module build found", str(ctx.exception))
self.koji_wrapper.koji_proxy.search.assert_called_once_with(
"testmodule2-master_dash-*", "build", "glob"
)
self.koji_wrapper.koji_proxy.getBuild.assert_called_once_with(
mock_build_ids[0]["id"]
)
self.koji_wrapper.koji_proxy.listArchives.assert_not_called()
self.koji_wrapper.koji_proxy.listRPMs.assert_not_called()
class TestSourceKoji(helpers.PungiTestCase):
@mock.patch("pungi.phases.pkgset.sources.source_koji.get_pkgset_from_koji")
@mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper")
def test_run(self, KojiWrapper, gpfk):
compose = helpers.DummyCompose(
self.topdir, {"koji_profile": "koji", "koji_cache": "/prefix"}
)
KojiWrapper.return_value.koji_module.config.topdir = "/prefix"
phase = source_koji.PkgsetSourceKoji(compose)
pkgsets, path_prefix = phase()
self.assertEqual(pkgsets, gpfk.return_value)
self.assertEqual(path_prefix, "/prefix")
self.assertEqual(KojiWrapper.mock_calls, [mock.call(compose)])
class TestCorrectNVR(helpers.PungiTestCase):
def setUp(self):
super(TestCorrectNVR, self).setUp()
self.compose = helpers.DummyCompose(self.topdir, {})
self.nv = "base-runtime-f26"
self.nvr = "base-runtime-f26-20170502134116"
self.release_regex = re.compile(r"^(\d){14}$")
self.new_nv = "base-runtime:f26"
self.new_nvr = "base-runtime:f26:20170502134116"
self.new_nvrc = "base-runtime:f26:20170502134116:0123abcd"
def test_nv(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.nv)
expectedKeys = ["stream", "name"]
six.assertCountEqual(self, module_info.keys(), expectedKeys)
def test_nvr(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.nvr)
expectedKeys = ["stream", "name", "version"]
six.assertCountEqual(self, module_info.keys(), expectedKeys)
def test_correct_release(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.nvr)
self.assertIsNotNone(self.release_regex.match(module_info["version"]))
def test_new_nv(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.new_nv)
expected = {"name": "base-runtime", "stream": "f26"}
self.assertEqual(module_info, expected)
def test_new_nvr(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.new_nvr)
expected = {
"name": "base-runtime",
"stream": "f26",
"version": "20170502134116",
}
self.assertEqual(module_info, expected)
def test_new_nvrc(self):
module_info = source_koji.variant_dict_from_str(self.compose, self.new_nvrc)
expected = {
"name": "base-runtime",
"stream": "f26",
"version": "20170502134116",
"context": "0123abcd",
}
self.assertEqual(module_info, expected)
def test_new_garbage_value(self):
self.assertRaises(
ValueError,
source_koji.variant_dict_from_str,
self.compose,
"foo:bar:baz:quux:qaar",
)
class TestFilterInherited(unittest.TestCase):
def test_empty_module_list(self):
event = {"id": 123456}
koji_proxy = mock.Mock()
module_builds = []
top_tag = "top-tag"
koji_proxy.getFullInheritance.return_value = [
{"name": "middle-tag"},
{"name": "bottom-tag"},
]
result = source_koji.filter_inherited(koji_proxy, event, module_builds, top_tag)
self.assertEqual(result, [])
self.assertEqual(
koji_proxy.mock_calls,
[mock.call.getFullInheritance("top-tag", event=123456)],
)
def test_exclude_middle_and_bottom_tag(self):
event = {"id": 123456}
koji_proxy = mock.Mock()
top_tag = "top-tag"
koji_proxy.getFullInheritance.return_value = [
{"name": "middle-tag"},
{"name": "bottom-tag"},
]
m1 = _mk_module_build("1", "top-tag")
m2 = _mk_module_build("2", "middle-tag")
m3 = _mk_module_build("3", "bottom-tag")
module_builds = [m1, m2, m3]
result = source_koji.filter_inherited(koji_proxy, event, module_builds, top_tag)
six.assertCountEqual(self, result, [m1])
self.assertEqual(
koji_proxy.mock_calls,
[mock.call.getFullInheritance("top-tag", event=123456)],
)
def test_missing_from_top_tag(self):
event = {"id": 123456}
koji_proxy = mock.Mock()
top_tag = "top-tag"
koji_proxy.getFullInheritance.return_value = [
{"name": "middle-tag"},
{"name": "bottom-tag"},
]
m2 = _mk_module_build("2", "bottom-tag")
m3 = _mk_module_build("3", "middle-tag")
module_builds = [m2, m3]
result = source_koji.filter_inherited(koji_proxy, event, module_builds, top_tag)
six.assertCountEqual(self, result, [m3])
self.assertEqual(
koji_proxy.mock_calls,
[mock.call.getFullInheritance("top-tag", event=123456)],
)
def test_build_in_multiple_tags(self):
event = {"id": 123456}
koji_proxy = mock.Mock()
top_tag = "top-tag"
koji_proxy.getFullInheritance.return_value = [
{"name": "middle-tag"},
{"name": "bottom-tag"},
]
m = _mk_module_build("1", "middle-tag")
module_builds = [m, m]
result = source_koji.filter_inherited(koji_proxy, event, module_builds, top_tag)
six.assertCountEqual(self, result, [m])
self.assertEqual(
koji_proxy.mock_calls,
[mock.call.getFullInheritance("top-tag", event=123456)],
)
class TestFilterByWhitelist(unittest.TestCase):
def _build(self, n, s, v, c):
s = s.replace("-", "_")
return {
"nvr": "%s-%s-%s.%s" % (n, s, v, c),
"name": n,
"version": s,
"release": "%s.%s" % (v, c),
}
def test_no_modules(self):
compose = mock.Mock()
module_builds = []
input_modules = [{"name": "foo:1"}]
expected = set(["foo:1"])
source_koji.filter_by_whitelist(compose, module_builds, input_modules, expected)
self.assertEqual(expected, set(["foo:1"]))
def test_filter_by_NS(self):
compose = mock.Mock()
module_builds = [
self._build("foo", "1", "201809031048", "cafebabe"),
self._build("foo", "1", "201809031047", "deadbeef"),
self._build("foo", "2", "201809031047", "deadbeef"),
]
input_modules = [{"name": "foo:1"}]
expected = set(["foo:1"])
result = source_koji.filter_by_whitelist(
compose, module_builds, input_modules, expected
)
six.assertCountEqual(self, result, [module_builds[0], module_builds[1]])
self.assertEqual(expected, set())
def test_filter_by_NSV(self):
compose = mock.Mock()
module_builds = [
self._build("foo", "1", "201809031048", "cafebabe"),
self._build("foo", "1", "201809031047", "deadbeef"),
self._build("foo", "2", "201809031047", "deadbeef"),
]
input_modules = [{"name": "foo:1:201809031047"}]
expected = set(["foo:1:201809031047"])
result = source_koji.filter_by_whitelist(
compose, module_builds, input_modules, expected
)
self.assertEqual(result, [module_builds[1]])
self.assertEqual(expected, set())
def test_filter_by_NSVC(self):
compose = mock.Mock()
module_builds = [
self._build("foo", "1", "201809031048", "cafebabe"),
self._build("foo", "1", "201809031047", "deadbeef"),
self._build("foo", "1", "201809031047", "cafebabe"),
self._build("foo", "2", "201809031047", "deadbeef"),
]
input_modules = [{"name": "foo:1:201809031047:deadbeef"}]
expected = set()
result = source_koji.filter_by_whitelist(
compose, module_builds, input_modules, expected
)
self.assertEqual(result, [module_builds[1]])
self.assertEqual(expected, set())
def test_filter_by_wildcard(self):
compose = mock.Mock()
module_builds = [
self._build("foo", "1", "201809031048", "cafebabe"),
self._build("foo", "1", "201809031047", "deadbeef"),
self._build("foo", "2", "201809031047", "deadbeef"),
]
input_modules = [{"name": "*"}]
expected = set(["*"])
result = source_koji.filter_by_whitelist(
compose, module_builds, input_modules, expected
)
six.assertCountEqual(self, result, module_builds)
self.assertEqual(expected, set())
@unittest.skipIf(Modulemd is None, "Skipping tests, no module support")
class TestAddModuleToVariant(helpers.PungiTestCase):
def setUp(self):
super(TestAddModuleToVariant, self).setUp()
self.koji = mock.Mock()
self.compose = helpers.DummyCompose(self.topdir, {})
self.koji.koji_module.pathinfo.typedir.return_value = MMDS_DIR
files = ["modulemd.x86_64.txt", "modulemd.armv7hl.txt", "modulemd.txt"]
self.koji.koji_proxy.listArchives.return_value = [
{"btype": "module", "filename": fname} for fname in files
] + [{"btype": "foo"}]
self.buildinfo = {
"id": 1234,
"extra": {
"typeinfo": {
"module": {
"name": "module",
"stream": "master",
"version": "20190318",
"context": "abcdef",
},
},
},
}
def test_adding_module(self):
variant = mock.Mock(arches=["armhfp", "x86_64"], arch_mmds={}, modules=[])
source_koji._add_module_to_variant(
self.koji, variant, self.buildinfo, compose=self.compose
)
mod1 = variant.arch_mmds["armhfp"]["module:master:20190318:abcdef"]
self.assertEqual(mod1.get_NSVCA(), "module:master:20190318:abcdef:armhfp")
mod2 = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod2.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
self.assertEqual(len(variant.arch_mmds), 2)
self.assertEqual(variant.modules, [])
def test_adding_module_to_existing(self):
variant = mock.Mock(
arches=["armhfp", "x86_64"],
arch_mmds={
"x86_64": {
"m1:latest:20190101:cafe": read_single_module_stream_from_file(
os.path.join(MMDS_DIR, "m1.x86_64.txt")
)
}
},
modules=[{"name": "m1:latest-20190101:cafe", "glob": False}],
)
source_koji._add_module_to_variant(
self.koji, variant, self.buildinfo, compose=self.compose
)
mod1 = variant.arch_mmds["armhfp"]["module:master:20190318:abcdef"]
self.assertEqual(mod1.get_NSVCA(), "module:master:20190318:abcdef:armhfp")
mod2 = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod2.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
mod3 = variant.arch_mmds["x86_64"]["m1:latest:20190101:cafe"]
self.assertEqual(mod3.get_NSVCA(), "m1:latest:20190101:cafe:x86_64")
self.assertEqual(
variant.modules, [{"name": "m1:latest-20190101:cafe", "glob": False}]
)
def test_adding_module_with_add_module(self):
variant = mock.Mock(arches=["armhfp", "x86_64"], arch_mmds={}, modules=[])
source_koji._add_module_to_variant(
self.koji,
variant,
self.buildinfo,
compose=self.compose,
add_to_variant_modules=True,
)
mod1 = variant.arch_mmds["armhfp"]["module:master:20190318:abcdef"]
self.assertEqual(mod1.get_NSVCA(), "module:master:20190318:abcdef:armhfp")
mod2 = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod2.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
self.assertEqual(
variant.modules, [{"name": "module:master:20190318:abcdef", "glob": False}]
)
def test_adding_module_to_existing_with_add_module(self):
variant = mock.Mock(
arches=["armhfp", "x86_64"],
arch_mmds={
"x86_64": {
"m1:latest:20190101:cafe": read_single_module_stream_from_file(
os.path.join(MMDS_DIR, "m1.x86_64.txt")
)
}
},
modules=[{"name": "m1:latest-20190101:cafe", "glob": False}],
)
source_koji._add_module_to_variant(
self.koji,
variant,
self.buildinfo,
compose=self.compose,
add_to_variant_modules=True,
)
mod1 = variant.arch_mmds["armhfp"]["module:master:20190318:abcdef"]
self.assertEqual(mod1.get_NSVCA(), "module:master:20190318:abcdef:armhfp")
mod2 = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod2.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
mod3 = variant.arch_mmds["x86_64"]["m1:latest:20190101:cafe"]
self.assertEqual(mod3.get_NSVCA(), "m1:latest:20190101:cafe:x86_64")
self.assertEqual(
variant.modules,
[
{"name": "m1:latest-20190101:cafe", "glob": False},
{"name": "module:master:20190318:abcdef", "glob": False},
],
)
def test_adding_module_but_filtered(self):
compose = helpers.DummyCompose(
self.topdir, {"filter_modules": [(".*", {"*": ["module:*"]})]}
)
variant = mock.Mock(
arches=["armhfp", "x86_64"], arch_mmds={}, modules=[], uid="Variant"
)
nsvc = source_koji._add_module_to_variant(
self.koji,
variant,
self.buildinfo,
add_to_variant_modules=True,
compose=compose,
)
self.assertIsNone(nsvc)
self.assertEqual(variant.arch_mmds, {})
self.assertEqual(variant.modules, [])
@mock.patch("pungi.module_util.Modulemd.ModuleStream.read_file", new=MockModule)
@unittest.skipIf(Modulemd is None, "Skipping tests, no module support")
class TestAddModuleToVariantForKojiMock(helpers.PungiTestCase):
def setUp(self):
super(TestAddModuleToVariantForKojiMock, self).setUp()
self.koji = mock.Mock()
self.koji.koji_module.pathinfo.typedir.return_value = "/koji"
self.compose = helpers.DummyCompose(self.topdir, {})
self.koji.koji_module.pathinfo.topdir = MMDS_DIR
files = [
"modulemd.x86_64.txt",
"scratch-module.x86_64.txt",
]
self.koji.koji_proxy.listArchives.return_value = [
{"btype": "module", "filename": fname} for fname in files
]
self.buildinfo = {
"id": 1234,
"arch": "x86_64",
"extra": {
"typeinfo": {
"module": {
"name": "module",
"stream": "master",
"version": "20190318",
"context": "abcdef",
'content_koji_tag': 'module:master-20190318-abcdef'
},
},
},
}
def test_adding_module(self):
variant = mock.Mock(
arches=[
"x86_64"
],
arch_mmds={},
modules=[],
)
source_kojimock._add_module_to_variant(
self.koji,
variant,
self.buildinfo,
)
mod = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
self.assertEqual(len(variant.arch_mmds), 1)
self.assertEqual(variant.modules, [])
def test_adding_module_to_existing(self):
variant = mock.Mock(
arches=[
"x86_64"
],
arch_mmds={
"x86_64": {
"m1:latest:20190101:cafe": read_single_module_stream_from_file(
os.path.join(MMDS_DIR, "m1.x86_64.txt")
)}
},
modules=[{"name": "m1:latest-20190101:cafe", "glob": False}],
)
source_koji._add_module_to_variant(
self.koji, variant, self.buildinfo, compose=self.compose
)
mod = variant.arch_mmds["x86_64"]["m1:latest:20190101:cafe"]
self.assertEqual(mod.get_NSVCA(), "m1:latest:20190101:cafe:x86_64")
self.assertEqual(
variant.modules,
[{"name": "m1:latest-20190101:cafe", "glob": False}]
)
def test_adding_module_with_add_module(self):
variant = mock.Mock(arches=[
"x86_64"
], arch_mmds={}, modules=[])
source_kojimock._add_module_to_variant(
self.koji, variant, self.buildinfo, add_to_variant_modules=True
)
mod = variant.arch_mmds["x86_64"]["module:master:20190318:abcdef"]
self.assertEqual(mod.get_NSVCA(), "module:master:20190318:abcdef:x86_64")
self.assertEqual(
variant.modules, [{"name": "module:master:20190318:abcdef", "glob": False}]
)
def test_adding_module_to_existing_with_add_module(self):
variant = mock.Mock(
arches=[
"x86_64"
],
arch_mmds={
"x86_64": {"m1:latest:20190101:cafe": read_single_module_stream_from_file(
os.path.join(MMDS_DIR, "m1.x86_64.txt")
)
}
},
modules=[{"name": "m1:latest-20190101:cafe", "glob": False}],
)
source_kojimock._add_module_to_variant(
self.koji, variant, self.buildinfo, add_to_variant_modules=True
)
mod = variant.arch_mmds["x86_64"]["m1:latest:20190101:cafe"]
self.assertEqual(mod.get_NSVCA(), "m1:latest:20190101:cafe:x86_64")
self.assertEqual(
variant.modules,
[
{"name": "m1:latest-20190101:cafe", "glob": False},
{"name": "module:master:20190318:abcdef", "glob": False},
],
)
def test_adding_module_but_filtered(self):
compose = helpers.DummyCompose(
self.topdir, {"filter_modules": [(".*", {"*": ["module:*"]})]}
)
variant = mock.Mock(
arches=[
"x86_64"
], arch_mmds={}, modules=[], uid="Variant"
)
nsvc = source_kojimock._add_module_to_variant(
self.koji,
variant,
self.buildinfo,
add_to_variant_modules=True,
compose=compose,
)
self.assertIsNone(nsvc)
self.assertEqual(variant.arch_mmds, {})
self.assertEqual(variant.modules, [])
class TestIsModuleFiltered(helpers.PungiTestCase):
def assertIsFiltered(self, name, stream):
self.assertTrue(
source_koji._is_filtered_out(
self.compose, self.compose.variants["Server"], "x86_64", name, stream
)
)
def assertIsNotFiltered(self, name, stream):
self.assertFalse(
source_koji._is_filtered_out(
self.compose, self.compose.variants["Server"], "x86_64", name, stream
)
)
def test_no_filters(self):
self.compose = helpers.DummyCompose(self.topdir, {})
self.assertIsNotFiltered("foo", "master")
def test_filter_by_name(self):
self.compose = helpers.DummyCompose(
self.topdir, {"filter_modules": [(".*", {"*": ["foo"]})]}
)
self.assertIsFiltered("foo", "master")
self.assertIsNotFiltered("bar", "master")
def test_filter_by_stream(self):
self.compose = helpers.DummyCompose(
self.topdir, {"filter_modules": [(".*", {"*": ["foo:master"]})]}
)
self.assertIsFiltered("foo", "master")
self.assertIsNotFiltered("bar", "master")
self.assertIsNotFiltered("foo", "stable")
class MockMBS(object):
def __init__(self, api_url):
self.api_url = api_url
def get_module_build_by_nsvc(self, nsvc):
return {"id": 1, "koji_tag": "scratch-module-tag", "name": "scratch-module"}
def final_modulemd(self, module_build_id):
with open(os.path.join(MMDS_DIR, "scratch-module.x86_64.txt")) as f:
return {"x86_64": f.read()}
@mock.patch("pungi.phases.pkgset.sources.source_koji.MBSWrapper", new=MockMBS)
@unittest.skipIf(Modulemd is None, "Skipping tests, no module support")
class TestAddScratchModuleToVariant(helpers.PungiTestCase):
def setUp(self):
super(TestAddScratchModuleToVariant, self).setUp()
self.compose = helpers.DummyCompose(
self.topdir, {"mbs_api_url": "http://mbs.local/module-build-service/2"}
)
self.nsvc = "scratch-module:master:20200710:abcdef"
def test_adding_scratch_module(self):
variant = mock.Mock(
arches=[
# "armhfp",
"x86_64"
],
arch_mmds={},
modules=[],
module_uid_to_koji_tag={},
)
variant_tags = {variant: []}
tag_to_mmd = {}
scratch_modules = [self.nsvc]
source_koji._add_scratch_modules_to_variant(
self.compose, variant, scratch_modules, variant_tags, tag_to_mmd
)
self.assertEqual(variant_tags, {variant: ["scratch-module-tag"]})
self.assertEqual(
variant.arch_mmds["x86_64"][self.nsvc].get_NSVCA(),
"scratch-module:master:20200710:abcdef:x86_64",
)
self.assertTrue(isinstance(tag_to_mmd["scratch-module-tag"]["x86_64"], set))
self.assertEqual(
list(tag_to_mmd["scratch-module-tag"]["x86_64"])[0].get_NSVCA(),
"scratch-module:master:20200710:abcdef:x86_64",
)
self.assertEqual(variant.modules, [])
def test_adding_scratch_module_nontest_compose(self):
self.compose.compose_type = "production"
scratch_modules = [self.nsvc]
source_koji._add_scratch_modules_to_variant(
self.compose, mock.Mock(), scratch_modules, {}, {}
)
self.compose.log_warning.assert_called_once_with(
"Only test composes could include scratch module builds"
)
@ddt
class TestSourceKoji(unittest.TestCase):
@unpack
@data(
(
'AppStream',
[
'x86_64',
'i386'
],
{
'python39-devel:3.9',
'python39:3.9',
},
[
(
'^(BaseOS|AppStream|PowerTools)$',
{
'x86_64': [
'python39:3.9',
],
'aarch64': [
'python39-devel:3.9',
]
}
)
],
{
'python39-devel:3.9',
}
),
(
'AppStream',
[
'x86_64',
'i386'
],
{
'python39-devel:3.9',
'python39:3.9',
'python38-devel:3.8',
'python38:3.8',
},
[
(
'^(BaseOS|AppStream|PowerTools)$',
{
'x86_64': [
'python39:3.9',
],
'*': [
'python38-devel:3.8',
]
}
)
],
{
'python39-devel:3.9',
'python38:3.8',
}
),
(
'AppStream',
[
'x86_64',
'i386'
],
{
'python39-devel:3.9',
'python39:3.9',
'python38-devel:3.8',
'python38:3.8',
},
[
(
'^(BaseOS|AppStream|PowerTools)$',
{
'x86_64': [
'python39:3.9',
],
'aarch64': [
'python38-devel:3.8',
]
}
),
(
'*',
{
'*': [
'python38-devel:3.8',
]
}
),
],
{
'python39-devel:3.9',
'python38:3.8',
}
),
)
def test__filter_expected_modules(
self,
variant_name: AnyStr,
variant_arches: List[AnyStr],
expected_modules: Set[AnyStr],
filtered_modules: List[Tuple[AnyStr, Dict[AnyStr, List[AnyStr]]]],
expected_result: Set[AnyStr],
) -> None:
real_result = source_kojimock._filter_expected_modules(
variant_name=variant_name,
variant_arches=variant_arches,
expected_modules=expected_modules,
filtered_modules=filtered_modules,
)
self.assertSetEqual(
real_result,
expected_result,
)
Reference in New Issue View Git Blame Copy Permalink