mirror of
https://pagure.io/fm-orchestrator.git
synced 2026-02-03 05:03:43 +08:00
cff90e587a
Since we never pass in subprocess.PIPE for stdin/stdout/stderr, subprocess.communicate() does nothing, and the return out/err tuple was always empty.
387 lines
12 KiB
Python
387 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# SPDX-License-Identifier: MIT
|
|
from __future__ import absolute_import
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
from mock import call, MagicMock, Mock, patch, PropertyMock
|
|
import pytest
|
|
|
|
from module_build_service.builder import utils
|
|
from module_build_service.common import models
|
|
from module_build_service.common.config import conf
|
|
from module_build_service.common.errors import ProgrammingError, ValidationError
|
|
from module_build_service.common.utils import load_mmd, import_mmd, mmd_to_str
|
|
from module_build_service.scheduler.db_session import db_session
|
|
from tests import read_staged_data, scheduler_init_data
|
|
|
|
|
|
def test_execute_cmd(tmpdir, caplog):
|
|
logfile = str(tmpdir / "out.log")
|
|
with open(logfile, "w") as f:
|
|
utils.execute_cmd(["echo", "hello"], output=f)
|
|
|
|
with open(logfile) as f:
|
|
assert f.read() == "hello\n"
|
|
|
|
assert 'Executing the command "echo hello", output log: %s' % logfile in caplog.text
|
|
|
|
|
|
def test_execute_cmd_fail():
|
|
with pytest.raises(RuntimeError):
|
|
utils.execute_cmd(["false"])
|
|
|
|
|
|
@pytest.mark.parametrize("variation", ("none", "empty", "already_downloaded"))
|
|
@patch("requests.get")
|
|
@patch("koji.ClientSession")
|
|
@patch("module_build_service.builder.utils.execute_cmd")
|
|
def test_create_local_repo_from_koji_tag(mock_exec_cmd, mock_koji_session, mock_get, variation):
|
|
session = Mock()
|
|
rpms = [
|
|
{
|
|
"arch": "src",
|
|
"build_id": 875991,
|
|
"name": "module-build-macros",
|
|
"release": "1.module_92011fe6",
|
|
"size": 42,
|
|
"version": "0.1",
|
|
},
|
|
{
|
|
"arch": "noarch",
|
|
"build_id": 875991,
|
|
"name": "module-build-macros",
|
|
"release": "1.module_92011fe6",
|
|
"size": 42,
|
|
"version": "0.1",
|
|
},
|
|
{
|
|
"arch": "x86_64",
|
|
"build_id": 875636,
|
|
"name": "ed-debuginfo",
|
|
"release": "2.module_bd6e0eb1",
|
|
"size": 42,
|
|
"version": "1.14.1",
|
|
},
|
|
{
|
|
"arch": "x86_64",
|
|
"build_id": 875636,
|
|
"name": "ed",
|
|
"release": "2.module_bd6e0eb1",
|
|
"size": 42,
|
|
"version": "1.14.1",
|
|
},
|
|
{
|
|
"arch": "x86_64",
|
|
"build_id": 875640,
|
|
"name": "mksh-debuginfo",
|
|
"release": "2.module_bd6e0eb1",
|
|
"size": 42,
|
|
"version": "54",
|
|
},
|
|
{
|
|
"arch": "x86_64",
|
|
"build_id": 875640,
|
|
"name": "mksh",
|
|
"release": "2.module_bd6e0eb1",
|
|
"size": 42,
|
|
"version": "54",
|
|
},
|
|
]
|
|
|
|
builds = [
|
|
{
|
|
"build_id": 875640,
|
|
"name": "mksh",
|
|
"release": "2.module_bd6e0eb1",
|
|
"version": "54",
|
|
"volume_name": "prod",
|
|
},
|
|
{
|
|
"build_id": 875636,
|
|
"name": "ed",
|
|
"release": "2.module_bd6e0eb1",
|
|
"version": "1.14.1",
|
|
"volume_name": "prod",
|
|
},
|
|
{
|
|
"build_id": 875991,
|
|
"name": "module-build-macros",
|
|
"release": "1.module_92011fe6",
|
|
"version": "0.1",
|
|
"volume_name": "prod",
|
|
},
|
|
]
|
|
|
|
url_one = (
|
|
"https://kojipkgs.stg.fedoraproject.org//vol/prod/packages/module-build-macros/"
|
|
"0.1/1.module_92011fe6/noarch/module-build-macros-0.1-1.module_92011fe6.noarch.rpm"
|
|
)
|
|
url_two = (
|
|
"https://kojipkgs.stg.fedoraproject.org//vol/prod/packages/ed/1.14.1/"
|
|
"2.module_bd6e0eb1/x86_64/ed-1.14.1-2.module_bd6e0eb1.x86_64.rpm"
|
|
)
|
|
url_three = (
|
|
"https://kojipkgs.stg.fedoraproject.org//vol/prod/packages/mksh/54/"
|
|
"2.module_bd6e0eb1/x86_64/mksh-54-2.module_bd6e0eb1.x86_64.rpm"
|
|
)
|
|
|
|
if variation == "empty":
|
|
rpms = []
|
|
builds = []
|
|
|
|
session.listTaggedRPMS.return_value = (rpms, builds)
|
|
session.opts = {"topurl": "https://kojipkgs.stg.fedoraproject.org/"}
|
|
mock_koji_session.return_value = session
|
|
|
|
tag = "module-testmodule-master-20170405123740-build"
|
|
temp_dir = tempfile.mkdtemp()
|
|
try:
|
|
if variation == "already_downloaded":
|
|
for url in (url_one, url_two, url_three):
|
|
print(os.path.join(temp_dir, os.path.basename(url)))
|
|
with open(os.path.join(temp_dir, os.path.basename(url)), "w") as f:
|
|
f.write("x" * 42)
|
|
|
|
utils.create_local_repo_from_koji_tag(conf, tag, temp_dir)
|
|
finally:
|
|
shutil.rmtree(temp_dir)
|
|
|
|
if variation == "none":
|
|
expected_calls = [
|
|
call(url_one, stream=True, timeout=60),
|
|
call(url_two, stream=True, timeout=60),
|
|
call(url_three, stream=True, timeout=60),
|
|
]
|
|
else:
|
|
expected_calls = []
|
|
|
|
for expected_call in expected_calls:
|
|
assert expected_call in mock_get.call_args_list
|
|
assert len(mock_get.call_args_list) == len(expected_calls)
|
|
|
|
|
|
def test_validate_koji_tag_wrong_tag_arg_during_programming():
|
|
""" Test that we fail on a wrong param name (non-existing one) due to
|
|
programming error. """
|
|
|
|
@utils.validate_koji_tag("wrong_tag_arg")
|
|
def validate_koji_tag_programming_error(good_tag_arg, other_arg):
|
|
pass
|
|
|
|
with pytest.raises(ProgrammingError):
|
|
validate_koji_tag_programming_error("dummy", "other_val")
|
|
|
|
|
|
def test_validate_koji_tag_bad_tag_value():
|
|
""" Test that we fail on a bad tag value. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_bad_tag_value(tag_arg):
|
|
pass
|
|
|
|
with pytest.raises(ValidationError):
|
|
validate_koji_tag_bad_tag_value("forbiddentagprefix-foo")
|
|
|
|
|
|
def test_validate_koji_tag_bad_tag_value_in_list():
|
|
""" Test that we fail on a list containing bad tag value. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_bad_tag_value_in_list(tag_arg):
|
|
pass
|
|
|
|
with pytest.raises(ValidationError):
|
|
validate_koji_tag_bad_tag_value_in_list(["module-foo", "forbiddentagprefix-bar"])
|
|
|
|
|
|
def test_validate_koji_tag_good_tag_value():
|
|
""" Test that we pass on a good tag value. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_good_tag_value(tag_arg):
|
|
return True
|
|
|
|
assert validate_koji_tag_good_tag_value("module-foo") is True
|
|
|
|
|
|
def test_validate_koji_tag_good_tag_values_in_list():
|
|
""" Test that we pass on a list of good tag values. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_good_tag_values_in_list(tag_arg):
|
|
return True
|
|
|
|
assert validate_koji_tag_good_tag_values_in_list(["module-foo", "module-bar"]) is True
|
|
|
|
|
|
def test_validate_koji_tag_good_tag_value_in_dict():
|
|
""" Test that we pass on a dict arg with default key
|
|
and a good value. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_good_tag_value_in_dict(tag_arg):
|
|
return True
|
|
|
|
assert validate_koji_tag_good_tag_value_in_dict({"name": "module-foo"}) is True
|
|
|
|
|
|
def test_validate_koji_tag_good_tag_value_in_dict_nondefault_key():
|
|
""" Test that we pass on a dict arg with non-default key
|
|
and a good value. """
|
|
|
|
@utils.validate_koji_tag("tag_arg", dict_key="nondefault")
|
|
def validate_koji_tag_good_tag_value_in_dict_nondefault_key(tag_arg):
|
|
return True
|
|
|
|
assert (
|
|
validate_koji_tag_good_tag_value_in_dict_nondefault_key({"nondefault": "module-foo"})
|
|
is True
|
|
)
|
|
|
|
|
|
def test_validate_koji_tag_double_trouble_good():
|
|
""" Test that we pass on a list of tags that are good. """
|
|
|
|
expected = "foo"
|
|
|
|
@utils.validate_koji_tag(["tag_arg1", "tag_arg2"])
|
|
def validate_koji_tag_double_trouble(tag_arg1, tag_arg2):
|
|
return expected
|
|
|
|
actual = validate_koji_tag_double_trouble("module-1", "module-2")
|
|
assert actual == expected
|
|
|
|
|
|
def test_validate_koji_tag_double_trouble_bad():
|
|
""" Test that we fail on a list of tags that are bad. """
|
|
|
|
@utils.validate_koji_tag(["tag_arg1", "tag_arg2"])
|
|
def validate_koji_tag_double_trouble(tag_arg1, tag_arg2):
|
|
pass
|
|
|
|
with pytest.raises(ValidationError):
|
|
validate_koji_tag_double_trouble("module-1", "BADNEWS-2")
|
|
|
|
|
|
def test_validate_koji_tag_is_None():
|
|
""" Test that we fail on a tag which is None. """
|
|
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_is_None(tag_arg):
|
|
pass
|
|
|
|
with pytest.raises(ValidationError) as cm:
|
|
validate_koji_tag_is_None(None)
|
|
assert str(cm.value).endswith(" No value provided.") is True
|
|
|
|
|
|
@patch(
|
|
"module_build_service.common.config.Config.allowed_privileged_module_names",
|
|
new_callable=PropertyMock,
|
|
return_value=["testmodule"],
|
|
)
|
|
def test_validate_koji_tag_previleged_module_name(conf_apmn):
|
|
@utils.validate_koji_tag("tag_arg")
|
|
def validate_koji_tag_priv_mod_name(self, tag_arg):
|
|
pass
|
|
|
|
builder = MagicMock()
|
|
builder.module_str = 'testmodule'
|
|
validate_koji_tag_priv_mod_name(builder, "abc")
|
|
|
|
|
|
@pytest.mark.parametrize("provide_test_data", [{"contexts": True}], indirect=True)
|
|
def test_get_rpm_release_mse(provide_test_data):
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
release_one = utils.get_rpm_release(db_session, build_one)
|
|
assert release_one == "module+2+b8645bbb"
|
|
|
|
build_two = models.ModuleBuild.get_by_id(db_session, 3)
|
|
release_two = utils.get_rpm_release(db_session, build_two)
|
|
assert release_two == "module+2+17e35784"
|
|
|
|
|
|
def test_get_rpm_release_platform_stream():
|
|
scheduler_init_data(1)
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
release = utils.get_rpm_release(db_session, build_one)
|
|
assert release == "module+f28+2+814cfa39"
|
|
|
|
|
|
def test_get_rpm_release_platform_stream_override():
|
|
scheduler_init_data(1)
|
|
|
|
# Set the disttag_marking override on the platform
|
|
platform = (
|
|
db_session.query(models.ModuleBuild)
|
|
.filter_by(name="platform", stream="f28")
|
|
.first()
|
|
)
|
|
platform_mmd = platform.mmd()
|
|
platform_xmd = platform_mmd.get_xmd()
|
|
platform_xmd["mbs"]["disttag_marking"] = "fedora28"
|
|
platform_mmd.set_xmd(platform_xmd)
|
|
platform.modulemd = mmd_to_str(platform_mmd)
|
|
db_session.add(platform)
|
|
db_session.commit()
|
|
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
release = utils.get_rpm_release(db_session, build_one)
|
|
assert release == "module+fedora28+2+814cfa39"
|
|
|
|
|
|
@patch(
|
|
"module_build_service.common.config.Config.allowed_privileged_module_names",
|
|
new_callable=PropertyMock,
|
|
return_value=["build"],
|
|
)
|
|
def test_get_rpm_release_metadata_br_stream_override(mock_admmn):
|
|
"""
|
|
Test that when a module buildrequires a module in conf.allowed_privileged_module_names,
|
|
and that module has the xmd.mbs.disttag_marking field set, it should influence the disttag.
|
|
"""
|
|
scheduler_init_data(1)
|
|
metadata_mmd = load_mmd(read_staged_data("build_metadata_module"))
|
|
import_mmd(db_session, metadata_mmd)
|
|
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
mmd = build_one.mmd()
|
|
deps = mmd.get_dependencies()[0]
|
|
deps.add_buildtime_stream("build", "product1.2")
|
|
xmd = mmd.get_xmd()
|
|
xmd["mbs"]["buildrequires"]["build"] = {
|
|
"filtered_rpms": [],
|
|
"ref": "virtual",
|
|
"stream": "product1.2",
|
|
"version": "1",
|
|
"context": "00000000",
|
|
}
|
|
mmd.set_xmd(xmd)
|
|
build_one.modulemd = mmd_to_str(mmd)
|
|
db_session.add(build_one)
|
|
db_session.commit()
|
|
|
|
release = utils.get_rpm_release(db_session, build_one)
|
|
assert release == "module+product12+2+814cfa39"
|
|
|
|
|
|
@pytest.mark.parametrize("provide_test_data",
|
|
[{"contexts": True, "scratch": True}], indirect=True)
|
|
def test_get_rpm_release_mse_scratch(provide_test_data):
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
release_one = utils.get_rpm_release(db_session, build_one)
|
|
assert release_one == "scrmod+2+b8645bbb"
|
|
|
|
build_two = models.ModuleBuild.get_by_id(db_session, 3)
|
|
release_two = utils.get_rpm_release(db_session, build_two)
|
|
assert release_two == "scrmod+2+17e35784"
|
|
|
|
|
|
def test_get_rpm_release_platform_stream_scratch():
|
|
scheduler_init_data(1, scratch=True)
|
|
build_one = models.ModuleBuild.get_by_id(db_session, 2)
|
|
release = utils.get_rpm_release(db_session, build_one)
|
|
assert release == "scrmod+f28+2+814cfa39"
|