arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-11-19 23:13:41 +00:00
Code Issues Projects Releases Wiki Activity

feat: allow to use one application for multiple repositories (#111)

Browse Source 
* allow to use one application for multiple repositories

* update tests

* handle None append argument everywhere

* rewrite repository definition logic

* drop optional flags from docs

* support of new schema in systemd units

* add migration docs and ability to migrate tree automatically

* use repostory id instead

* verbose multiarchitectureerror

* object path support for s3 sync

* fix tests after rebase
This commit is contained in:
Evgenii Alekseev
2023-09-08 03:42:28 +03:00
parent 99eecdebf3
commit 59356e905a
191 changed files with 3441 additions and 1319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
tests/ahriman/application/handlers/test_handler.py
View File

@ -8,48 +8,7 @@ from ahriman.application.handlers import Handler
from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import ExitCode, MissingArchitectureError, MultipleArchitecturesError
from ahriman.models.log_handler import LogHandler
def test_architectures_extract(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
"""
must generate list of available architectures
"""
args.configuration = configuration.path
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
Handler.architectures_extract(args)
known_architectures_mock.assert_called_once_with(configuration.getpath("repository", "root"))
def test_architectures_extract_empty(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must raise exception if no available architectures found
"""
args.command = "config"
args.configuration = configuration.path
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures", return_value=set())
with pytest.raises(MissingArchitectureError):
Handler.architectures_extract(args)
def test_architectures_extract_exception(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must raise exception on missing architectures
"""
args.command = "config"
mocker.patch.object(Handler, "ALLOW_AUTO_ARCHITECTURE_RUN", False)
with pytest.raises(MissingArchitectureError):
Handler.architectures_extract(args)
def test_architectures_extract_specified(args: argparse.Namespace) -> None:
"""
must return architecture list if it has been specified
"""
architectures = args.architecture = ["i686", "x86_64"]
assert Handler.architectures_extract(args) == sorted(set(architectures))
from ahriman.models.repository_id import RepositoryId
def test_call(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
@ -62,20 +21,21 @@ def test_call(args: argparse.Namespace, configuration: Configuration, mocker: Mo
args.report = False
mocker.patch("ahriman.application.handlers.Handler.run")
configuration_mock = mocker.patch("ahriman.core.configuration.Configuration.from_path", return_value=configuration)
log_handler_mock = mocker.patch("ahriman.core.log.Log.handler", return_value=args.log_handler)
log_load_mock = mocker.patch("ahriman.core.log.Log.load")
log_handler_mock = mocker.patch("ahriman.core.log.log_loader.LogLoader.handler", return_value=args.log_handler)
log_load_mock = mocker.patch("ahriman.core.log.log_loader.LogLoader.load")
enter_mock = mocker.patch("ahriman.application.lock.Lock.__enter__")
exit_mock = mocker.patch("ahriman.application.lock.Lock.__exit__")
assert Handler.call(args, "x86_64")
configuration_mock.assert_called_once_with(args.configuration, "x86_64")
_, repository_id = configuration.check_loaded()
assert Handler.call(args, repository_id)
configuration_mock.assert_called_once_with(args.configuration, repository_id)
log_handler_mock.assert_called_once_with(args.log_handler)
log_load_mock.assert_called_once_with(configuration, args.log_handler, quiet=args.quiet, report=args.report)
enter_mock.assert_called_once_with()
exit_mock.assert_called_once_with(None, None, None)
def test_call_exception(args: argparse.Namespace, mocker: MockerFixture) -> None:
def test_call_exception(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
"""
must process exception
"""
@ -84,11 +44,12 @@ def test_call_exception(args: argparse.Namespace, mocker: MockerFixture) -> None
mocker.patch("ahriman.core.configuration.Configuration.from_path", side_effect=Exception())
logging_mock = mocker.patch("logging.Logger.exception")
assert not Handler.call(args, "x86_64")
_, repository_id = configuration.check_loaded()
assert not Handler.call(args, repository_id)
logging_mock.assert_called_once_with(pytest.helpers.anyvar(str, strict=True))
def test_call_exit_code(args: argparse.Namespace, mocker: MockerFixture) -> None:
def test_call_exit_code(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
"""
must process exitcode exception
"""
@ -97,7 +58,8 @@ def test_call_exit_code(args: argparse.Namespace, mocker: MockerFixture) -> None
mocker.patch("ahriman.core.configuration.Configuration.from_path", side_effect=ExitCode())
logging_mock = mocker.patch("logging.Logger.exception")
assert not Handler.call(args, "x86_64")
_, repository_id = configuration.check_loaded()
assert not Handler.call(args, repository_id)
logging_mock.assert_not_called()
@ -105,33 +67,39 @@ def test_execute(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must run execution in multiple processes
"""
args.architecture = ["i686", "x86_64"]
ids = [
RepositoryId("i686", "aur-clone"),
RepositoryId("x86_64", "aur-clone"),
]
mocker.patch("ahriman.application.handlers.Handler.repositories_extract", return_value=ids)
starmap_mock = mocker.patch("multiprocessing.pool.Pool.starmap")
Handler.execute(args)
starmap_mock.assert_called_once_with(Handler.call, [(args, architecture) for architecture in args.architecture])
starmap_mock.assert_called_once_with(Handler.call, [(args, repository_id) for repository_id in ids])
def test_execute_multiple_not_supported(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must raise an exception if multiple architectures are not supported by the handler
"""
args.architecture = ["i686", "x86_64"]
args.command = "web"
mocker.patch("ahriman.application.handlers.Handler.repositories_extract", return_value=[
RepositoryId("i686", "aur-clone"),
RepositoryId("x86_64", "aur-clone"),
])
mocker.patch.object(Handler, "ALLOW_MULTI_ARCHITECTURE_RUN", False)
with pytest.raises(MultipleArchitecturesError):
Handler.execute(args)
def test_execute_single(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
def test_execute_single(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must run execution in current process if only one architecture supplied
"""
args.architecture = ["x86_64"]
args.configuration = Path("")
args.quiet = False
mocker.patch("ahriman.core.configuration.Configuration.from_path", return_value=configuration)
mocker.patch("ahriman.application.handlers.Handler.repositories_extract", return_value=[
RepositoryId("x86_64", "aur-clone"),
])
starmap_mock = mocker.patch("multiprocessing.pool.Pool.starmap")
Handler.execute(args)
@ -142,8 +110,117 @@ def test_run(args: argparse.Namespace, configuration: Configuration) -> None:
"""
must raise NotImplemented for missing method
"""
_, repository_id = configuration.check_loaded()
with pytest.raises(NotImplementedError):
Handler.run(args, "x86_64", configuration, report=True)
Handler.run(args, repository_id, configuration, report=True)
def test_repositories_extract(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
"""
must generate list of available repositories based on flags
"""
args.architecture = ["arch"]
args.configuration = configuration.path
args.repository = ["repo"]
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories")
assert Handler.repositories_extract(args) == [RepositoryId("arch", "repo")]
known_architectures_mock.assert_not_called()
known_repositories_mock.assert_not_called()
def test_repositories_extract_repository(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must generate list of available repositories based on flags and tree
"""
args.architecture = ["arch"]
args.configuration = configuration.path
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories",
return_value={"repo"})
assert Handler.repositories_extract(args) == [RepositoryId("arch", "repo")]
known_architectures_mock.assert_not_called()
known_repositories_mock.assert_called_once_with(configuration.repository_paths.root)
def test_repositories_extract_repository_legacy(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must generate list of available repositories based on flags and tree
"""
args.architecture = ["arch"]
args.configuration = configuration.path
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories",
return_value=set())
assert Handler.repositories_extract(args) == [RepositoryId("arch", "aur-clone")]
known_architectures_mock.assert_not_called()
known_repositories_mock.assert_called_once_with(configuration.repository_paths.root)
def test_repositories_extract_architecture(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must read repository name from config
"""
args.configuration = configuration.path
args.repository = ["repo"]
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures",
return_value={"arch"})
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories")
assert Handler.repositories_extract(args) == [RepositoryId("arch", "repo")]
known_architectures_mock.assert_called_once_with(configuration.repository_paths.root, "repo")
known_repositories_mock.assert_not_called()
def test_repositories_extract_empty(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must raise exception if no available architectures found
"""
args.command = "config"
args.configuration = configuration.path
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures", return_value=set())
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories", return_value=set())
with pytest.raises(MissingArchitectureError):
Handler.repositories_extract(args)
def test_repositories_extract_systemd(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must extract repository list for systemd units
"""
args.configuration = configuration.path
args.repository_id = "i686/some/repo/name"
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories")
assert Handler.repositories_extract(args) == [RepositoryId("i686", "some-repo-name")]
known_architectures_mock.assert_not_called()
known_repositories_mock.assert_not_called()
def test_repositories_extract_systemd_legacy(args: argparse.Namespace, configuration: Configuration,
mocker: MockerFixture) -> None:
"""
must extract repository list for systemd units in legacy format
"""
args.configuration = configuration.path
args.repository_id = "i686"
known_architectures_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_architectures")
known_repositories_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.known_repositories",
return_value=set())
assert Handler.repositories_extract(args) == [RepositoryId("i686", "aur-clone")]
known_architectures_mock.assert_not_called()
known_repositories_mock.assert_called_once_with(configuration.repository_paths.root)
def test_check_if_empty() -> None: