arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2026-04-08 03:13:37 +00:00
Code Issues Projects Releases Wiki Activity

feat: allow to run daemon mode with split packages check (#120)

Browse Source
This commit is contained in:
Evgenii Alekseev
2023-12-27 03:05:44 +02:00
committed by GitHub
parent f6cdd806b2
commit b4fa10781b
12 changed files with 335 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
tests/ahriman/application/handlers/test_handler_daemon.py
View File

@@ -1,10 +1,11 @@
import argparse
from pytest_mock import MockerFixture
from unittest.mock import call as MockCall
from ahriman.application.handlers import Daemon
from ahriman.core.configuration import Configuration
from ahriman.core.repository import Repository
from ahriman.models.package import Package
def _default_args(args: argparse.Namespace) -> argparse.Namespace:
@@ -18,35 +19,60 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
argparse.Namespace: generated arguments for these test cases
"""
args.interval = 60 * 60 * 12
args.aur = True
args.local = True
args.manual = True
args.vcs = True
args.partitions = True
args.refresh = 0
return args
def test_run(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
def test_run(args: argparse.Namespace, configuration: Configuration, package_ahriman: Package, repository: Repository,
mocker: MockerFixture) -> None:
"""
must run command
"""
args = _default_args(args)
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
run_mock = mocker.patch("ahriman.application.handlers.Update.run")
wait_mock = mocker.patch("threading.Event.wait", side_effect=[False, True])
iter_mock = mocker.patch("ahriman.application.application.updates_iterator.UpdatesIterator.__iter__",
return_value=iter([[package_ahriman.base]]))
_, repository_id = configuration.check_loaded()
Daemon.run(args, repository_id, configuration, report=True)
args.package = [package_ahriman.base]
run_mock.assert_called_once_with(args, repository_id, configuration, report=True)
iter_mock.assert_called_once_with()
def test_run_no_partitions(args: argparse.Namespace, configuration: Configuration, repository: Repository,
mocker: MockerFixture) -> None:
"""
must run command without partitioning
"""
args = _default_args(args)
args.partitions = False
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
run_mock = mocker.patch("ahriman.application.handlers.Update.run")
iter_mock = mocker.patch("ahriman.application.application.updates_iterator.UpdatesIterator.__iter__",
return_value=iter([[]]))
_, repository_id = configuration.check_loaded()
Daemon.run(args, repository_id, configuration, report=True)
run_mock.assert_called_once_with(args, repository_id, configuration, report=True)
wait_mock.assert_has_calls([MockCall(args.interval), MockCall(args.interval)])
iter_mock.assert_called_once_with()
def test_run_keyboard_interrupt(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
def test_run_no_updates(args: argparse.Namespace, configuration: Configuration, package_ahriman: Package,
repository: Repository, mocker: MockerFixture) -> None:
"""
must handle KeyboardInterrupt exception
must skip empty update list
"""
args = _default_args(args)
mocker.patch("ahriman.application.handlers.Update.run", side_effect=KeyboardInterrupt)
wait_mock = mocker.patch("threading.Event.wait", return_value=False)
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
run_mock = mocker.patch("ahriman.application.handlers.Update.run")
iter_mock = mocker.patch("ahriman.application.application.updates_iterator.UpdatesIterator.__iter__",
return_value=iter([[package_ahriman.base], None]))
_, repository_id = configuration.check_loaded()
Daemon.run(args, repository_id, configuration, report=True)
wait_mock.assert_called_once_with(args.interval)
args.package = [package_ahriman.base]
run_mock.assert_called_once_with(args, repository_id, configuration, report=True)
iter_mock.assert_called_once_with()