mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-07-29 13:49:57 +00:00
* add models tests (#1) also replace single quote to double one to confort PEP docstring + move _check_output to class properties to make it available for mocking * alpm tests implementation * try to replace os with pathlib * update tests for pathlib * fix includes glob and trim version from dependencies * build_tools package tests * repository component tests * add sign tests * complete status tests * handle exceptions in actual_version calls * complete core tests * move configuration to root conftest * application tests * complete application tests * change copyright to more generic one * base web tests * complete web tests * complete testkit also add argument parsers test
This commit is contained in:
10
tests/ahriman/core/alpm/test_pacman.py
Normal file
10
tests/ahriman/core/alpm/test_pacman.py
Normal file
@ -0,0 +1,10 @@
|
||||
from ahriman.core.alpm.pacman import Pacman
|
||||
|
||||
|
||||
def test_all_packages(pacman: Pacman) -> None:
|
||||
"""
|
||||
package list must not be empty
|
||||
"""
|
||||
packages = pacman.all_packages()
|
||||
assert packages
|
||||
assert "pacman" in packages
|
47
tests/ahriman/core/alpm/test_repo.py
Normal file
47
tests/ahriman/core/alpm/test_repo.py
Normal file
@ -0,0 +1,47 @@
|
||||
import pytest
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.alpm.repo import Repo
|
||||
|
||||
|
||||
def test_repo_path(repo: Repo) -> None:
|
||||
"""
|
||||
name must be something like archive name
|
||||
"""
|
||||
assert repo.repo_path.name.endswith("db.tar.gz")
|
||||
|
||||
|
||||
def test_repo_add(repo: Repo, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call repo-add on package addition
|
||||
"""
|
||||
check_output_mock = mocker.patch("ahriman.core.alpm.repo.Repo._check_output")
|
||||
|
||||
repo.add(Path("path"))
|
||||
check_output_mock.assert_called_once()
|
||||
assert check_output_mock.call_args[0][0] == "repo-add"
|
||||
|
||||
|
||||
def test_repo_remove(repo: Repo, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call repo-remove on package addition
|
||||
"""
|
||||
mocker.patch("pathlib.Path.glob", return_value=[])
|
||||
check_output_mock = mocker.patch("ahriman.core.alpm.repo.Repo._check_output")
|
||||
|
||||
repo.remove("package", Path("package.pkg.tar.xz"))
|
||||
check_output_mock.assert_called_once()
|
||||
assert check_output_mock.call_args[0][0] == "repo-remove"
|
||||
|
||||
|
||||
def test_repo_remove_fail_no_file(repo: Repo, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must fail on missing file
|
||||
"""
|
||||
mocker.patch("pathlib.Path.glob", return_value=[Path("package.pkg.tar.xz")])
|
||||
mocker.patch("pathlib.Path.unlink", side_effect=FileNotFoundError())
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
repo.remove("package", Path("package.pkg.tar.xz"))
|
64
tests/ahriman/core/build_tools/test_task.py
Normal file
64
tests/ahriman/core/build_tools/test_task.py
Normal file
@ -0,0 +1,64 @@
|
||||
import pytest
|
||||
import shutil
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.build_tools.task import Task
|
||||
|
||||
|
||||
def test_fetch_existing(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must fetch new package via clone command
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.task.Task._check_output")
|
||||
|
||||
local = Path("local")
|
||||
Task.fetch(local, "remote", "master")
|
||||
check_output_mock.assert_has_calls([
|
||||
mock.call("git", "fetch", "origin", "master",
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
cwd=local, logger=pytest.helpers.anyvar(int)),
|
||||
mock.call("git", "checkout", "--force", "master",
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
cwd=local, logger=pytest.helpers.anyvar(int)),
|
||||
mock.call("git", "reset", "--hard", "origin/master",
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
cwd=local, logger=pytest.helpers.anyvar(int))
|
||||
])
|
||||
|
||||
|
||||
def test_fetch_new(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must fetch new package via clone command
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=False)
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.task.Task._check_output")
|
||||
|
||||
local = Path("local")
|
||||
Task.fetch(local, "remote", "master")
|
||||
check_output_mock.assert_has_calls([
|
||||
mock.call("git", "clone", "remote", str(local),
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
logger=pytest.helpers.anyvar(int)),
|
||||
mock.call("git", "checkout", "--force", "master",
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
cwd=local, logger=pytest.helpers.anyvar(int)),
|
||||
mock.call("git", "reset", "--hard", "origin/master",
|
||||
exception=pytest.helpers.anyvar(int),
|
||||
cwd=local, logger=pytest.helpers.anyvar(int))
|
||||
])
|
||||
|
||||
|
||||
def test_init_with_cache(task_ahriman: Task, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must copy tree instead of fetch
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.fetch")
|
||||
copytree_mock = mocker.patch("shutil.copytree")
|
||||
|
||||
task_ahriman.init(None)
|
||||
copytree_mock.assert_called_once()
|
34
tests/ahriman/core/conftest.py
Normal file
34
tests/ahriman/core/conftest.py
Normal file
@ -0,0 +1,34 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.alpm.pacman import Pacman
|
||||
from ahriman.core.alpm.repo import Repo
|
||||
from ahriman.core.build_tools.task import Task
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.tree import Leaf
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def leaf_ahriman(package_ahriman: Package) -> Leaf:
|
||||
return Leaf(package_ahriman, set())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def leaf_python_schedule(package_python_schedule: Package) -> Leaf:
|
||||
return Leaf(package_python_schedule, set())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pacman(configuration: Configuration) -> Pacman:
|
||||
return Pacman(configuration)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(configuration: Configuration, repository_paths: RepositoryPaths) -> Repo:
|
||||
return Repo(configuration.get("repository", "name"), repository_paths, [])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def task_ahriman(package_ahriman: Package, configuration: Configuration, repository_paths: RepositoryPaths) -> Task:
|
||||
return Task(package_ahriman, "x86_64", configuration, repository_paths)
|
0
tests/ahriman/core/report/test_html.py
Normal file
0
tests/ahriman/core/report/test_html.py
Normal file
0
tests/ahriman/core/report/test_report.py
Normal file
0
tests/ahriman/core/report/test_report.py
Normal file
49
tests/ahriman/core/repository/conftest.py
Normal file
49
tests/ahriman/core/repository/conftest.py
Normal file
@ -0,0 +1,49 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
from ahriman.core.repository.executor import Executor
|
||||
from ahriman.core.repository.properties import Properties
|
||||
from ahriman.core.repository.repository import Repository
|
||||
from ahriman.core.repository.update_handler import UpdateHandler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cleaner(configuration: Configuration, mocker: MockerFixture) -> Cleaner:
|
||||
mocker.patch("pathlib.Path.mkdir")
|
||||
return Cleaner("x86_64", configuration)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def executor(configuration: Configuration, mocker: MockerFixture) -> Executor:
|
||||
mocker.patch("pathlib.Path.mkdir")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_build")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_cache")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_chroot")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_manual")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_packages")
|
||||
return Executor("x86_64", configuration)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repository(configuration: Configuration, mocker: MockerFixture) -> Repository:
|
||||
mocker.patch("pathlib.Path.mkdir")
|
||||
return Repository("x86_64", configuration)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def properties(configuration: Configuration) -> Properties:
|
||||
return Properties("x86_64", configuration)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def update_handler(configuration: Configuration, mocker: MockerFixture) -> UpdateHandler:
|
||||
mocker.patch("pathlib.Path.mkdir")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_build")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_cache")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_chroot")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_manual")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_packages")
|
||||
return UpdateHandler("x86_64", configuration)
|
68
tests/ahriman/core/repository/test_cleaner.py
Normal file
68
tests/ahriman/core/repository/test_cleaner.py
Normal file
@ -0,0 +1,68 @@
|
||||
import shutil
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
|
||||
|
||||
def _mock_clear(mocker: MockerFixture) -> None:
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[Path("a"), Path("b"), Path("c")])
|
||||
mocker.patch("shutil.rmtree")
|
||||
|
||||
|
||||
def _mock_clear_check() -> None:
|
||||
shutil.rmtree.assert_has_calls([
|
||||
mock.call(Path("a")),
|
||||
mock.call(Path("b")),
|
||||
mock.call(Path("c"))
|
||||
])
|
||||
|
||||
|
||||
def test_clear_build(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove directories with sources
|
||||
"""
|
||||
_mock_clear(mocker)
|
||||
cleaner.clear_build()
|
||||
_mock_clear_check()
|
||||
|
||||
|
||||
def test_clear_cache(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove every cached sources
|
||||
"""
|
||||
_mock_clear(mocker)
|
||||
cleaner.clear_cache()
|
||||
_mock_clear_check()
|
||||
|
||||
|
||||
def test_clear_chroot(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clear chroot
|
||||
"""
|
||||
_mock_clear(mocker)
|
||||
cleaner.clear_chroot()
|
||||
_mock_clear_check()
|
||||
|
||||
|
||||
def test_clear_manual(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clear directory with manual packages
|
||||
"""
|
||||
_mock_clear(mocker)
|
||||
cleaner.clear_manual()
|
||||
_mock_clear_check()
|
||||
|
||||
|
||||
def test_clear_packages(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must delete built packages
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.packages_built",
|
||||
return_value=[Path("a"), Path("b"), Path("c")])
|
||||
mocker.patch("pathlib.Path.unlink")
|
||||
|
||||
cleaner.clear_packages()
|
||||
Path.unlink.assert_has_calls([mock.call(), mock.call(), mock.call()])
|
188
tests/ahriman/core/repository/test_executor.py
Normal file
188
tests/ahriman/core/repository/test_executor.py
Normal file
@ -0,0 +1,188 @@
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.repository.executor import Executor
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_process_build(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run build process
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages_built", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.build", return_value=[Path(package_ahriman.base)])
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.init")
|
||||
move_mock = mocker.patch("shutil.move")
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_building")
|
||||
|
||||
# must return list of built packages
|
||||
assert executor.process_build([package_ahriman]) == [package_ahriman]
|
||||
# must move files (once)
|
||||
move_mock.assert_called_once()
|
||||
# must update status
|
||||
status_client_mock.assert_called_once()
|
||||
# must clear directory
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
Cleaner.clear_build.assert_called_once()
|
||||
|
||||
|
||||
def test_process_build_failure(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run correct process failed builds
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages_built")
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.build", return_value=[Path(package_ahriman.base)])
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.init")
|
||||
mocker.patch("shutil.move", side_effect=Exception())
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_failed")
|
||||
|
||||
executor.process_build([package_ahriman])
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_remove_base(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run remove process for whole base
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_ahriman])
|
||||
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
|
||||
|
||||
executor.process_remove([package_ahriman.base])
|
||||
# must remove via alpm wrapper
|
||||
repo_remove_mock.assert_called_once()
|
||||
# must update status
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_remove_base_multiple(executor: Executor, package_python_schedule: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run remove process for whole base with multiple packages
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_python_schedule])
|
||||
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
|
||||
|
||||
executor.process_remove([package_python_schedule.base])
|
||||
# must remove via alpm wrapper
|
||||
repo_remove_mock.assert_has_calls([
|
||||
mock.call(package, Path(props.filename))
|
||||
for package, props in package_python_schedule.packages.items()
|
||||
], any_order=True)
|
||||
# must update status
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_remove_base_single(executor: Executor, package_python_schedule: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run remove process for single package in base
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_python_schedule])
|
||||
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
|
||||
|
||||
executor.process_remove(["python2-schedule"])
|
||||
# must remove via alpm wrapper
|
||||
repo_remove_mock.assert_called_once()
|
||||
# must not update status
|
||||
status_client_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_process_remove_nothing(executor: Executor, package_ahriman: Package, package_python_schedule: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not remove anything if it was not requested
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_ahriman])
|
||||
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
|
||||
|
||||
executor.process_remove([package_python_schedule.base])
|
||||
repo_remove_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_process_report_auto(executor: Executor, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process report in auto mode if no targets supplied
|
||||
"""
|
||||
config_getlist_mock = mocker.patch("ahriman.core.configuration.Configuration.getlist")
|
||||
|
||||
executor.process_report(None)
|
||||
config_getlist_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_sync_auto(executor: Executor, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process sync in auto mode if no targets supplied
|
||||
"""
|
||||
config_getlist_mock = mocker.patch("ahriman.core.configuration.Configuration.getlist")
|
||||
|
||||
executor.process_sync(None)
|
||||
config_getlist_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_update(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run update process
|
||||
"""
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
move_mock = mocker.patch("shutil.move")
|
||||
repo_add_mock = mocker.patch("ahriman.core.alpm.repo.Repo.add")
|
||||
sign_package_mock = mocker.patch("ahriman.core.sign.gpg.GPG.sign_package", side_effect=lambda fn, _: [fn])
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_success")
|
||||
|
||||
# must return complete
|
||||
assert executor.process_update([Path(package.filename) for package in package_ahriman.packages.values()])
|
||||
# must move files (once)
|
||||
move_mock.assert_called_once()
|
||||
# must sign package
|
||||
sign_package_mock.assert_called_once()
|
||||
# must add package
|
||||
repo_add_mock.assert_called_once()
|
||||
# must update status
|
||||
status_client_mock.assert_called_once()
|
||||
# must clear directory
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
Cleaner.clear_packages.assert_called_once()
|
||||
|
||||
|
||||
def test_process_update_group(executor: Executor, package_python_schedule: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must group single packages under one base
|
||||
"""
|
||||
mocker.patch("shutil.move")
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_python_schedule)
|
||||
repo_add_mock = mocker.patch("ahriman.core.alpm.repo.Repo.add")
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_success")
|
||||
|
||||
executor.process_update([Path(package.filename) for package in package_python_schedule.packages.values()])
|
||||
repo_add_mock.assert_has_calls([
|
||||
mock.call(executor.paths.repository / package.filename)
|
||||
for package in package_python_schedule.packages.values()
|
||||
], any_order=True)
|
||||
status_client_mock.assert_called_with(package_python_schedule)
|
||||
|
||||
|
||||
def test_process_update_failed(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process update for failed package
|
||||
"""
|
||||
mocker.patch("shutil.move", side_effect=Exception())
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_failed")
|
||||
|
||||
executor.process_update([Path(package.filename) for package in package_ahriman.packages.values()])
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_process_update_failed_on_load(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process update even with failed package load
|
||||
"""
|
||||
mocker.patch("shutil.move")
|
||||
mocker.patch("ahriman.models.package.Package.load", side_effect=Exception())
|
||||
|
||||
assert executor.process_update([Path(package.filename) for package in package_ahriman.packages.values()])
|
14
tests/ahriman/core/repository/test_properties.py
Normal file
14
tests/ahriman/core/repository/test_properties.py
Normal file
@ -0,0 +1,14 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.repository.properties import Properties
|
||||
|
||||
|
||||
def test_create_tree_on_load(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create tree on load
|
||||
"""
|
||||
create_tree_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.create_tree")
|
||||
Properties("x86_64", configuration)
|
||||
|
||||
create_tree_mock.assert_called_once()
|
33
tests/ahriman/core/repository/test_repository.py
Normal file
33
tests/ahriman/core/repository/test_repository.py
Normal file
@ -0,0 +1,33 @@
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.repository.repository import Repository
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_packages(package_ahriman: Package, package_python_schedule: Package,
|
||||
repository: Repository, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return all packages grouped by package base
|
||||
"""
|
||||
single_packages = [
|
||||
Package(base=package_python_schedule.base,
|
||||
version=package_python_schedule.version,
|
||||
aur_url=package_python_schedule.aur_url,
|
||||
packages={package: props})
|
||||
for package, props in package_python_schedule.packages.items()
|
||||
] + [package_ahriman]
|
||||
|
||||
mocker.patch("pathlib.Path.iterdir",
|
||||
return_value=[Path("a.pkg.tar.xz"), Path("b.pkg.tar.xz"), Path("c.pkg.tar.xz")])
|
||||
mocker.patch("ahriman.models.package.Package.load", side_effect=single_packages)
|
||||
|
||||
packages = repository.packages()
|
||||
assert len(packages) == 2
|
||||
assert {package.base for package in packages} == {package_ahriman.base, package_python_schedule.base}
|
||||
|
||||
archives = sum([list(package.packages.keys()) for package in packages], start=[])
|
||||
assert len(archives) == 3
|
||||
expected = set(package_ahriman.packages.keys())
|
||||
expected.update(package_python_schedule.packages.keys())
|
||||
assert set(archives) == expected
|
124
tests/ahriman/core/repository/test_update_handler.py
Normal file
124
tests/ahriman/core/repository/test_update_handler.py
Normal file
@ -0,0 +1,124 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.repository.update_handler import UpdateHandler
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_updates_aur(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must provide updates with status updates
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.models.package.Package.is_outdated", return_value=True)
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_pending")
|
||||
|
||||
assert update_handler.updates_aur([], False) == [package_ahriman]
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_aur_failed(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update status via client for failed load
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.models.package.Package.load", side_effect=Exception())
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_failed")
|
||||
|
||||
update_handler.updates_aur([], False)
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_aur_filter(update_handler: UpdateHandler, package_ahriman: Package, package_python_schedule: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must provide updates only for filtered packages
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages",
|
||||
return_value=[package_ahriman, package_python_schedule])
|
||||
mocker.patch("ahriman.models.package.Package.is_outdated", return_value=True)
|
||||
package_load_mock = mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
|
||||
assert update_handler.updates_aur([package_ahriman.base], False) == [package_ahriman]
|
||||
package_load_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_aur_ignore(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip ignore packages
|
||||
"""
|
||||
mocker.patch("ahriman.core.configuration.Configuration.getlist", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
package_load_mock = mocker.patch("ahriman.models.package.Package.load")
|
||||
|
||||
update_handler.updates_aur([], False)
|
||||
package_load_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_updates_aur_ignore_vcs(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip VCS packages check if requested
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.models.package.Package.is_vcs", return_value=True)
|
||||
package_is_outdated_mock = mocker.patch("ahriman.models.package.Package.is_outdated")
|
||||
|
||||
update_handler.updates_aur([], True)
|
||||
package_is_outdated_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_updates_manual_clear(update_handler: UpdateHandler, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
requesting manual updates must clear packages directory
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages")
|
||||
|
||||
update_handler.updates_manual()
|
||||
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
Cleaner.clear_manual.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_manual_status_known(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create record for known package via reporter
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_pending")
|
||||
|
||||
update_handler.updates_manual()
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_manual_status_unknown(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create record for unknown package via reporter
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[])
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
status_client_mock = mocker.patch("ahriman.core.status.client.Client.set_unknown")
|
||||
|
||||
update_handler.updates_manual()
|
||||
status_client_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_updates_manual_with_failures(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process through the packages with failure
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[])
|
||||
mocker.patch("ahriman.models.package.Package.load", side_effect=Exception())
|
||||
|
||||
assert update_handler.updates_manual() == []
|
9
tests/ahriman/core/sign/conftest.py
Normal file
9
tests/ahriman/core/sign/conftest.py
Normal file
@ -0,0 +1,9 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.sign.gpg import GPG
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gpg(configuration: Configuration) -> GPG:
|
||||
return GPG("x86_64", configuration)
|
61
tests/ahriman/core/sign/gpg.py
Normal file
61
tests/ahriman/core/sign/gpg.py
Normal file
@ -0,0 +1,61 @@
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.sign.gpg import GPG
|
||||
from ahriman.models.sign_settings import SignSettings
|
||||
|
||||
|
||||
def test_repository_sign_args(gpg: GPG) -> None:
|
||||
"""
|
||||
must generate correct sign args
|
||||
"""
|
||||
gpg.target = {SignSettings.SignRepository}
|
||||
assert gpg.repository_sign_args
|
||||
|
||||
|
||||
def test_sign_package(gpg: GPG, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must sign package
|
||||
"""
|
||||
result = [Path("a"), Path("a.sig")]
|
||||
process_mock = mocker.patch("ahriman.core.sign.gpg.process", return_value=result)
|
||||
|
||||
for target in ({SignSettings.SignPackages}, {SignSettings.SignPackages, SignSettings.SignRepository}):
|
||||
gpg.target = target
|
||||
assert gpg.sign_package(Path("a"), "a") == result
|
||||
process_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_sign_package_skip(gpg: GPG, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not sign package if it is not set
|
||||
"""
|
||||
process_mock = mocker.patch("ahriman.core.sign.gpg.process")
|
||||
|
||||
for target in ({}, {SignSettings.SignRepository}):
|
||||
gpg.target = target
|
||||
process_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_sign_repository(gpg: GPG, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must sign repository
|
||||
"""
|
||||
result = [Path("a"), Path("a.sig")]
|
||||
process_mock = mocker.patch("ahriman.core.sign.gpg.process", return_value=result)
|
||||
|
||||
for target in ({SignSettings.SignRepository}, {SignSettings.SignPackages, SignSettings.SignRepository}):
|
||||
gpg.target = target
|
||||
assert gpg.sign_repository(Path("a")) == result
|
||||
process_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_sign_repository_skip(gpg: GPG, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not sign repository if it is not set
|
||||
"""
|
||||
process_mock = mocker.patch("ahriman.core.sign.gpg.process")
|
||||
|
||||
for target in ({}, {SignSettings.SignPackages}):
|
||||
gpg.target = target
|
||||
process_mock.assert_not_called()
|
30
tests/ahriman/core/status/conftest.py
Normal file
30
tests/ahriman/core/status/conftest.py
Normal file
@ -0,0 +1,30 @@
|
||||
import pytest
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
from ahriman.core.status.client import Client
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
# helpers
|
||||
@pytest.helpers.register
|
||||
def get_package_status(package: Package) -> Dict[str, Any]:
|
||||
return {"status": BuildStatusEnum.Unknown.value, "package": package.view()}
|
||||
|
||||
|
||||
@pytest.helpers.register
|
||||
def get_package_status_extended(package: Package) -> Dict[str, Any]:
|
||||
return {"status": BuildStatus().view(), "package": package.view()}
|
||||
|
||||
|
||||
# fixtures
|
||||
@pytest.fixture
|
||||
def client() -> Client:
|
||||
return Client()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def web_client() -> WebClient:
|
||||
return WebClient("localhost", 8080)
|
116
tests/ahriman/core/status/test_client.py
Normal file
116
tests/ahriman/core/status/test_client.py
Normal file
@ -0,0 +1,116 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.status.client import Client
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_add(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must process package addition without errors
|
||||
"""
|
||||
client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_get(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return empty package list
|
||||
"""
|
||||
assert client.get(package_ahriman.base) == []
|
||||
assert client.get(None) == []
|
||||
|
||||
|
||||
def test_get_self(client: Client) -> None:
|
||||
"""
|
||||
must return unknown status for service
|
||||
"""
|
||||
assert client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_remove(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must process remove without errors
|
||||
"""
|
||||
client.remove(package_ahriman.base)
|
||||
|
||||
|
||||
def test_update(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must update package status without errors
|
||||
"""
|
||||
client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_update_self(client: Client) -> None:
|
||||
"""
|
||||
must update self status without errors
|
||||
"""
|
||||
client.update_self(BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_set_building(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must set building status to the package
|
||||
"""
|
||||
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
|
||||
client.set_building(package_ahriman.base)
|
||||
|
||||
update_mock.assert_called_with(package_ahriman.base, BuildStatusEnum.Building)
|
||||
|
||||
|
||||
def test_set_failed(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must set failed status to the package
|
||||
"""
|
||||
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
|
||||
client.set_failed(package_ahriman.base)
|
||||
|
||||
update_mock.assert_called_with(package_ahriman.base, BuildStatusEnum.Failed)
|
||||
|
||||
|
||||
def test_set_pending(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must set building status to the package
|
||||
"""
|
||||
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
|
||||
client.set_pending(package_ahriman.base)
|
||||
|
||||
update_mock.assert_called_with(package_ahriman.base, BuildStatusEnum.Pending)
|
||||
|
||||
|
||||
def test_set_success(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must set success status to the package
|
||||
"""
|
||||
add_mock = mocker.patch("ahriman.core.status.client.Client.add")
|
||||
client.set_success(package_ahriman)
|
||||
|
||||
add_mock.assert_called_with(package_ahriman, BuildStatusEnum.Success)
|
||||
|
||||
|
||||
def test_set_unknown(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must add new package with unknown status
|
||||
"""
|
||||
add_mock = mocker.patch("ahriman.core.status.client.Client.add")
|
||||
client.set_unknown(package_ahriman)
|
||||
|
||||
add_mock.assert_called_with(package_ahriman, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_load_dummy_client(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load dummy client if no settings set
|
||||
"""
|
||||
assert isinstance(Client.load("x86_64", configuration), Client)
|
||||
|
||||
|
||||
def test_load_full_client(configuration: Configuration) -> None:
|
||||
"""
|
||||
must load full client if no settings set
|
||||
"""
|
||||
configuration.set("web", "host", "localhost")
|
||||
configuration.set("web", "port", "8080")
|
||||
assert isinstance(Client.load("x86_64", configuration), WebClient)
|
219
tests/ahriman/core/status/test_watcher.py
Normal file
219
tests/ahriman/core/status/test_watcher.py
Normal file
@ -0,0 +1,219 @@
|
||||
import pytest
|
||||
import tempfile
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest.mock import PropertyMock
|
||||
|
||||
from ahriman.core.exceptions import UnknownPackage
|
||||
from ahriman.core.status.watcher import Watcher
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_cache_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load state from cache
|
||||
"""
|
||||
response = {"packages": [pytest.helpers.get_package_status_extended(package_ahriman)]}
|
||||
|
||||
mocker.patch("pathlib.Path.is_file", return_value=True)
|
||||
mocker.patch("pathlib.Path.open")
|
||||
mocker.patch("json.load", return_value=response)
|
||||
|
||||
watcher.known = {package_ahriman.base: (None, None)}
|
||||
watcher._cache_load()
|
||||
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_cache_load_json_error(watcher: Watcher, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on json errors
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_file", return_value=True)
|
||||
mocker.patch("pathlib.Path.open")
|
||||
mocker.patch("json.load", side_effect=Exception())
|
||||
|
||||
watcher._cache_load()
|
||||
assert not watcher.known
|
||||
|
||||
|
||||
def test_cache_load_no_file(watcher: Watcher, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on missing file
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_file", return_value=False)
|
||||
|
||||
watcher._cache_load()
|
||||
assert not watcher.known
|
||||
|
||||
|
||||
def test_cache_load_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not load unknown package
|
||||
"""
|
||||
response = {"packages": [pytest.helpers.get_package_status_extended(package_ahriman)]}
|
||||
|
||||
mocker.patch("pathlib.Path.is_file", return_value=True)
|
||||
mocker.patch("pathlib.Path.open")
|
||||
mocker.patch("json.load", return_value=response)
|
||||
|
||||
watcher._cache_load()
|
||||
assert not watcher.known
|
||||
|
||||
|
||||
def test_cache_save(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must save state to cache
|
||||
"""
|
||||
mocker.patch("pathlib.Path.open")
|
||||
json_mock = mocker.patch("json.dump")
|
||||
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
watcher._cache_save()
|
||||
json_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_cache_save_failed(watcher: Watcher, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on dumping packages
|
||||
"""
|
||||
mocker.patch("pathlib.Path.open")
|
||||
mocker.patch("json.dump", side_effect=Exception())
|
||||
|
||||
watcher._cache_save()
|
||||
|
||||
|
||||
def test_cache_save_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must save state to cache which can be loaded later
|
||||
"""
|
||||
dump_file = Path(tempfile.mktemp())
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.cache_path",
|
||||
new_callable=PropertyMock, return_value=dump_file)
|
||||
known_current = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.known = known_current
|
||||
watcher._cache_save()
|
||||
|
||||
watcher.known = {package_ahriman.base: (None, None)}
|
||||
watcher._cache_load()
|
||||
assert watcher.known == known_current
|
||||
|
||||
dump_file.unlink()
|
||||
|
||||
|
||||
def test_get(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return package status
|
||||
"""
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
package, status = watcher.get(package_ahriman.base)
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must fail on unknown package
|
||||
"""
|
||||
with pytest.raises(UnknownPackage):
|
||||
watcher.get(package_ahriman.base)
|
||||
|
||||
|
||||
def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly load packages
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_load")
|
||||
|
||||
watcher.load()
|
||||
cache_mock.assert_called_once()
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly load packages with known statuses
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher._cache_load")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus(BuildStatusEnum.Success))}
|
||||
|
||||
watcher.load()
|
||||
_, status = watcher.known[package_ahriman.base]
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
def test_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove package base
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
assert not watcher.known
|
||||
cache_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on unknown base removal
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
cache_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
|
||||
cache_mock.assert_called_once()
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must update package status only for known package
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Success, None)
|
||||
cache_mock.assert_called_once()
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
def test_update_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must fail on unknown package status update only
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
|
||||
with pytest.raises(UnknownPackage):
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, None)
|
||||
cache_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_update_self(watcher: Watcher) -> None:
|
||||
"""
|
||||
must update service status
|
||||
"""
|
||||
watcher.update_self(BuildStatusEnum.Success)
|
||||
assert watcher.status.status == BuildStatusEnum.Success
|
163
tests/ahriman/core/status/test_web_client.py
Normal file
163
tests/ahriman/core/status/test_web_client.py
Normal file
@ -0,0 +1,163 @@
|
||||
import json
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from requests import Response
|
||||
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_ahriman_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate service status url correctly
|
||||
"""
|
||||
assert web_client._ahriman_url().startswith(f"http://{web_client.host}:{web_client.port}")
|
||||
assert web_client._ahriman_url().endswith("/api/v1/ahriman")
|
||||
|
||||
|
||||
def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must generate package status correctly
|
||||
"""
|
||||
assert web_client._package_url(package_ahriman.base).startswith(f"http://{web_client.host}:{web_client.port}")
|
||||
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
|
||||
|
||||
|
||||
def test_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process package addition
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
payload = pytest.helpers.get_package_status(package_ahriman)
|
||||
|
||||
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json=payload)
|
||||
|
||||
|
||||
def test_add_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during addition
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_get_all(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return all packages status
|
||||
"""
|
||||
response = [pytest.helpers.get_package_status_extended(package_ahriman)]
|
||||
response_obj = Response()
|
||||
response_obj._content = json.dumps(response).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get(None)
|
||||
requests_mock.assert_called_once()
|
||||
assert len(result) == len(response)
|
||||
assert (package_ahriman, BuildStatusEnum.Unknown) in [(package, status.status) for package, status in result]
|
||||
|
||||
|
||||
def test_get_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=Exception())
|
||||
assert web_client.get(None) == []
|
||||
|
||||
|
||||
def test_get_single(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return single package status
|
||||
"""
|
||||
response = [pytest.helpers.get_package_status_extended(package_ahriman)]
|
||||
response_obj = Response()
|
||||
response_obj._content = json.dumps(response).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get(package_ahriman.base)
|
||||
requests_mock.assert_called_once()
|
||||
assert len(result) == len(response)
|
||||
assert (package_ahriman, BuildStatusEnum.Unknown) in [(package, status.status) for package, status in result]
|
||||
|
||||
|
||||
def test_get_self(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return service status
|
||||
"""
|
||||
response_obj = Response()
|
||||
response_obj._content = json.dumps(BuildStatus().view()).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get_self()
|
||||
requests_mock.assert_called_once()
|
||||
assert result.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_get_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during service status getting
|
||||
"""
|
||||
mocker.patch("requests.get", side_effect=Exception())
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process package removal
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.delete")
|
||||
|
||||
web_client.remove(package_ahriman.base)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True))
|
||||
|
||||
|
||||
def test_remove_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during removal
|
||||
"""
|
||||
mocker.patch("requests.delete", side_effect=Exception())
|
||||
web_client.remove(package_ahriman.base)
|
||||
|
||||
|
||||
def test_update(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process package update
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
|
||||
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json={"status": BuildStatusEnum.Unknown.value})
|
||||
|
||||
|
||||
def test_update_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
|
||||
|
||||
|
||||
def test_update_self(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process service update
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.post")
|
||||
|
||||
web_client.update_self(BuildStatusEnum.Unknown)
|
||||
requests_mock.assert_called_with(pytest.helpers.anyvar(str, True), json={"status": BuildStatusEnum.Unknown.value})
|
||||
|
||||
|
||||
def test_update_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during service update
|
||||
"""
|
||||
mocker.patch("requests.post", side_effect=Exception())
|
||||
web_client.update_self(BuildStatusEnum.Unknown)
|
129
tests/ahriman/core/test_configuration.py
Normal file
129
tests/ahriman/core/test_configuration.py
Normal file
@ -0,0 +1,129 @@
|
||||
from pathlib import Path
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
|
||||
|
||||
def test_from_path(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load configuration
|
||||
"""
|
||||
read_mock = mocker.patch("configparser.RawConfigParser.read")
|
||||
load_includes_mock = mocker.patch("ahriman.core.configuration.Configuration.load_includes")
|
||||
load_logging_mock = mocker.patch("ahriman.core.configuration.Configuration.load_logging")
|
||||
path = Path("path")
|
||||
|
||||
config = Configuration.from_path(path, True)
|
||||
assert config.path == path
|
||||
read_mock.assert_called_with(path)
|
||||
load_includes_mock.assert_called_once()
|
||||
load_logging_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_absolute_path_for_absolute(configuration: Configuration) -> None:
|
||||
"""
|
||||
must not change path for absolute path in settings
|
||||
"""
|
||||
path = Path("/a/b/c")
|
||||
configuration.set("build", "path", str(path))
|
||||
assert configuration.getpath("build", "path") == path
|
||||
|
||||
|
||||
def test_absolute_path_for_relative(configuration: Configuration) -> None:
|
||||
"""
|
||||
must prepend root path to relative path
|
||||
"""
|
||||
path = Path("a")
|
||||
configuration.set("build", "path", str(path))
|
||||
result = configuration.getpath("build", "path")
|
||||
assert result.is_absolute()
|
||||
assert result.parent == configuration.path.parent
|
||||
assert result.name == path.name
|
||||
|
||||
|
||||
def test_dump(configuration: Configuration) -> None:
|
||||
"""
|
||||
dump must not be empty
|
||||
"""
|
||||
assert configuration.dump("x86_64")
|
||||
|
||||
|
||||
def test_dump_architecture_specific(configuration: Configuration) -> None:
|
||||
"""
|
||||
dump must contain architecture specific settings
|
||||
"""
|
||||
configuration.add_section("build_x86_64")
|
||||
configuration.set("build_x86_64", "archbuild_flags", "")
|
||||
|
||||
dump = configuration.dump("x86_64")
|
||||
assert dump
|
||||
assert "build" not in dump
|
||||
assert "build_x86_64" in dump
|
||||
|
||||
|
||||
def test_getlist(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return list of string correctly
|
||||
"""
|
||||
configuration.set("build", "test_list", "a b c")
|
||||
assert configuration.getlist("build", "test_list") == ["a", "b", "c"]
|
||||
|
||||
|
||||
def test_getlist_empty(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return list of string correctly for non-existing option
|
||||
"""
|
||||
assert configuration.getlist("build", "test_list") == []
|
||||
configuration.set("build", "test_list", "")
|
||||
assert configuration.getlist("build", "test_list") == []
|
||||
|
||||
|
||||
def test_getlist_single(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return list of strings for single string
|
||||
"""
|
||||
configuration.set("build", "test_list", "a")
|
||||
assert configuration.getlist("build", "test_list") == ["a"]
|
||||
|
||||
|
||||
def test_get_section_name(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return architecture specific group
|
||||
"""
|
||||
configuration.add_section("build_x86_64")
|
||||
configuration.set("build_x86_64", "archbuild_flags", "")
|
||||
assert configuration.get_section_name("build", "x86_64") == "build_x86_64"
|
||||
|
||||
|
||||
def test_get_section_name_missing(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return default group if architecture depending group does not exist
|
||||
"""
|
||||
assert configuration.get_section_name("prefix", "suffix") == "prefix"
|
||||
assert configuration.get_section_name("build", "x86_64") == "build"
|
||||
|
||||
|
||||
def test_load_includes_missing(configuration: Configuration) -> None:
|
||||
"""
|
||||
must not fail if not include directory found
|
||||
"""
|
||||
configuration.set("settings", "include", "path")
|
||||
configuration.load_includes()
|
||||
|
||||
|
||||
def test_load_logging_fallback(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must fallback to stderr without errors
|
||||
"""
|
||||
mocker.patch("logging.config.fileConfig", side_effect=PermissionError())
|
||||
configuration.load_logging(True)
|
||||
|
||||
|
||||
def test_load_logging_stderr(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must use stderr if flag set
|
||||
"""
|
||||
logging_mock = mocker.patch("logging.config.fileConfig")
|
||||
configuration.load_logging(False)
|
||||
logging_mock.assert_not_called()
|
78
tests/ahriman/core/test_tree.py
Normal file
78
tests/ahriman/core/test_tree.py
Normal file
@ -0,0 +1,78 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.tree import Leaf, Tree
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_leaf_is_root_empty(leaf_ahriman: Leaf) -> None:
|
||||
"""
|
||||
must be root for empty packages list
|
||||
"""
|
||||
assert leaf_ahriman.is_root([])
|
||||
|
||||
|
||||
def test_leaf_is_root_false(leaf_ahriman: Leaf, leaf_python_schedule: Leaf) -> None:
|
||||
"""
|
||||
must be root for empty dependencies list or if does not depend on packages
|
||||
"""
|
||||
assert leaf_ahriman.is_root([leaf_python_schedule])
|
||||
leaf_ahriman.dependencies = {"ahriman-dependency"}
|
||||
assert leaf_ahriman.is_root([leaf_python_schedule])
|
||||
|
||||
|
||||
def test_leaf_is_root_true(leaf_ahriman: Leaf, leaf_python_schedule: Leaf) -> None:
|
||||
"""
|
||||
must not be root if depends on packages
|
||||
"""
|
||||
leaf_ahriman.dependencies = {"python-schedule"}
|
||||
assert not leaf_ahriman.is_root([leaf_python_schedule])
|
||||
|
||||
leaf_ahriman.dependencies = {"python2-schedule"}
|
||||
assert not leaf_ahriman.is_root([leaf_python_schedule])
|
||||
|
||||
leaf_ahriman.dependencies = set(leaf_python_schedule.package.packages.keys())
|
||||
assert not leaf_ahriman.is_root([leaf_python_schedule])
|
||||
|
||||
|
||||
def test_leaf_load(package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load with dependencies
|
||||
"""
|
||||
tempdir_mock = mocker.patch("tempfile.mkdtemp")
|
||||
fetch_mock = mocker.patch("ahriman.core.build_tools.task.Task.fetch")
|
||||
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies", return_value={"ahriman-dependency"})
|
||||
rmtree_mock = mocker.patch("shutil.rmtree")
|
||||
|
||||
leaf = Leaf.load(package_ahriman)
|
||||
assert leaf.package == package_ahriman
|
||||
assert leaf.dependencies == {"ahriman-dependency"}
|
||||
tempdir_mock.assert_called_once()
|
||||
fetch_mock.assert_called_once()
|
||||
dependencies_mock.assert_called_once()
|
||||
rmtree_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_tree_levels(leaf_ahriman: Leaf, leaf_python_schedule: Leaf, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must generate correct levels in the simples case
|
||||
"""
|
||||
leaf_ahriman.dependencies = set(leaf_python_schedule.package.packages.keys())
|
||||
|
||||
tree = Tree([leaf_ahriman, leaf_python_schedule])
|
||||
assert len(tree.levels()) == 2
|
||||
first, second = tree.levels()
|
||||
assert first == [leaf_python_schedule.package]
|
||||
assert second == [leaf_ahriman.package]
|
||||
|
||||
|
||||
def test_tree_load(package_ahriman: Package, package_python_schedule: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must package list
|
||||
"""
|
||||
mocker.patch("tempfile.mkdtemp")
|
||||
mocker.patch("ahriman.core.build_tools.task.Task.fetch")
|
||||
mocker.patch("ahriman.models.package.Package.dependencies")
|
||||
mocker.patch("shutil.rmtree")
|
||||
|
||||
tree = Tree.load([package_ahriman, package_python_schedule])
|
||||
assert len(tree.leaves) == 2
|
131
tests/ahriman/core/test_util.py
Normal file
131
tests/ahriman/core/test_util.py
Normal file
@ -0,0 +1,131 @@
|
||||
import logging
|
||||
import pytest
|
||||
import subprocess
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.util import check_output, package_like, pretty_datetime, pretty_size
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_check_output(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run command and log result
|
||||
"""
|
||||
logger_mock = mocker.patch("logging.Logger.debug")
|
||||
|
||||
assert check_output("echo", "hello", exception=None) == "hello"
|
||||
logger_mock.assert_not_called()
|
||||
|
||||
assert check_output("echo", "hello", exception=None, logger=logging.getLogger("")) == "hello"
|
||||
logger_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_check_output_failure(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process exception correctly
|
||||
"""
|
||||
logger_mock = mocker.patch("logging.Logger.debug")
|
||||
mocker.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "echo"))
|
||||
|
||||
with pytest.raises(subprocess.CalledProcessError):
|
||||
check_output("echo", "hello", exception=None)
|
||||
logger_mock.assert_not_called()
|
||||
|
||||
with pytest.raises(subprocess.CalledProcessError):
|
||||
check_output("echo", "hello", exception=None, logger=logging.getLogger(""))
|
||||
logger_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_check_output_failure_log(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process exception correctly and log it
|
||||
"""
|
||||
logger_mock = mocker.patch("logging.Logger.debug")
|
||||
mocker.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "echo", output=b"result"))
|
||||
|
||||
with pytest.raises(subprocess.CalledProcessError):
|
||||
check_output("echo", "hello", exception=None, logger=logging.getLogger(""))
|
||||
logger_mock.assert_called_once()
|
||||
|
||||
|
||||
def test_package_like(package_ahriman: Package) -> None:
|
||||
"""
|
||||
package_like must return true for archives
|
||||
"""
|
||||
assert package_like(package_ahriman.packages[package_ahriman.base].filepath)
|
||||
|
||||
|
||||
def test_package_like_sig(package_ahriman: Package) -> None:
|
||||
"""
|
||||
package_like must return false for signature files
|
||||
"""
|
||||
package_file = package_ahriman.packages[package_ahriman.base].filepath
|
||||
sig_file = package_file.parent / f"{package_file.name}.sig"
|
||||
assert not package_like(sig_file)
|
||||
|
||||
|
||||
def test_pretty_datetime() -> None:
|
||||
"""
|
||||
must generate string from timestamp value
|
||||
"""
|
||||
assert pretty_datetime(0) == "1970-01-01 00:00:00"
|
||||
|
||||
|
||||
def test_pretty_datetime_empty() -> None:
|
||||
"""
|
||||
must generate empty string from None timestamp
|
||||
"""
|
||||
assert pretty_datetime(None) == ""
|
||||
|
||||
|
||||
def test_pretty_size_bytes() -> None:
|
||||
"""
|
||||
must generate bytes string for bytes value
|
||||
"""
|
||||
value, abbrev = pretty_size(42).split()
|
||||
assert value == "42.0"
|
||||
assert abbrev == "B"
|
||||
|
||||
|
||||
def test_pretty_size_kbytes() -> None:
|
||||
"""
|
||||
must generate kibibytes string for kibibytes value
|
||||
"""
|
||||
value, abbrev = pretty_size(42 * 1024).split()
|
||||
assert value == "42.0"
|
||||
assert abbrev == "KiB"
|
||||
|
||||
|
||||
def test_pretty_size_mbytes() -> None:
|
||||
"""
|
||||
must generate mebibytes string for mebibytes value
|
||||
"""
|
||||
value, abbrev = pretty_size(42 * 1024 * 1024).split()
|
||||
assert value == "42.0"
|
||||
assert abbrev == "MiB"
|
||||
|
||||
|
||||
def test_pretty_size_gbytes() -> None:
|
||||
"""
|
||||
must generate gibibytes string for gibibytes value
|
||||
"""
|
||||
value, abbrev = pretty_size(42 * 1024 * 1024 * 1024).split()
|
||||
assert value == "42.0"
|
||||
assert abbrev == "GiB"
|
||||
|
||||
|
||||
def test_pretty_size_pbytes() -> None:
|
||||
"""
|
||||
must generate pebibytes string for pebibytes value
|
||||
"""
|
||||
value, abbrev = pretty_size(42 * 1024 * 1024 * 1024 * 1024).split()
|
||||
assert value == "43008.0"
|
||||
assert abbrev == "GiB"
|
||||
|
||||
|
||||
def test_pretty_size_empty() -> None:
|
||||
"""
|
||||
must generate empty string for None value
|
||||
"""
|
||||
assert pretty_size(None) == ""
|
0
tests/ahriman/core/upload/test_rsync.py
Normal file
0
tests/ahriman/core/upload/test_rsync.py
Normal file
0
tests/ahriman/core/upload/test_s3.py
Normal file
0
tests/ahriman/core/upload/test_s3.py
Normal file
0
tests/ahriman/core/upload/test_uploader.py
Normal file
0
tests/ahriman/core/upload/test_uploader.py
Normal file
Reference in New Issue
Block a user