mirror of
https://pagure.io/fm-orchestrator.git
synced 2026-02-03 05:03:43 +08:00
239 lines
8.9 KiB
Python
239 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
# SPDX-License-Identifier: MIT
|
|
from __future__ import absolute_import
|
|
|
|
import logging
|
|
from mock import patch
|
|
import pytest
|
|
|
|
from module_build_service import app, manage as mbs_manager
|
|
from module_build_service.common import models
|
|
from module_build_service.common.models import BUILD_STATES, ModuleBuild
|
|
from module_build_service.scheduler.db_session import db_session
|
|
from module_build_service.web.utils import deps_to_dict
|
|
from tests import clean_database, staged_data_filename
|
|
|
|
|
|
@pytest.mark.usefixtures("provide_test_data")
|
|
class TestMBSManage:
|
|
|
|
@pytest.mark.parametrize(
|
|
("identifier", "is_valid"),
|
|
(
|
|
("", False),
|
|
("spam", False),
|
|
("spam:bacon", True),
|
|
("spam:bacon:eggs", True),
|
|
("spam:bacon:eggs:ham", True),
|
|
("spam:bacon:eggs:ham:sausage", False),
|
|
),
|
|
)
|
|
def test_retire_identifier_validation(self, identifier, is_valid):
|
|
if is_valid:
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
mbs_manager.cli(["retire", identifier])
|
|
assert 0 == exc_info
|
|
else:
|
|
with pytest.raises(ValueError):
|
|
mbs_manager.cli(["retire", identifier])
|
|
|
|
@pytest.mark.parametrize(
|
|
("overrides", "identifier", "changed_count"),
|
|
(
|
|
({"name": "pickme"}, "pickme:eggs", 1),
|
|
({"stream": "pickme"}, "spam:pickme", 1),
|
|
({"version": "pickme"}, "spam:eggs:pickme", 1),
|
|
({"context": "pickme"}, "spam:eggs:ham:pickme", 1),
|
|
({}, "spam:eggs", 2),
|
|
({"version": "pickme"}, "spam:eggs", 2),
|
|
({"context": "pickme"}, "spam:eggs:ham", 2),
|
|
),
|
|
)
|
|
@patch("click.confirm")
|
|
def test_retire_build(self, confirm, overrides, identifier, changed_count):
|
|
confirm.return_value = True
|
|
|
|
module_builds = (
|
|
db_session.query(ModuleBuild)
|
|
.filter_by(state=BUILD_STATES["ready"])
|
|
.order_by(ModuleBuild.id.desc())
|
|
.all()
|
|
)
|
|
# Verify our assumption of the amount of 'ready' ModuleBuilds in database
|
|
assert len(module_builds) == 2
|
|
|
|
for x, build in enumerate(module_builds):
|
|
build.name = "spam"
|
|
build.stream = "eggs"
|
|
build.version = "ham"
|
|
build.context = str(x)
|
|
|
|
for attr, value in overrides.items():
|
|
setattr(module_builds[0], attr, value)
|
|
|
|
db_session.commit()
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
mbs_manager.cli(["retire", identifier])
|
|
assert 0 == exc_info.value
|
|
|
|
retired_module_builds = (
|
|
db_session.query(ModuleBuild)
|
|
.filter_by(state=BUILD_STATES["garbage"])
|
|
.order_by(ModuleBuild.id.desc())
|
|
.all()
|
|
)
|
|
|
|
assert len(retired_module_builds) == changed_count
|
|
for x in range(changed_count):
|
|
assert retired_module_builds[x].id == module_builds[x].id
|
|
assert retired_module_builds[x].state == BUILD_STATES["garbage"]
|
|
|
|
@pytest.mark.parametrize(
|
|
("confirm_prompt", "confirm_arg", "confirm_expected"),
|
|
(
|
|
(True, False, True),
|
|
(True, True, True),
|
|
(False, False, False),
|
|
(False, True, True)
|
|
),
|
|
)
|
|
@patch("click.confirm")
|
|
def test_retire_build_confirm_prompt(
|
|
self, confirm, confirm_prompt, confirm_arg, confirm_expected
|
|
):
|
|
confirm.return_value = confirm_prompt
|
|
|
|
module_builds = db_session.query(ModuleBuild).filter_by(state=BUILD_STATES["ready"]).all()
|
|
# Verify our assumption of the amount of ModuleBuilds in database
|
|
assert len(module_builds) == 2
|
|
|
|
for x, build in enumerate(module_builds):
|
|
build.name = "spam" + str(x) if x > 0 else "spam"
|
|
build.stream = "eggs"
|
|
db_session.commit()
|
|
|
|
cmd = ["retire", "spam:eggs"] + (["--confirm"] if confirm_arg else [])
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
mbs_manager.cli(cmd)
|
|
assert 0 == exc_info.value
|
|
|
|
expected_changed_count = 1 if confirm_expected else 0
|
|
retired_module_builds = (
|
|
db_session.query(ModuleBuild).filter_by(state=BUILD_STATES["garbage"]).all()
|
|
)
|
|
assert len(retired_module_builds) == expected_changed_count
|
|
|
|
|
|
class TestCommandBuildModuleLocally:
|
|
"""Test mbs-manager subcommand build_module_locally"""
|
|
|
|
def setup_method(self, test_method):
|
|
clean_database()
|
|
|
|
# Do not allow flask_script exits by itself because we have to assert
|
|
# something after the command finishes.
|
|
self.sys_exit_patcher = patch("sys.exit")
|
|
self.mock_sys_exit = self.sys_exit_patcher.start()
|
|
|
|
# The consumer is not required to run actually, so it does not make
|
|
# sense to publish message after creating a module build.
|
|
self.publish_patcher = patch("module_build_service.common.messaging.publish")
|
|
self.mock_publish = self.publish_patcher.start()
|
|
|
|
# Don't allow conf.set_item call to modify conf actually inside command
|
|
self.set_item_patcher = patch("module_build_service.manage.conf.set_item")
|
|
self.mock_set_item = self.set_item_patcher.start()
|
|
|
|
# Avoid to create the local sqlite database for the command, which is
|
|
# useless for running tests here.
|
|
self.create_all_patcher = patch("module_build_service.manage.db.create_all")
|
|
self.mock_create_all = self.create_all_patcher.start()
|
|
|
|
def teardown_method(self, test_method):
|
|
self.create_all_patcher.stop()
|
|
self.mock_set_item.stop()
|
|
self.publish_patcher.stop()
|
|
self.sys_exit_patcher.stop()
|
|
|
|
def _run_manager_wrapper(self, cli_cmd):
|
|
# build_module_locally changes database uri to a local SQLite database file.
|
|
# Restore the uri to original one in order to not impact the database
|
|
# session in subsequent tests.
|
|
original_db_uri = app.config["SQLALCHEMY_DATABASE_URI"]
|
|
try:
|
|
with patch("sys.argv", new=cli_cmd):
|
|
mbs_manager.cli()
|
|
finally:
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = original_db_uri
|
|
|
|
@patch("module_build_service.scheduler.local.main")
|
|
def test_set_stream(self, main):
|
|
cli_cmd = [
|
|
"mbs-manager", "build_module_locally",
|
|
"--set-stream", "platform:f28",
|
|
"--file", staged_data_filename("testmodule-local-build.yaml")
|
|
]
|
|
|
|
self._run_manager_wrapper(cli_cmd)
|
|
|
|
# Since module_build_service.scheduler.local.main is mocked, MBS does
|
|
# not really build the testmodule for this test. Following lines assert
|
|
# the fact:
|
|
# Module testmodule-local-build is expanded and stored into database,
|
|
# and this build has buildrequires platform:f28 and requires
|
|
# platform:f28.
|
|
# Please note that, the f28 is specified from command line option
|
|
# --set-stream, which is the point this test tests.
|
|
|
|
builds = db_session.query(models.ModuleBuild).filter_by(
|
|
name="testmodule-local-build").all()
|
|
assert 1 == len(builds)
|
|
|
|
testmodule_build = builds[0]
|
|
mmd_deps = testmodule_build.mmd().get_dependencies()
|
|
|
|
deps_dict = deps_to_dict(mmd_deps[0], "buildtime")
|
|
assert ["f28"] == deps_dict["platform"]
|
|
deps_dict = deps_to_dict(mmd_deps[0], "runtime")
|
|
assert ["f28"] == deps_dict["platform"]
|
|
|
|
@patch("module_build_service.manage.logging")
|
|
def test_ambiguous_stream(self, logging):
|
|
cli_cmd = [
|
|
"mbs-manager", "build_module_locally",
|
|
"--file", staged_data_filename("testmodule-local-build.yaml")
|
|
]
|
|
|
|
self._run_manager_wrapper(cli_cmd)
|
|
|
|
args, _ = logging.error.call_args_list[0]
|
|
assert "There are multiple streams to choose from for module platform." == args[0]
|
|
args, _ = logging.error.call_args_list[1]
|
|
assert "Use '-s module_name:module_stream' to choose the stream" == args[0]
|
|
|
|
def test_module_build_failed(self, caplog):
|
|
cli_cmd = [
|
|
"mbs-manager", "build_module_locally",
|
|
"--set-stream", "platform:f28",
|
|
"--file", staged_data_filename("testmodule-local-build.yaml")
|
|
]
|
|
|
|
def init_side_effect(*args):
|
|
build = db_session.query(models.ModuleBuild).filter(
|
|
models.ModuleBuild.name == "testmodule-local-build"
|
|
).first()
|
|
build.state = models.BUILD_STATES["failed"]
|
|
db_session.commit()
|
|
|
|
# We don't run consumer actually, but it could be patched to mark some
|
|
# module build failed for test purpose.
|
|
|
|
with patch("module_build_service.scheduler.local.modules_init_handler",
|
|
side_effect=init_side_effect):
|
|
self._run_manager_wrapper(cli_cmd)
|
|
|
|
r = [r for r in caplog.records if r.levelno == logging.ERROR
|
|
and "Local module build failed" in r.getMessage()]
|
|
assert len(r) > 0
|