pungi/tests/test_initphase.py
Lubomír Sedlář 465ecf229c Add test for materialized pkgset
Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
2019-09-03 09:31:18 +02:00

564 lines
23 KiB
Python

#!/usr/bin/env python
# -*- 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 import Modulemd
from pungi.phases import init
from tests.helpers import DummyCompose, PungiTestCase, touch, mk_boom, fake_run_in_threads
@mock.patch("pungi.phases.init.run_in_threads", new=fake_run_in_threads)
@mock.patch("pungi.phases.init.validate_comps")
@mock.patch("pungi.phases.init.validate_module_defaults")
@mock.patch("pungi.phases.init.write_module_defaults")
@mock.patch("pungi.phases.init.write_global_comps")
@mock.patch("pungi.phases.init.write_arch_comps")
@mock.patch("pungi.phases.init.create_comps_repo")
@mock.patch("pungi.phases.init.write_variant_comps")
@mock.patch("pungi.phases.init.write_prepopulate_file")
class TestInitPhase(PungiTestCase):
def test_run(
self,
write_prepopulate,
write_variant,
create_comps,
write_arch,
write_global,
write_defaults,
validate_defaults,
validate_comps,
):
compose = DummyCompose(self.topdir, {})
compose.has_comps = True
compose.has_module_defaults = False
compose.setup_optional()
phase = init.InitPhase(compose)
phase.run()
self.assertEqual(write_global.mock_calls, [mock.call(compose)])
self.assertEqual(write_prepopulate.mock_calls, [mock.call(compose)])
self.assertItemsEqual(write_arch.mock_calls,
[mock.call(compose, 'x86_64'), mock.call(compose, 'amd64')])
self.assertItemsEqual(create_comps.mock_calls,
[mock.call(compose, 'x86_64', None), mock.call(compose, 'amd64', None),
mock.call(compose, 'x86_64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Client']),
mock.call(compose, 'x86_64', compose.variants['Everything']),
mock.call(compose, 'amd64', compose.variants['Everything']),
mock.call(compose, 'x86_64', compose.all_variants['Server-optional'])])
self.assertItemsEqual(write_variant.mock_calls,
[mock.call(compose, 'x86_64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Client']),
mock.call(compose, 'x86_64', compose.variants['Everything']),
mock.call(compose, 'amd64', compose.variants['Everything']),
mock.call(compose, 'x86_64', compose.all_variants['Server-optional'])])
self.assertItemsEqual(write_defaults, [])
self.assertItemsEqual(validate_defaults, [])
def test_run_with_preserve(
self,
write_prepopulate,
write_variant,
create_comps,
write_arch,
write_global,
write_defaults,
validate_defaults,
validate_comps,
):
compose = DummyCompose(self.topdir, {})
compose.has_comps = True
compose.has_module_defaults = False
compose.variants['Everything'].groups = []
compose.variants['Everything'].modules = []
phase = init.InitPhase(compose)
phase.run()
self.assertEqual(write_global.mock_calls, [mock.call(compose)])
self.assertEqual(
validate_comps.call_args_list, [mock.call(write_global.return_value)]
)
self.assertEqual(write_prepopulate.mock_calls, [mock.call(compose)])
self.assertItemsEqual(write_arch.mock_calls,
[mock.call(compose, 'x86_64'), mock.call(compose, 'amd64')])
self.assertItemsEqual(create_comps.mock_calls,
[mock.call(compose, 'x86_64', None), mock.call(compose, 'amd64', None),
mock.call(compose, 'x86_64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Client']),
mock.call(compose, 'x86_64', compose.variants['Everything']),
mock.call(compose, 'amd64', compose.variants['Everything'])])
self.assertItemsEqual(write_variant.mock_calls,
[mock.call(compose, 'x86_64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Server']),
mock.call(compose, 'amd64', compose.variants['Client']),
mock.call(compose, 'x86_64', compose.variants['Everything']),
mock.call(compose, 'amd64', compose.variants['Everything'])])
self.assertItemsEqual(write_defaults, [])
self.assertItemsEqual(validate_defaults, [])
def test_run_without_comps(
self,
write_prepopulate,
write_variant,
create_comps,
write_arch,
write_global,
write_defaults,
validate_defaults,
validate_comps,
):
compose = DummyCompose(self.topdir, {})
compose.has_comps = False
compose.has_module_defaults = False
phase = init.InitPhase(compose)
phase.run()
self.assertItemsEqual(write_global.mock_calls, [])
self.assertEqual(validate_comps.call_args_list, [])
self.assertItemsEqual(write_prepopulate.mock_calls, [mock.call(compose)])
self.assertItemsEqual(write_arch.mock_calls, [])
self.assertItemsEqual(create_comps.mock_calls, [])
self.assertItemsEqual(write_variant.mock_calls, [])
self.assertItemsEqual(write_defaults, [])
self.assertItemsEqual(validate_defaults, [])
def test_with_module_defaults(
self,
write_prepopulate,
write_variant,
create_comps,
write_arch,
write_global,
write_defaults,
validate_defaults,
validate_comps,
):
compose = DummyCompose(self.topdir, {})
compose.has_comps = False
compose.has_module_defaults = True
phase = init.InitPhase(compose)
phase.run()
self.assertItemsEqual(write_global.mock_calls, [])
self.assertEqual(validate_comps.call_args_list, [])
self.assertItemsEqual(write_prepopulate.mock_calls, [mock.call(compose)])
self.assertItemsEqual(write_arch.mock_calls, [])
self.assertItemsEqual(create_comps.mock_calls, [])
self.assertItemsEqual(write_variant.mock_calls, [])
self.assertItemsEqual(write_defaults.call_args_list, [mock.call(compose)])
self.assertItemsEqual(
validate_defaults.call_args_list,
[mock.call(compose.paths.work.module_defaults_dir())],
)
class TestWriteArchComps(PungiTestCase):
@mock.patch('pungi.phases.init.run')
def test_run(self, run):
compose = DummyCompose(self.topdir, {})
init.write_arch_comps(compose, 'x86_64')
self.assertEqual(run.mock_calls,
[mock.call(['comps_filter', '--arch=x86_64', '--no-cleanup',
'--output=%s/work/x86_64/comps/comps-x86_64.xml' % self.topdir,
self.topdir + '/work/global/comps/comps-global.xml'])])
class TestCreateCompsRepo(PungiTestCase):
@mock.patch('pungi.phases.init.run')
def test_run(self, run):
compose = DummyCompose(self.topdir, {
'createrepo_checksum': 'sha256',
})
init.create_comps_repo(compose, 'x86_64', None)
self.assertEqual(run.mock_calls,
[mock.call(['createrepo_c', self.topdir + '/work/x86_64/comps_repo',
'--outputdir=%s/work/x86_64/comps_repo' % self.topdir,
'--groupfile=%s/work/x86_64/comps/comps-x86_64.xml' % self.topdir,
'--update', '--no-database', '--checksum=sha256',
'--unique-md-filenames'],
logfile=self.topdir + '/logs/x86_64/comps_repo.x86_64.log',
show_cmd=True)])
@mock.patch('pungi.phases.init.run')
def test_run_with_variant(self, run):
compose = DummyCompose(self.topdir, {
'createrepo_checksum': 'sha256',
})
init.create_comps_repo(compose, 'x86_64', compose.variants['Server'])
self.assertEqual(run.mock_calls,
[mock.call(['createrepo_c', self.topdir + '/work/x86_64/comps_repo_Server',
'--outputdir=%s/work/x86_64/comps_repo_Server' % self.topdir,
'--groupfile=%s/work/x86_64/comps/comps-Server.x86_64.xml' % self.topdir,
'--update', '--no-database', '--checksum=sha256',
'--unique-md-filenames'],
logfile=self.topdir + '/logs/x86_64/comps_repo-Server.x86_64.log',
show_cmd=True)])
class TestWriteGlobalComps(PungiTestCase):
@mock.patch('pungi.phases.init.get_file_from_scm')
def test_run_local_file(self, get_file):
compose = DummyCompose(self.topdir, {'comps_file': 'some-file.xml'})
def gen_file(src, dest, logger=None):
self.assertEqual(src, '/home/releng/config/some-file.xml')
touch(os.path.join(dest, 'some-file.xml'))
get_file.side_effect = gen_file
init.write_global_comps(compose)
self.assertTrue(os.path.isfile(self.topdir + '/work/global/comps/comps-global.xml'))
class TestWriteVariantComps(PungiTestCase):
@mock.patch('pungi.phases.init.run')
@mock.patch('pungi.phases.init.CompsWrapper')
def test_run(self, CompsWrapper, run):
compose = DummyCompose(self.topdir, {})
variant = compose.variants['Server']
comps = CompsWrapper.return_value
comps.filter_groups.return_value = []
init.write_variant_comps(compose, 'x86_64', variant)
self.assertEqual(run.mock_calls,
[mock.call(['comps_filter', '--arch=x86_64', '--keep-empty-group=conflicts',
'--keep-empty-group=conflicts-server',
'--variant=Server',
'--output=%s/work/x86_64/comps/comps-Server.x86_64.xml' % self.topdir,
self.topdir + '/work/global/comps/comps-global.xml'])])
self.assertEqual(CompsWrapper.call_args_list,
[mock.call(self.topdir + '/work/x86_64/comps/comps-Server.x86_64.xml')])
self.assertEqual(comps.filter_groups.call_args_list, [mock.call(variant.groups)])
self.assertEqual(comps.filter_environments.mock_calls,
[mock.call(variant.environments)])
self.assertEqual(comps.write_comps.mock_calls, [mock.call()])
@mock.patch("pungi.phases.init.get_lookaside_groups")
@mock.patch("pungi.phases.init.run")
@mock.patch("pungi.phases.init.CompsWrapper")
def test_run_with_lookaside_groups(self, CompsWrapper, run, glg):
compose = DummyCompose(self.topdir, {})
variant = compose.variants["Server"]
comps = CompsWrapper.return_value
comps.filter_groups.return_value = []
glg.return_value = ["foo", "bar"]
init.write_variant_comps(compose, "x86_64", variant)
self.assertEqual(
run.mock_calls,
[
mock.call(
[
"comps_filter",
"--arch=x86_64",
"--keep-empty-group=conflicts",
"--keep-empty-group=conflicts-server",
"--variant=Server",
"--output=%s/work/x86_64/comps/comps-Server.x86_64.xml" % self.topdir,
self.topdir + "/work/global/comps/comps-global.xml",
"--lookaside-group=foo",
"--lookaside-group=bar",
]
),
],
)
self.assertEqual(
CompsWrapper.call_args_list,
[mock.call(self.topdir + "/work/x86_64/comps/comps-Server.x86_64.xml")],
)
self.assertEqual(
comps.filter_groups.call_args_list, [mock.call(variant.groups)]
)
self.assertEqual(
comps.filter_environments.mock_calls, [mock.call(variant.environments)]
)
self.assertEqual(comps.write_comps.mock_calls, [mock.call()])
@mock.patch('pungi.phases.init.run')
@mock.patch('pungi.phases.init.CompsWrapper')
def test_run_no_filter_without_groups(self, CompsWrapper, run):
compose = DummyCompose(self.topdir, {})
variant = compose.variants['Server']
variant.groups = []
comps = CompsWrapper.return_value
comps.filter_groups.return_value = []
init.write_variant_comps(compose, 'x86_64', variant)
self.assertEqual(run.mock_calls,
[mock.call(['comps_filter', '--arch=x86_64', '--keep-empty-group=conflicts',
'--keep-empty-group=conflicts-server',
'--variant=Server',
'--output=%s/work/x86_64/comps/comps-Server.x86_64.xml' % self.topdir,
self.topdir + '/work/global/comps/comps-global.xml'])])
self.assertEqual(CompsWrapper.call_args_list,
[mock.call(self.topdir + '/work/x86_64/comps/comps-Server.x86_64.xml')])
self.assertEqual(comps.filter_groups.call_args_list, [])
self.assertEqual(comps.filter_environments.mock_calls,
[mock.call(variant.environments)])
self.assertEqual(comps.write_comps.mock_calls, [mock.call()])
@mock.patch('pungi.phases.init.run')
@mock.patch('pungi.phases.init.CompsWrapper')
def test_run_filter_for_modular(self, CompsWrapper, run):
compose = DummyCompose(self.topdir, {})
variant = compose.variants['Server']
variant.groups = []
variant.modules = ['testmodule:2.0']
comps = CompsWrapper.return_value
comps.filter_groups.return_value = []
init.write_variant_comps(compose, 'x86_64', variant)
self.assertEqual(run.mock_calls,
[mock.call(['comps_filter', '--arch=x86_64', '--keep-empty-group=conflicts',
'--keep-empty-group=conflicts-server',
'--variant=Server',
'--output=%s/work/x86_64/comps/comps-Server.x86_64.xml' % self.topdir,
self.topdir + '/work/global/comps/comps-global.xml'])])
self.assertEqual(CompsWrapper.call_args_list,
[mock.call(self.topdir + '/work/x86_64/comps/comps-Server.x86_64.xml')])
self.assertEqual(comps.filter_groups.call_args_list, [mock.call([])])
self.assertEqual(comps.filter_environments.mock_calls,
[mock.call(variant.environments)])
self.assertEqual(comps.write_comps.mock_calls, [mock.call()])
@mock.patch('pungi.phases.init.run')
@mock.patch('pungi.phases.init.CompsWrapper')
def test_run_report_unmatched(self, CompsWrapper, run):
compose = DummyCompose(self.topdir, {})
variant = compose.variants['Server']
comps = CompsWrapper.return_value
comps.filter_groups.return_value = ['foo', 'bar']
init.write_variant_comps(compose, 'x86_64', variant)
self.assertEqual(run.mock_calls,
[mock.call(['comps_filter', '--arch=x86_64', '--keep-empty-group=conflicts',
'--keep-empty-group=conflicts-server',
'--variant=Server',
'--output=%s/work/x86_64/comps/comps-Server.x86_64.xml' % self.topdir,
self.topdir + '/work/global/comps/comps-global.xml'])])
self.assertEqual(CompsWrapper.call_args_list,
[mock.call(self.topdir + '/work/x86_64/comps/comps-Server.x86_64.xml')])
self.assertEqual(comps.filter_groups.call_args_list, [mock.call(variant.groups)])
self.assertEqual(comps.filter_environments.mock_calls,
[mock.call(variant.environments)])
self.assertEqual(comps.write_comps.mock_calls, [mock.call()])
self.assertEqual(
compose.log_warning.call_args_list,
[mock.call(init.UNMATCHED_GROUP_MSG % ('Server', 'x86_64', 'foo')),
mock.call(init.UNMATCHED_GROUP_MSG % ('Server', 'x86_64', 'bar'))])
class TestGetLookasideGroups(PungiTestCase):
def test_toplevel_variant(self):
compose = DummyCompose(self.topdir, {})
self.assertItemsEqual(
init.get_lookaside_groups(compose, compose.variants["Server"]), []
)
def test_classic_addon(self):
compose = DummyCompose(self.topdir, {})
compose.setup_addon()
compose.variants["Server"].groups = [{"name": "foo"}]
self.assertItemsEqual(
init.get_lookaside_groups(compose, compose.all_variants["Server-HA"]),
["foo"],
)
def test_variant_as_lookaside(self):
compose = DummyCompose(
self.topdir, {"variant_as_lookaside": [("Server", "Client")]}
)
compose.variants["Client"].groups = [{"name": "foo"}]
self.assertItemsEqual(
init.get_lookaside_groups(compose, compose.variants["Server"]),
["foo"],
)
@mock.patch("shutil.copytree")
@mock.patch("pungi.phases.init.get_dir_from_scm")
class TestWriteModuleDefaults(PungiTestCase):
def test_clone_git(self, gdfs, ct):
conf = {"scm": "git", "repo": "https://pagure.io/pungi.git", "dir": "."}
compose = DummyCompose(self.topdir, {"module_defaults_dir": conf})
init.write_module_defaults(compose)
self.assertEqual(
gdfs.call_args_list, [mock.call(conf, mock.ANY, logger=mock.ANY)]
)
self.assertEqual(
ct.call_args_list,
[
mock.call(
gdfs.call_args_list[0][0][1],
os.path.join(self.topdir, "work/global/module_defaults"),
)
],
)
def test_clone_file_scm(self, gdfs, ct):
conf = {"scm": "file", "dir": "defaults"}
compose = DummyCompose(self.topdir, {"module_defaults_dir": conf})
compose.config_dir = "/home/releng/configs"
init.write_module_defaults(compose)
self.assertEqual(
gdfs.call_args_list,
[
mock.call(
{"scm": "file", "dir": "/home/releng/configs/defaults"},
mock.ANY,
logger=mock.ANY,
)
],
)
self.assertEqual(
ct.call_args_list,
[
mock.call(
gdfs.call_args_list[0][0][1],
os.path.join(self.topdir, "work/global/module_defaults"),
)
],
)
def test_clone_file_str(self, gdfs, ct):
conf = "defaults"
compose = DummyCompose(self.topdir, {"module_defaults_dir": conf})
compose.config_dir = "/home/releng/configs"
init.write_module_defaults(compose)
self.assertEqual(
gdfs.call_args_list,
[mock.call("/home/releng/configs/defaults", mock.ANY, logger=mock.ANY)],
)
self.assertEqual(
ct.call_args_list,
[
mock.call(
gdfs.call_args_list[0][0][1],
os.path.join(self.topdir, "work/global/module_defaults"),
)
],
)
@unittest.skipUnless(Modulemd, "Skipped test, no module support.")
class TestValidateModuleDefaults(PungiTestCase):
def _write_defaults(self, defs):
for mod_name, streams in defs.items():
for stream in streams:
mod_index = Modulemd.ModuleIndex.new()
mmddef = Modulemd.DefaultsV1.new(mod_name)
mmddef.set_default_stream(stream)
mod_index.add_defaults(mmddef)
filename = "%s-%s.yaml" % (mod_name, stream)
with open(os.path.join(self.topdir, filename), "w") as f:
f.write(mod_index.dump_to_string())
def test_valid_files(self):
self._write_defaults({"httpd": ["1"], "python": ["3.6"]})
init.validate_module_defaults(self.topdir)
def test_duplicated_stream(self):
self._write_defaults({"httpd": ["1"], "python": ["3.6", "3.5"]})
with self.assertRaises(RuntimeError) as ctx:
init.validate_module_defaults(self.topdir)
self.assertIn(
"Module python has multiple defaults: 3.5, 3.6", str(ctx.exception)
)
def test_reports_all(self):
self._write_defaults({"httpd": ["1", "2"], "python": ["3.6", "3.5"]})
with self.assertRaises(RuntimeError) as ctx:
init.validate_module_defaults(self.topdir)
self.assertIn("Module httpd has multiple defaults: 1, 2", str(ctx.exception))
self.assertIn(
"Module python has multiple defaults: 3.5, 3.6", str(ctx.exception)
)
def test_handles_non_defaults_file(self):
self._write_defaults({"httpd": ["1"], "python": ["3.6"]})
touch(
os.path.join(self.topdir, "boom.yaml"),
"\n".join(
[
"document: modulemd",
"version: 2",
"data:",
" summary: dummy module",
" description: dummy module",
" license:",
" module: [GPL]",
" content: [GPL]",
]
),
)
init.validate_module_defaults(self.topdir)
@mock.patch("pungi.phases.init.CompsWrapper")
class TestValidateComps(unittest.TestCase):
def test_ok(self, CompsWrapper):
init.validate_comps("/path")
self.assertEqual(
CompsWrapper.mock_calls, [mock.call("/path"), mock.call().validate()]
)
def test_fail(self, CompsWrapper):
CompsWrapper.return_value.validate.side_effect = mk_boom()
with self.assertRaises(Exception):
init.validate_comps("/path")
self.assertEqual(
CompsWrapper.mock_calls, [mock.call("/path"), mock.call().validate()]
)
if __name__ == "__main__":
unittest.main()