almalinux/pungi
14
1
Fork 0
Code Issues Pull Requests Releases Activity
de53dd0bbd
pungi/tests/test_koji_wrapper.py
oshyshatskyi 903db91c0f LNX-102: Patch pungi tool to use local koji mock 
Instead of koji.mbox use local koji-like wrapper.

@BS-LINKED-5ff8b8cb6f44affc6c5e9a7a
@BS-TARGET-CL8

Change-Id: I82a2bc8bc71ae06240656898f3df71bb28bcb9e9
2021-01-25 16:17:33 +02:00

1202 lines
40 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
import json
import mock
try:
import unittest2 as unittest
except ImportError:
import unittest
import tempfile
import os
import six
from pungi.wrappers.kojiwrapper import KojiWrapper, get_buildroot_rpms
from .helpers import FIXTURE_DIR
class DumbMock(object):
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
def mock_imagebuild_path(id):
if isinstance(id, int):
return "/koji/imagebuild/" + str(id)
return "/koji/imagebuild/" + str(hash(str(id)))
class KojiWrapperBaseTestCase(unittest.TestCase):
def setUp(self):
_, self.tmpfile = tempfile.mkstemp()
self.koji_profile = mock.Mock()
with mock.patch("pungi.wrappers.kojiwrapper.koji") as koji:
koji.gssapi_login = mock.Mock()
koji.get_profile_module = mock.Mock(
return_value=mock.Mock(
config=DumbMock(
server="koji.example.com", authtype="kerberos", cert="",
),
pathinfo=mock.Mock(
work=mock.Mock(return_value="/koji"),
taskrelpath=mock.Mock(side_effect=lambda id: "task/" + str(id)),
imagebuild=mock.Mock(side_effect=mock_imagebuild_path),
),
)
)
self.koji_profile = koji.get_profile_module.return_value
self.koji = KojiWrapper("custom-koji", real_koji=True)
def tearDown(self):
os.remove(self.tmpfile)
class KojiWrapperTest(KojiWrapperBaseTestCase):
def test_kerberos_login(self):
self.koji.koji_module.config.keytab = "testkeytab"
self.koji.koji_module.config.principal = "testprincipal"
self.koji.login()
self.koji.koji_proxy.gssapi_login.assert_called_with(
"testprincipal", "testkeytab"
)
def test_get_image_build_cmd_without_required_data(self):
with self.assertRaises(AssertionError):
self.koji.get_image_build_cmd(
{"image-build": {"name": "test-name"}}, self.tmpfile
)
def test_get_image_build_cmd_correct(self):
cmd = self.koji.get_image_build_cmd(
{
"image-build": {
"name": "test-name",
"version": "1",
"target": "test-target",
"install_tree": "/tmp/test/install_tree",
"arches": "x86_64",
"format": ["docker", "qcow2"],
"kickstart": "test-kickstart",
"ksurl": "git://example.com/ks.git",
"distro": "test-distro",
"release": "20160222.0",
"disk_size": 4,
}
},
self.tmpfile,
)
self.assertEqual(cmd[:3], ["koji", "--profile=custom-koji", "image-build"])
six.assertCountEqual(self, cmd[3:], ["--config=" + self.tmpfile, "--wait"])
with open(self.tmpfile, "r") as f:
lines = f.read().strip().split("\n")
self.assertEqual(lines[0], "[image-build]")
six.assertCountEqual(
self,
lines[1:],
[
"name = test-name",
"version = 1",
"target = test-target",
"install_tree = /tmp/test/install_tree",
"arches = x86_64",
"format = docker,qcow2",
"kickstart = test-kickstart",
"ksurl = git://example.com/ks.git",
"distro = test-distro",
"release = 20160222.0",
"disk_size = 4",
],
)
def test_get_image_paths(self):
# The data for this tests is obtained from the actual Koji build. It
# includes lots of fields that are not used, but for the sake of
# completeness is fully preserved.
getTaskChildren_data = {
12387273: [
{
"arch": "i386",
"awaited": False,
"channel_id": 12,
"completion_time": "2016-01-03 05:34:08.374262",
"completion_ts": 1451799248.37426,
"create_time": "2016-01-03 05:15:20.311599",
"create_ts": 1451798120.3116,
"host_id": 158,
"id": 12387276,
"label": "i386",
"method": "createImage",
"owner": 131,
"parent": 12387273,
"priority": 19,
"request": [
"Fedora-Cloud-Base",
"23",
"20160103",
"i386",
{
"build_tag": 299,
"build_tag_name": "f23-build",
"dest_tag": 294,
"dest_tag_name": "f23-updates-candidate",
"id": 144,
"name": "f23-candidate",
},
299,
{
"create_event": 14011966,
"create_ts": 1451761803.33528,
"creation_time": "2016-01-02 19:10:03.335283",
"id": 563977,
"state": 1,
},
"http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/i386/os/", # noqa: E501
{
"disk_size": "3",
"distro": "Fedora-20",
"format": ["qcow2", "raw-xz"],
"kickstart": "work/cli-image/1451798116.800155.wYJWTVHw/fedora-cloud-base-2878aa0.ks", # noqa: E501
"release": "20160103",
"repo": [
"http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/$arch/os/", # noqa: E501
"http://infrastructure.fedoraproject.org/pub/fedora/linux/updates/23/$arch/", # noqa: E501
],
"scratch": True,
},
],
"start_time": "2016-01-03 05:15:29.828081",
"start_ts": 1451798129.82808,
"state": 2,
"waiting": None,
"weight": 2.0,
},
{
"arch": "x86_64",
"awaited": False,
"channel_id": 12,
"completion_time": "2016-01-03 05:33:20.066366",
"completion_ts": 1451799200.06637,
"create_time": "2016-01-03 05:15:20.754201",
"create_ts": 1451798120.7542,
"host_id": 156,
"id": 12387277,
"label": "x86_64",
"method": "createImage",
"owner": 131,
"parent": 12387273,
"priority": 19,
"request": [
"Fedora-Cloud-Base",
"23",
"20160103",
"x86_64",
{
"build_tag": 299,
"build_tag_name": "f23-build",
"dest_tag": 294,
"dest_tag_name": "f23-updates-candidate",
"id": 144,
"name": "f23-candidate",
},
299,
{
"create_event": 14011966,
"create_ts": 1451761803.33528,
"creation_time": "2016-01-02 19:10:03.335283",
"id": 563977,
"state": 1,
},
"http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/x86_64/os/", # noqa: E501
{
"disk_size": "3",
"distro": "Fedora-20",
"format": ["qcow2", "raw-xz"],
"kickstart": "work/cli-image/1451798116.800155.wYJWTVHw/fedora-cloud-base-2878aa0.ks", # noqa: E501
"release": "20160103",
"repo": [
"http://infrastructure.fedoraproject.org/pub/alt/releases/23/Cloud/$arch/os/", # noqa: E501
"http://infrastructure.fedoraproject.org/pub/fedora/linux/updates/23/$arch/", # noqa: E501
],
"scratch": True,
},
],
"start_time": "2016-01-03 05:15:35.196043",
"start_ts": 1451798135.19604,
"state": 2,
"waiting": None,
"weight": 2.0,
},
]
}
getTaskResult_data = {
12387276: {
"arch": "i386",
"files": [
"tdl-i386.xml",
"fedora-cloud-base-2878aa0.ks",
"koji-f23-build-12387276-base.ks",
"libvirt-qcow2-i386.xml",
"Fedora-Cloud-Base-23-20160103.i386.qcow2",
"libvirt-raw-xz-i386.xml",
"Fedora-Cloud-Base-23-20160103.i386.raw.xz",
],
"logs": ["oz-i386.log"],
"name": "Fedora-Cloud-Base",
"release": "20160103",
"rpmlist": [],
"task_id": 12387276,
"version": "23",
},
12387277: {
"arch": "x86_64",
"files": [
"tdl-x86_64.xml",
"fedora-cloud-base-2878aa0.ks",
"koji-f23-build-12387277-base.ks",
"libvirt-qcow2-x86_64.xml",
"Fedora-Cloud-Base-23-20160103.x86_64.qcow2",
"libvirt-raw-xz-x86_64.xml",
"Fedora-Cloud-Base-23-20160103.x86_64.raw.xz",
],
"logs": ["oz-x86_64.log"],
"name": "Fedora-Cloud-Base",
"release": "20160103",
"rpmlist": [],
"task_id": 12387277,
"version": "23",
},
}
self.koji.koji_proxy = mock.Mock(
getTaskChildren=mock.Mock(
side_effect=lambda task_id, request: getTaskChildren_data.get(task_id)
),
getTaskResult=mock.Mock(
side_effect=lambda task_id: getTaskResult_data.get(task_id)
),
)
result = self.koji.get_image_paths(12387273)
six.assertCountEqual(self, result.keys(), ["i386", "x86_64"])
self.maxDiff = None
six.assertCountEqual(
self,
result["i386"],
[
"/koji/task/12387276/tdl-i386.xml",
"/koji/task/12387276/fedora-cloud-base-2878aa0.ks",
"/koji/task/12387276/koji-f23-build-12387276-base.ks",
"/koji/task/12387276/libvirt-qcow2-i386.xml",
"/koji/task/12387276/Fedora-Cloud-Base-23-20160103.i386.qcow2",
"/koji/task/12387276/libvirt-raw-xz-i386.xml",
"/koji/task/12387276/Fedora-Cloud-Base-23-20160103.i386.raw.xz",
],
)
six.assertCountEqual(
self,
result["x86_64"],
[
"/koji/task/12387277/tdl-x86_64.xml",
"/koji/task/12387277/fedora-cloud-base-2878aa0.ks",
"/koji/task/12387277/koji-f23-build-12387277-base.ks",
"/koji/task/12387277/libvirt-qcow2-x86_64.xml",
"/koji/task/12387277/Fedora-Cloud-Base-23-20160103.x86_64.qcow2",
"/koji/task/12387277/libvirt-raw-xz-x86_64.xml",
"/koji/task/12387277/Fedora-Cloud-Base-23-20160103.x86_64.raw.xz",
],
)
def test_get_image_paths_failed_subtask(self):
failed = set()
def failed_callback(arch):
failed.add(arch)
with open(os.path.join(FIXTURE_DIR, "task_children_25643870.json")) as f:
getTaskChildren_data = json.load(f)
with open(
os.path.join(FIXTURE_DIR, "children_task_results_25643870.json")
) as f:
getTaskResult_data = json.load(f)
self.koji.koji_proxy = mock.Mock(
getTaskChildren=mock.Mock(return_value=getTaskChildren_data),
getTaskResult=mock.Mock(
side_effect=lambda task_id: getTaskResult_data.get(str(task_id))
),
getImageBuild=mock.Mock(side_effect=lambda name: {}),
)
result = self.koji.get_image_paths(25643870, callback=failed_callback)
six.assertCountEqual(self, result.keys(), ["aarch64", "armhfp", "x86_64"])
six.assertCountEqual(self, failed, ["ppc64le", "s390x"])
def test_multicall_map(self):
self.koji.koji_proxy = mock.Mock()
self.koji.koji_proxy.multiCall.return_value = [[1], [2]]
ret = self.koji.multicall_map(
self.koji.koji_proxy,
self.koji.koji_proxy.getBuild,
["foo", "bar"],
[{"x": 1}, {"x": 2}],
)
six.assertCountEqual(
self,
self.koji.koji_proxy.getBuild.mock_calls,
[mock.call("foo", x=1), mock.call("bar", x=2)],
)
self.koji.koji_proxy.multiCall.assert_called_with(strict=True)
self.assertEqual(ret, [1, 2])
class LiveMediaTestCase(KojiWrapperBaseTestCase):
def test_get_live_media_cmd_minimal(self):
opts = {
"name": "name",
"version": "1",
"target": "tgt",
"arch": "x,y,z",
"ksfile": "kickstart",
"install_tree": "/mnt/os",
}
cmd = self.koji.get_live_media_cmd(opts)
self.assertEqual(
cmd,
[
"koji",
"--profile=custom-koji",
"spin-livemedia",
"name",
"1",
"tgt",
"x,y,z",
"kickstart",
"--install-tree=/mnt/os",
"--wait",
],
)
def test_get_live_media_cmd_full(self):
opts = {
"name": "name",
"version": "1",
"target": "tgt",
"arch": "x,y,z",
"ksfile": "kickstart",
"install_tree": "/mnt/os",
"scratch": True,
"repo": ["repo-1", "repo-2"],
"skip_tag": True,
"ksurl": "git://example.com/ksurl.git",
"release": "20160222.1",
}
cmd = self.koji.get_live_media_cmd(opts)
self.assertEqual(
cmd[:9],
[
"koji",
"--profile=custom-koji",
"spin-livemedia",
"name",
"1",
"tgt",
"x,y,z",
"kickstart",
"--install-tree=/mnt/os",
],
)
six.assertCountEqual(
self,
cmd[9:],
[
"--repo=repo-1",
"--repo=repo-2",
"--skip-tag",
"--scratch",
"--wait",
"--ksurl=git://example.com/ksurl.git",
"--release=20160222.1",
],
)
class LiveImageKojiWrapperTest(KojiWrapperBaseTestCase):
def test_get_create_image_cmd_minimal(self):
cmd = self.koji.get_create_image_cmd(
"my_name", "1.0", "f24-candidate", "x86_64", "/path/to/ks", ["/repo/1"]
)
self.assertEqual(cmd[0:3], ["koji", "--profile=custom-koji", "spin-livecd"])
six.assertCountEqual(
self, cmd[3:7], ["--noprogress", "--scratch", "--wait", "--repo=/repo/1"]
)
self.assertEqual(
cmd[7:], ["my_name", "1.0", "f24-candidate", "x86_64", "/path/to/ks"]
)
def test_get_create_image_cmd_full(self):
cmd = self.koji.get_create_image_cmd(
"my_name",
"1.0",
"f24-candidate",
"x86_64",
"/path/to/ks",
["/repo/1", "/repo/2"],
release="1",
wait=False,
archive=True,
specfile="foo.spec",
ksurl="https://git.example.com/",
)
self.assertEqual(cmd[0:3], ["koji", "--profile=custom-koji", "spin-livecd"])
self.assertEqual(
cmd[-5:], ["my_name", "1.0", "f24-candidate", "x86_64", "/path/to/ks"]
)
six.assertCountEqual(
self,
cmd[3:-5],
[
"--noprogress",
"--nowait",
"--repo=/repo/1",
"--repo=/repo/2",
"--release=1",
"--specfile=foo.spec",
"--ksurl=https://git.example.com/",
],
)
def test_spin_livecd_with_format(self):
with self.assertRaises(ValueError):
self.koji.get_create_image_cmd(
"my_name",
"1.0",
"f24-candidate",
"x86_64",
"/path/to/ks",
[],
image_format="qcow",
)
def test_spin_appliance_with_format(self):
cmd = self.koji.get_create_image_cmd(
"my_name",
"1.0",
"f24-candidate",
"x86_64",
"/path/to/ks",
[],
image_type="appliance",
image_format="qcow",
)
self.assertEqual(cmd[0:3], ["koji", "--profile=custom-koji", "spin-appliance"])
six.assertCountEqual(
self, cmd[3:7], ["--noprogress", "--scratch", "--wait", "--format=qcow"]
)
self.assertEqual(
cmd[7:], ["my_name", "1.0", "f24-candidate", "x86_64", "/path/to/ks"]
)
def test_spin_appliance_with_wrong_format(self):
with self.assertRaises(ValueError):
self.koji.get_create_image_cmd(
"my_name",
"1.0",
"f24-candidate",
"x86_64",
"/path/to/ks",
[],
image_type="appliance",
image_format="pretty",
)
@mock.patch.dict("os.environ", {"FOO": "BAR"}, clear=True)
class RunrootKojiWrapperTest(KojiWrapperBaseTestCase):
def test_get_cmd_minimal(self):
cmd = self.koji.get_runroot_cmd("tgt", "s390x", "date", use_shell=False)
self.assertEqual(len(cmd), 9)
self.assertEqual(
cmd[:5],
["koji", "--profile=custom-koji", "runroot", "--nowait", "--task-id"],
)
self.assertEqual(cmd[-3], "tgt")
self.assertEqual(cmd[-2], "s390x")
self.assertEqual(
cmd[-1], "rm -f /var/lib/rpm/__db*; rm -rf /var/cache/yum/*; set -x; date"
)
six.assertCountEqual(self, cmd[5:-3], ["--channel-override=runroot-local"])
def test_get_cmd_full(self):
cmd = self.koji.get_runroot_cmd(
"tgt",
"s390x",
["/bin/echo", "&"],
quiet=True,
channel="chan",
packages=["lorax", "some_other_package"],
mounts=["/tmp"],
weight=1000,
)
self.assertEqual(len(cmd), 15)
self.assertEqual(cmd[:3], ["koji", "--profile=custom-koji", "runroot"])
self.assertEqual(cmd[-3], "tgt")
self.assertEqual(cmd[-2], "s390x")
self.assertEqual(
cmd[-1],
"rm -f /var/lib/rpm/__db*; rm -rf /var/cache/yum/*; set -x; /bin/echo '&'",
)
six.assertCountEqual(
self,
cmd[4:-3],
[
"--channel-override=chan",
"--quiet",
"--use-shell",
"--task-id",
"--weight=1000",
"--package=some_other_package",
"--package=lorax",
"--mount=/tmp",
],
)
@mock.patch("os.getuid", new=lambda: 1010)
def test_with_chown_paths(self):
cmd = self.koji.get_runroot_cmd(
"tgt",
"s390x",
["/bin/echo", "&"],
quiet=True,
channel="chan",
packages=["lorax", "some_other_package"],
mounts=["/tmp"],
weight=1000,
chown_paths=["/output dir", "/foo"],
)
self.assertEqual(len(cmd), 15)
self.assertEqual(cmd[:3], ["koji", "--profile=custom-koji", "runroot"])
self.assertEqual(cmd[-3], "tgt")
self.assertEqual(cmd[-2], "s390x")
self.assertEqual(
cmd[-1],
"rm -f /var/lib/rpm/__db*; rm -rf /var/cache/yum/*; set -x; /bin/echo '&' && chmod -R a+r '/output dir' /foo && chown -R 1010 '/output dir' /foo", # noqa: E501
)
six.assertCountEqual(
self,
cmd[4:-3],
[
"--channel-override=chan",
"--quiet",
"--use-shell",
"--task-id",
"--weight=1000",
"--package=some_other_package",
"--package=lorax",
"--mount=/tmp",
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_run_runroot_cmd_no_task_id(self, run):
cmd = ["koji", "runroot"]
output = "Output ..."
run.return_value = (1, output)
with self.assertRaises(RuntimeError) as ctx:
self.koji.run_runroot_cmd(cmd)
self.assertEqual(
run.call_args_list,
[
mock.call(
cmd,
can_fail=True,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
logfile=None,
show_cmd=True,
universal_newlines=True,
)
],
)
self.assertEqual(
"Could not find task ID in output. Command '%s' returned '%s'."
% (" ".join(cmd), output),
str(ctx.exception),
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_run_runroot_cmd_with_task_id(self, run):
cmd = ["koji", "runroot", "--task-id"]
run.return_value = (0, "1234\n")
output = "Output ..."
self.koji._wait_for_task = mock.Mock(return_value=(0, output))
result = self.koji.run_runroot_cmd(cmd)
self.assertDictEqual(result, {"retcode": 0, "output": output, "task_id": 1234})
self.assertEqual(
run.call_args_list,
[
mock.call(
cmd,
can_fail=True,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
logfile=None,
show_cmd=True,
universal_newlines=True,
)
],
)
@mock.patch("shutil.rmtree")
@mock.patch("tempfile.mkdtemp")
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_run_runroot_cmd_with_keytab(self, run, mkdtemp, rmtree):
# We mock rmtree to avoid deleing something we did not create.
mkdtemp.return_value = "/tmp/foo"
self.koji.koji_module.config.keytab = "foo"
cmd = ["koji", "runroot"]
output = "Output ..."
run.return_value = (0, "1234\n")
self.koji._wait_for_task = mock.Mock(return_value=(0, output))
result = self.koji.run_runroot_cmd(cmd)
self.assertDictEqual(result, {"retcode": 0, "output": output, "task_id": 1234})
self.assertEqual(
run.call_args_list,
[
mock.call(
cmd,
can_fail=True,
env={
"KRB5CCNAME": "DIR:/tmp/foo",
"FOO": "BAR",
"PYTHONUNBUFFERED": "1",
},
buffer_size=-1,
logfile=None,
show_cmd=True,
universal_newlines=True,
)
],
)
@mock.patch.dict("os.environ", {"FOO": "BAR"}, clear=True)
class RunBlockingCmdTest(KojiWrapperBaseTestCase):
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_minimal(self, run):
output = "Created task: 1234\nHello\n"
run.return_value = (0, output)
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 0, "output": output, "task_id": 1234})
six.assertCountEqual(
self,
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
)
],
)
@mock.patch("pungi.util.temp_dir")
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_with_keytab(self, run, temp_dir):
temp_dir.return_value.__enter__.return_value = "/tmp/foo"
self.koji.koji_module.config.keytab = "foo"
output = "Created task: 1234\nHello\n"
run.return_value = (0, output)
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 0, "output": output, "task_id": 1234})
six.assertCountEqual(
self,
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={
"KRB5CCNAME": "DIR:/tmp/foo",
"FOO": "BAR",
"PYTHONUNBUFFERED": "1",
},
buffer_size=-1,
universal_newlines=True,
)
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_with_log(self, run):
output = "Created task: 1234\nHello\n"
run.return_value = (0, output)
result = self.koji.run_blocking_cmd("cmd", log_file="logfile")
self.assertDictEqual(result, {"retcode": 0, "output": output, "task_id": 1234})
six.assertCountEqual(
self,
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile="logfile",
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
)
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_fail_with_task_id(self, run):
output = "Created task: 1234\nBoom\n"
run.return_value = (1, output)
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 1, "output": output, "task_id": 1234})
six.assertCountEqual(
self,
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
)
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_fail_without_task_id(self, run):
output = "Not found\n"
run.return_value = (1, output)
with self.assertRaises(RuntimeError) as ctx:
self.koji.run_blocking_cmd("cmd")
six.assertCountEqual(
self,
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
)
],
)
self.assertIn("Could not find task ID", str(ctx.exception))
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_disconnect_and_retry(self, run):
output = "Created task: 1234\nerror: failed to connect\n"
retry = "Created task: 1234\nOook\n"
run.side_effect = [(1, output), (0, retry)]
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 0, "output": retry, "task_id": 1234})
self.assertEqual(
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_disconnect_and_retry_but_fail(self, run):
output = "Created task: 1234\nerror: failed to connect\n"
retry = "Created task: 1234\nNot working still\n"
run.side_effect = [(1, output), (1, retry)]
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 1, "output": retry, "task_id": 1234})
self.assertEqual(
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
],
)
@mock.patch("time.sleep")
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_disconnect_and_retry_multiple_times(self, run, sleep):
output = "Created task: 1234\nerror: failed to connect\n"
retry = "Created task: 1234\nOK\n"
run.side_effect = [(1, output), (1, output), (1, output), (0, retry)]
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 0, "output": retry, "task_id": 1234})
self.assertEqual(
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
],
)
self.assertEqual(sleep.mock_calls, [mock.call(i * 10) for i in range(1, 3)])
@mock.patch("time.sleep")
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_disconnect_and_never_reconnect(self, run, sleep):
output = "Created task: 1234\nerror: failed to connect\n"
run.side_effect = [(1, output), (1, output), (1, output), (1, output)]
with self.assertRaises(RuntimeError) as ctx:
self.koji.run_blocking_cmd("cmd", max_retries=2)
self.assertIn("Failed to wait", str(ctx.exception))
self.assertEqual(
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
],
)
self.assertEqual(sleep.mock_calls, [mock.call(i * 10) for i in range(1, 2)])
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_server_offline_and_retry(self, run):
output = "Created task: 1234\nkoji: ServerOffline:"
retry = "Created task: 1234\nOook\n"
run.side_effect = [(1, output), (0, retry)]
result = self.koji.run_blocking_cmd("cmd")
self.assertDictEqual(result, {"retcode": 0, "output": retry, "task_id": 1234})
self.assertEqual(
run.mock_calls,
[
mock.call(
"cmd",
can_fail=True,
logfile=None,
env={"FOO": "BAR", "PYTHONUNBUFFERED": "1"},
buffer_size=-1,
universal_newlines=True,
),
mock.call(
["koji", "--profile=custom-koji", "watch-task", "1234"],
can_fail=True,
logfile=None,
universal_newlines=True,
),
],
)
def test_get_pungi_buildinstall_cmd(self):
args = {"product": "Fedora 23"}
cmd = self.koji.get_pungi_buildinstall_cmd(
"rrt",
"x86_64",
args,
channel=None,
packages=["lorax"],
mounts=["/tmp"],
weight=123,
chown_uid=456,
)
expected_cmd = [
"koji",
"--profile=custom-koji",
"pungi-buildinstall",
"--nowait",
"--task-id",
"--channel-override=runroot-local",
"--weight=123",
"--package=lorax",
"--mount=/tmp",
"--chown-uid=456",
"rrt",
"x86_64",
"product=Fedora 23",
]
self.assertEqual(cmd, expected_cmd)
def test_get_pungi_ostree_cmd(self):
args = {"product": "Fedora 23"}
cmd = self.koji.get_pungi_ostree_cmd(
"rrt",
"x86_64",
args,
channel=None,
packages=["lorax"],
mounts=["/tmp"],
weight=123,
)
expected_cmd = [
"koji",
"--profile=custom-koji",
"pungi-ostree",
"--nowait",
"--task-id",
"--channel-override=runroot-local",
"--weight=123",
"--package=lorax",
"--mount=/tmp",
"rrt",
"x86_64",
"product=Fedora 23",
]
self.assertEqual(cmd, expected_cmd)
RPM_QA_QF_OUTPUT = """
cjkuni-uming-fonts-0.2.20080216.1-56.fc23.noarch
libmount-2.28-1.fc23.x86_64
ed-1.10-5.fc23.x86_64
kbd-2.0.2-8.fc23.x86_64
coreutils-8.24-6.fc23.x86_64
"""
BUILDROOT_LIST = [
{
"arch": "x86_64",
"br_type": 0,
"cg_id": None,
"cg_name": None,
"cg_version": None,
"container_arch": "x86_64",
"container_type": "chroot",
"create_event_id": 15862222,
"create_event_time": "2016-04-28 02:37:00.949772",
"create_ts": 1461811020.94977,
"extra": None,
"host_arch": None,
"host_id": 99,
"host_name": "buildhw-01.phx2.fedoraproject.org",
"host_os": None,
"id": 5458481,
"repo_create_event_id": 15861452,
"repo_create_event_time": "2016-04-28 00:02:40.639317",
"repo_id": 599173,
"repo_state": 1,
"retire_event_id": 15862276,
"retire_event_time": "2016-04-28 02:58:07.109387",
"retire_ts": 1461812287.10939,
"state": 3,
"tag_id": 315,
"tag_name": "f24-build",
"task_id": 13831904,
}
]
RPM_LIST = [
{
"arch": "noarch",
"build_id": 756072,
"buildroot_id": 5398084,
"buildtime": 1461100903,
"component_buildroot_id": 5458481,
"epoch": None,
"external_repo_id": 0,
"external_repo_name": "INTERNAL",
"extra": None,
"id": 7614370,
"is_update": True,
"metadata_only": False,
"name": "python3-kickstart",
"nvr": "python3-kickstart-2.25-2.fc24",
"payloadhash": "403723502d27e43955036d2dcd1b09e0",
"release": "2.fc24",
"size": 366038,
"version": "2.25",
},
{
"arch": "x86_64",
"build_id": 756276,
"buildroot_id": 5405310,
"buildtime": 1461165155,
"component_buildroot_id": 5458481,
"epoch": None,
"external_repo_id": 0,
"external_repo_name": "INTERNAL",
"extra": None,
"id": 7615629,
"is_update": False,
"metadata_only": False,
"name": "binutils",
"nvr": "binutils-2.26-18.fc24",
"payloadhash": "8ef08c8a64c52787d3559424e5f51d9d",
"release": "18.fc24",
"size": 6172094,
"version": "2.26",
},
{
"arch": "x86_64",
"build_id": 756616,
"buildroot_id": 5412029,
"buildtime": 1461252071,
"component_buildroot_id": 5458481,
"epoch": None,
"external_repo_id": 0,
"external_repo_name": "INTERNAL",
"extra": None,
"id": 7619636,
"is_update": False,
"metadata_only": False,
"name": "kernel-headers",
"nvr": "kernel-headers-4.5.2-301.fc24",
"payloadhash": "11c6d70580c8f0c202c28bc6b0fa98cc",
"release": "301.fc24",
"size": 1060138,
"version": "4.5.2",
},
]
class TestGetBuildrootRPMs(unittest.TestCase):
@mock.patch("pungi.wrappers.kojiwrapper.KojiWrapper")
def test_get_from_koji(self, KojiWrapper):
compose = mock.Mock(conf={"koji_profile": "koji"})
KojiWrapper.return_value.koji_proxy.listBuildroots.return_value = BUILDROOT_LIST
KojiWrapper.return_value.koji_proxy.listRPMs.return_value = RPM_LIST
rpms = get_buildroot_rpms(compose, 1234)
self.assertEqual(KojiWrapper.call_args_list, [mock.call("koji")])
self.assertEqual(
KojiWrapper.return_value.mock_calls,
[
mock.call.koji_proxy.listBuildroots(taskID=1234),
mock.call.koji_proxy.listRPMs(componentBuildrootID=5458481),
],
)
six.assertCountEqual(
self,
rpms,
[
"python3-kickstart-2.25-2.fc24.noarch",
"binutils-2.26-18.fc24.x86_64",
"kernel-headers-4.5.2-301.fc24.x86_64",
],
)
@mock.patch("pungi.wrappers.kojiwrapper.run")
def test_get_local(self, mock_run):
compose = mock.Mock()
mock_run.return_value = (0, RPM_QA_QF_OUTPUT)
rpms = get_buildroot_rpms(compose, None)
six.assertCountEqual(
self,
rpms,
[
"cjkuni-uming-fonts-0.2.20080216.1-56.fc23.noarch",
"libmount-2.28-1.fc23.x86_64",
"ed-1.10-5.fc23.x86_64",
"kbd-2.0.2-8.fc23.x86_64",
"coreutils-8.24-6.fc23.x86_64",
],
)
Reference in New Issue View Git Blame Copy Permalink