mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-06-28 14:51:43 +00:00
port part of settings to database (#54)
This commit is contained in:
@ -6,39 +6,46 @@ from ahriman.application.application.packages import Packages
|
||||
from ahriman.application.application.properties import Properties
|
||||
from ahriman.application.application.repository import Repository
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_packages(configuration: Configuration, mocker: MockerFixture) -> Packages:
|
||||
def application_packages(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Packages:
|
||||
"""
|
||||
fixture for application with package functions
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
return Packages("x86_64", configuration, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_properties(configuration: Configuration, mocker: MockerFixture) -> Properties:
|
||||
def application_properties(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Properties:
|
||||
"""
|
||||
fixture for application with properties only
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
return Properties("x86_64", configuration, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_repository(configuration: Configuration, mocker: MockerFixture) -> Repository:
|
||||
def application_repository(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Repository:
|
||||
"""
|
||||
fixture for application with repository functions
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
return Repository("x86_64", configuration, no_report=True, unsafe=False)
|
||||
|
@ -2,7 +2,6 @@ import pytest
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.application.application.packages import Packages
|
||||
@ -43,16 +42,17 @@ def test_add_aur(application_packages: Packages, package_ahriman: Package, mocke
|
||||
must add package from AUR
|
||||
"""
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
insert_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_insert")
|
||||
load_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.load")
|
||||
dependencies_mock = mocker.patch("ahriman.application.application.packages.Packages._process_dependencies")
|
||||
|
||||
application_packages._add_aur(package_ahriman.base, set(), False)
|
||||
insert_mock.assert_called_once_with(package_ahriman)
|
||||
load_mock.assert_called_once_with(
|
||||
application_packages.repository.paths.manual_for(package_ahriman.base),
|
||||
pytest.helpers.anyvar(int),
|
||||
package_ahriman.git_url,
|
||||
application_packages.repository.paths.patches_for(package_ahriman.base))
|
||||
dependencies_mock.assert_called_once_with(
|
||||
application_packages.repository.paths.manual_for(package_ahriman.base), set(), False)
|
||||
pytest.helpers.anyvar(int))
|
||||
dependencies_mock.assert_called_once_with(pytest.helpers.anyvar(int), set(), False)
|
||||
|
||||
|
||||
def test_add_directory(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
@ -75,18 +75,16 @@ def test_add_local(application_packages: Packages, package_ahriman: Package, moc
|
||||
"""
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
init_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.init")
|
||||
insert_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_insert")
|
||||
copytree_mock = mocker.patch("shutil.copytree")
|
||||
dependencies_mock = mocker.patch("ahriman.application.application.packages.Packages._process_dependencies")
|
||||
|
||||
application_packages._add_local(package_ahriman.base, set(), False)
|
||||
copytree_mock.assert_called_once_with(
|
||||
Path(package_ahriman.base), application_packages.repository.paths.cache_for(package_ahriman.base))
|
||||
init_mock.assert_called_once_with(application_packages.repository.paths.cache_for(package_ahriman.base))
|
||||
copytree_mock.assert_has_calls([
|
||||
mock.call(Path(package_ahriman.base), application_packages.repository.paths.cache_for(package_ahriman.base)),
|
||||
mock.call(application_packages.repository.paths.cache_for(package_ahriman.base),
|
||||
application_packages.repository.paths.manual_for(package_ahriman.base)),
|
||||
])
|
||||
dependencies_mock.assert_called_once_with(
|
||||
application_packages.repository.paths.manual_for(package_ahriman.base), set(), False)
|
||||
insert_mock.assert_called_once_with(package_ahriman)
|
||||
dependencies_mock.assert_called_once_with(pytest.helpers.anyvar(int), set(), False)
|
||||
|
||||
|
||||
def test_add_remote(application_packages: Packages, package_description_ahriman: PackageDescription,
|
||||
|
@ -17,21 +17,12 @@ def test_finalize(application_repository: Repository) -> None:
|
||||
application_repository._finalize([])
|
||||
|
||||
|
||||
def test_clean_build(application_repository: Repository, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clean build directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_build")
|
||||
application_repository.clean(True, False, False, False, False, False)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_clean_cache(application_repository: Repository, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clean cache directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_cache")
|
||||
application_repository.clean(False, True, False, False, False, False)
|
||||
application_repository.clean(True, False, False, False)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
@ -40,7 +31,7 @@ def test_clean_chroot(application_repository: Repository, mocker: MockerFixture)
|
||||
must clean chroot directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_chroot")
|
||||
application_repository.clean(False, False, True, False, False, False)
|
||||
application_repository.clean(False, True, False, False)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
@ -48,8 +39,8 @@ def test_clean_manual(application_repository: Repository, mocker: MockerFixture)
|
||||
"""
|
||||
must clean manual directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_manual")
|
||||
application_repository.clean(False, False, False, True, False, False)
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_queue")
|
||||
application_repository.clean(False, False, True, False)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
@ -58,16 +49,7 @@ def test_clean_packages(application_repository: Repository, mocker: MockerFixtur
|
||||
must clean packages directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_packages")
|
||||
application_repository.clean(False, False, False, False, True, False)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_clean_patches(application_repository: Repository, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clean packages directory
|
||||
"""
|
||||
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_patches")
|
||||
application_repository.clean(False, False, False, False, False, True)
|
||||
application_repository.clean(False, False, False, True)
|
||||
clear_mock.assert_called_once_with()
|
||||
|
||||
|
||||
|
@ -7,17 +7,20 @@ from ahriman.application.ahriman import _parser
|
||||
from ahriman.application.application import Application
|
||||
from ahriman.application.lock import Lock
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application(configuration: Configuration, mocker: MockerFixture) -> Application:
|
||||
def application(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Application:
|
||||
"""
|
||||
fixture for application
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
return Application("x86_64", configuration, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
|
@ -6,7 +6,7 @@ from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.application.handlers import Handler
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.exceptions import MissingArchitecture, MultipleArchitectures
|
||||
from ahriman.core.exceptions import ExitCode, MissingArchitecture, MultipleArchitectures
|
||||
|
||||
|
||||
def test_architectures_extract(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
@ -71,8 +71,26 @@ def test_call_exception(args: argparse.Namespace, mocker: MockerFixture) -> None
|
||||
"""
|
||||
must process exception
|
||||
"""
|
||||
mocker.patch("ahriman.application.lock.Lock.__enter__", side_effect=Exception())
|
||||
args.configuration = Path("")
|
||||
args.quiet = False
|
||||
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")
|
||||
logging_mock.assert_called_once_with(pytest.helpers.anyvar(str, strict=True))
|
||||
|
||||
|
||||
def test_call_exit_code(args: argparse.Namespace, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process exitcode exception
|
||||
"""
|
||||
args.configuration = Path("")
|
||||
args.quiet = False
|
||||
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")
|
||||
logging_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_execute(args: argparse.Namespace, mocker: MockerFixture) -> None:
|
||||
@ -98,11 +116,14 @@ def test_execute_multiple_not_supported(args: argparse.Namespace, mocker: Mocker
|
||||
Handler.execute(args)
|
||||
|
||||
|
||||
def test_execute_single(args: argparse.Namespace, mocker: MockerFixture) -> None:
|
||||
def test_execute_single(args: argparse.Namespace, configuration: Configuration, 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)
|
||||
starmap_mock = mocker.patch("multiprocessing.pool.Pool.starmap")
|
||||
|
||||
Handler.execute(args)
|
||||
|
@ -12,12 +12,10 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
|
||||
:param args: command line arguments fixture
|
||||
:return: generated arguments for these test cases
|
||||
"""
|
||||
args.build = False
|
||||
args.cache = False
|
||||
args.chroot = False
|
||||
args.manual = False
|
||||
args.packages = False
|
||||
args.patches = False
|
||||
return args
|
||||
|
||||
|
||||
@ -30,4 +28,4 @@ def test_run(args: argparse.Namespace, configuration: Configuration, mocker: Moc
|
||||
application_mock = mocker.patch("ahriman.application.application.Application.clean")
|
||||
|
||||
Clean.run(args, "x86_64", configuration, True, False)
|
||||
application_mock.assert_called_once_with(False, False, False, False, False, False)
|
||||
application_mock.assert_called_once_with(False, False, False, False)
|
||||
|
@ -67,24 +67,23 @@ def test_patch_set_list(application: Application, mocker: MockerFixture) -> None
|
||||
must list available patches for the command
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
glob_mock = mocker.patch("pathlib.Path.glob", return_value=[Path("local")])
|
||||
print_mock = mocker.patch("ahriman.application.handlers.patch.Patch._print")
|
||||
get_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.patches_list", return_value={"ahriman": "patch"})
|
||||
print_mock = mocker.patch("ahriman.core.formatters.printer.Printer.print")
|
||||
|
||||
Patch.patch_set_list(application, "ahriman")
|
||||
glob_mock.assert_called_once_with("*.patch")
|
||||
print_mock.assert_called()
|
||||
get_mock.assert_called_once_with("ahriman")
|
||||
print_mock.assert_called_once_with(verbose=True)
|
||||
|
||||
|
||||
def test_patch_set_list_no_dir(application: Application, mocker: MockerFixture) -> None:
|
||||
def test_patch_set_list_no_patches(application: Application, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail if no patches directory found
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=False)
|
||||
glob_mock = mocker.patch("pathlib.Path.glob")
|
||||
print_mock = mocker.patch("ahriman.application.handlers.patch.Patch._print")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.patches_get", return_value=None)
|
||||
print_mock = mocker.patch("ahriman.core.formatters.printer.Printer.print")
|
||||
|
||||
Patch.patch_set_list(application, "ahriman")
|
||||
glob_mock.assert_not_called()
|
||||
print_mock.assert_not_called()
|
||||
|
||||
|
||||
@ -94,21 +93,17 @@ def test_patch_set_create(application: Application, package_ahriman: Package, mo
|
||||
"""
|
||||
mocker.patch("pathlib.Path.mkdir")
|
||||
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
|
||||
remove_mock = mocker.patch("ahriman.application.handlers.patch.Patch.patch_set_remove")
|
||||
create_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.patch_create")
|
||||
patch_dir = application.repository.paths.patches_for(package_ahriman.base)
|
||||
mocker.patch("ahriman.core.build_tools.sources.Sources.patch_create", return_value="patch")
|
||||
create_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.patches_insert")
|
||||
|
||||
Patch.patch_set_create(application, Path("path"), ["*.patch"])
|
||||
remove_mock.assert_called_once_with(application, package_ahriman.base)
|
||||
create_mock.assert_called_once_with(Path("path"), patch_dir / "00-main.patch", "*.patch")
|
||||
Patch.patch_set_create(application, "path", ["*.patch"])
|
||||
create_mock.assert_called_once_with(package_ahriman.base, "patch")
|
||||
|
||||
|
||||
def test_patch_set_remove(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove patch set for the package
|
||||
"""
|
||||
remove_mock = mocker.patch("shutil.rmtree")
|
||||
patch_dir = application.repository.paths.patches_for(package_ahriman.base)
|
||||
|
||||
remove_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.patches_remove")
|
||||
Patch.patch_set_remove(application, package_ahriman.base)
|
||||
remove_mock.assert_called_once_with(patch_dir, ignore_errors=True)
|
||||
remove_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
@ -6,13 +6,25 @@ from pytest_mock import MockerFixture
|
||||
from ahriman.application.ahriman import _parser
|
||||
from ahriman.application.handlers import UnsafeCommands
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.exceptions import ExitCode
|
||||
|
||||
|
||||
def _default_args(args: argparse.Namespace) -> argparse.Namespace:
|
||||
"""
|
||||
default arguments for these test cases
|
||||
:param args: command line arguments fixture
|
||||
:return: generated arguments for these test cases
|
||||
"""
|
||||
args.parser = _parser
|
||||
args.command = None
|
||||
return args
|
||||
|
||||
|
||||
def test_run(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run command
|
||||
"""
|
||||
args.parser = _parser
|
||||
args = _default_args(args)
|
||||
commands_mock = mocker.patch("ahriman.application.handlers.UnsafeCommands.get_unsafe_commands",
|
||||
return_value=["command"])
|
||||
print_mock = mocker.patch("ahriman.core.formatters.printer.Printer.print")
|
||||
@ -22,6 +34,36 @@ def test_run(args: argparse.Namespace, configuration: Configuration, mocker: Moc
|
||||
print_mock.assert_called_once_with(verbose=True)
|
||||
|
||||
|
||||
def test_run_check(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run command and check if command is unsafe
|
||||
"""
|
||||
args = _default_args(args)
|
||||
args.command = "clean"
|
||||
commands_mock = mocker.patch("ahriman.application.handlers.UnsafeCommands.get_unsafe_commands",
|
||||
return_value=["command"])
|
||||
check_mock = mocker.patch("ahriman.application.handlers.UnsafeCommands.check_unsafe")
|
||||
|
||||
UnsafeCommands.run(args, "x86_64", configuration, True, False)
|
||||
commands_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
check_mock.assert_called_once_with("clean", ["command"], pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_check_unsafe() -> None:
|
||||
"""
|
||||
must check if command is unsafe
|
||||
"""
|
||||
with pytest.raises(ExitCode):
|
||||
UnsafeCommands.check_unsafe("repo-clean", ["repo-clean"], _parser())
|
||||
|
||||
|
||||
def test_check_unsafe_safe() -> None:
|
||||
"""
|
||||
must check if command is unsafe
|
||||
"""
|
||||
UnsafeCommands.check_unsafe("package-status", ["repo-clean"], _parser())
|
||||
|
||||
|
||||
def test_get_unsafe_commands() -> None:
|
||||
"""
|
||||
must return unsafe commands
|
||||
|
@ -3,10 +3,11 @@ import pytest
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.application.handlers import User
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.exceptions import InitializeException
|
||||
from ahriman.models.action import Action
|
||||
from ahriman.models.user import User as MUser
|
||||
from ahriman.models.user_access import UserAccess
|
||||
@ -21,96 +22,78 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
|
||||
args.username = "user"
|
||||
args.action = Action.Update
|
||||
args.as_service = False
|
||||
args.no_reload = False
|
||||
args.password = "pa55w0rd"
|
||||
args.role = UserAccess.Read
|
||||
args.secure = False
|
||||
return args
|
||||
|
||||
|
||||
def test_run(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_run(args: argparse.Namespace, configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run command
|
||||
"""
|
||||
args = _default_args(args)
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
user = MUser(args.username, args.password, args.role)
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
mocker.patch("ahriman.models.user.User.hash_password", return_value=user)
|
||||
get_auth_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_get")
|
||||
create_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_create")
|
||||
write_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_write")
|
||||
create_user_mock = mocker.patch("ahriman.application.handlers.User.user_create")
|
||||
get_salt_mock = mocker.patch("ahriman.application.handlers.User.get_salt")
|
||||
reload_mock = mocker.patch("ahriman.core.status.client.Client.reload_auth")
|
||||
create_user_mock = mocker.patch("ahriman.application.handlers.User.user_create", return_value=user)
|
||||
get_salt_mock = mocker.patch("ahriman.application.handlers.User.get_salt", return_value="salt")
|
||||
update_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.user_update")
|
||||
|
||||
User.run(args, "x86_64", configuration, True, False)
|
||||
get_auth_configuration_mock.assert_called_once_with(configuration.include)
|
||||
create_configuration_mock.assert_called_once_with(
|
||||
pytest.helpers.anyvar(int), pytest.helpers.anyvar(int), pytest.helpers.anyvar(int), args.as_service)
|
||||
create_configuration_mock.assert_called_once_with(pytest.helpers.anyvar(int), pytest.helpers.anyvar(int),
|
||||
pytest.helpers.anyvar(int), args.as_service, args.secure)
|
||||
create_user_mock.assert_called_once_with(args)
|
||||
get_salt_mock.assert_called_once_with(configuration)
|
||||
write_configuration_mock.assert_called_once_with(pytest.helpers.anyvar(int), args.secure)
|
||||
reload_mock.assert_called_once_with()
|
||||
update_mock.assert_called_once_with(user)
|
||||
|
||||
|
||||
def test_run_remove(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_run_list(args: argparse.Namespace, configuration: Configuration, database: SQLite, user: User,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must list avaiable users
|
||||
"""
|
||||
args = _default_args(args)
|
||||
args.action = Action.List
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
get_auth_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_get")
|
||||
list_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.user_list", return_value=[user])
|
||||
|
||||
User.run(args, "x86_64", configuration, True, False)
|
||||
get_auth_configuration_mock.assert_called_once_with(configuration.include)
|
||||
list_mock.assert_called_once_with("user", UserAccess.Read)
|
||||
|
||||
|
||||
def test_run_remove(args: argparse.Namespace, configuration: Configuration, database: SQLite,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove user if remove flag supplied
|
||||
"""
|
||||
args = _default_args(args)
|
||||
args.action = Action.Remove
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
get_auth_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_get")
|
||||
create_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_create")
|
||||
write_configuration_mock = mocker.patch("ahriman.application.handlers.User.configuration_write")
|
||||
reload_mock = mocker.patch("ahriman.core.status.client.Client.reload_auth")
|
||||
remove_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.user_remove")
|
||||
|
||||
User.run(args, "x86_64", configuration, True, False)
|
||||
get_auth_configuration_mock.assert_called_once_with(configuration.include)
|
||||
create_configuration_mock.assert_not_called()
|
||||
write_configuration_mock.assert_called_once_with(pytest.helpers.anyvar(int), args.secure)
|
||||
reload_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_run_no_reload(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run command with no reload
|
||||
"""
|
||||
args = _default_args(args)
|
||||
args.no_reload = True
|
||||
mocker.patch("ahriman.application.handlers.User.configuration_get")
|
||||
mocker.patch("ahriman.application.handlers.User.configuration_create")
|
||||
mocker.patch("ahriman.application.handlers.User.configuration_write")
|
||||
reload_mock = mocker.patch("ahriman.core.status.client.Client.reload_auth")
|
||||
|
||||
User.run(args, "x86_64", configuration, True, False)
|
||||
reload_mock.assert_not_called()
|
||||
remove_mock.assert_called_once_with(args.username)
|
||||
|
||||
|
||||
def test_configuration_create(configuration: Configuration, user: MUser, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly create configuration file
|
||||
"""
|
||||
section = Configuration.section_name("auth", user.access.value)
|
||||
mocker.patch("pathlib.Path.open")
|
||||
set_mock = mocker.patch("ahriman.core.configuration.Configuration.set_option")
|
||||
write_mock = mocker.patch("ahriman.application.handlers.User.configuration_write")
|
||||
|
||||
User.configuration_create(configuration, user, "salt", False)
|
||||
set_mock.assert_has_calls([
|
||||
mock.call("auth", "salt", pytest.helpers.anyvar(int)),
|
||||
mock.call(section, user.username, pytest.helpers.anyvar(int))
|
||||
])
|
||||
|
||||
|
||||
def test_configuration_create_user_exists(configuration: Configuration, user: MUser, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly update configuration file if user already exists
|
||||
"""
|
||||
section = Configuration.section_name("auth", user.access.value)
|
||||
configuration.set_option(section, user.username, "")
|
||||
mocker.patch("pathlib.Path.open")
|
||||
|
||||
User.configuration_create(configuration, user, "salt", False)
|
||||
generated = MUser.from_option(user.username, configuration.get(section, user.username))
|
||||
assert generated.check_credentials(user.password, configuration.get("auth", "salt"))
|
||||
User.configuration_create(configuration, user, "salt", False, False)
|
||||
set_mock.assert_called_once_with("auth", "salt", pytest.helpers.anyvar(int))
|
||||
write_mock.assert_called_once_with(configuration, False)
|
||||
|
||||
|
||||
def test_configuration_create_with_plain_password(
|
||||
@ -120,12 +103,11 @@ def test_configuration_create_with_plain_password(
|
||||
"""
|
||||
must set plain text password and user for the service
|
||||
"""
|
||||
section = Configuration.section_name("auth", user.access.value)
|
||||
mocker.patch("pathlib.Path.open")
|
||||
|
||||
User.configuration_create(configuration, user, "salt", True)
|
||||
User.configuration_create(configuration, user, "salt", True, False)
|
||||
|
||||
generated = MUser.from_option(user.username, configuration.get(section, user.username))
|
||||
generated = MUser.from_option(user.username, user.password).hash_password("salt")
|
||||
service = MUser.from_option(configuration.get("web", "username"), configuration.get("web", "password"))
|
||||
assert generated.username == service.username
|
||||
assert generated.check_credentials(service.password, configuration.get("auth", "salt"))
|
||||
@ -174,12 +156,9 @@ def test_configuration_write_not_loaded(configuration: Configuration, mocker: Mo
|
||||
"""
|
||||
configuration.path = None
|
||||
mocker.patch("pathlib.Path.open")
|
||||
write_mock = mocker.patch("ahriman.core.configuration.Configuration.write")
|
||||
chmod_mock = mocker.patch("pathlib.Path.chmod")
|
||||
|
||||
User.configuration_write(configuration, secure=True)
|
||||
write_mock.assert_not_called()
|
||||
chmod_mock.assert_not_called()
|
||||
with pytest.raises(InitializeException):
|
||||
User.configuration_write(configuration, secure=True)
|
||||
|
||||
|
||||
def test_get_salt_read(configuration: Configuration) -> None:
|
||||
@ -200,31 +179,6 @@ def test_get_salt_generate(configuration: Configuration) -> None:
|
||||
assert len(salt) == 16
|
||||
|
||||
|
||||
def test_user_clear(configuration: Configuration, user: MUser) -> None:
|
||||
"""
|
||||
must clear user from configuration
|
||||
"""
|
||||
section = Configuration.section_name("auth", user.access.value)
|
||||
configuration.set_option(section, user.username, user.password)
|
||||
|
||||
User.user_clear(configuration, user)
|
||||
assert configuration.get(section, user.username, fallback=None) is None
|
||||
|
||||
|
||||
def test_user_clear_multiple_sections(configuration: Configuration, user: MUser) -> None:
|
||||
"""
|
||||
must clear user from configuration from all sections
|
||||
"""
|
||||
for role in UserAccess:
|
||||
section = Configuration.section_name("auth", role.value)
|
||||
configuration.set_option(section, user.username, user.password)
|
||||
|
||||
User.user_clear(configuration, user)
|
||||
for role in UserAccess:
|
||||
section = Configuration.section_name("auth", role.value)
|
||||
assert configuration.get(section, user.username, fallback=None) is None
|
||||
|
||||
|
||||
def test_user_create(args: argparse.Namespace, user: MUser) -> None:
|
||||
"""
|
||||
must create user
|
||||
|
@ -438,6 +438,38 @@ def test_subparsers_user_add_option_role(parser: argparse.ArgumentParser) -> Non
|
||||
assert isinstance(args.role, UserAccess)
|
||||
|
||||
|
||||
def test_subparsers_user_list(parser: argparse.ArgumentParser) -> None:
|
||||
"""
|
||||
user-list command must imply action, architecture, lock, no-report, password, quiet and unsafe
|
||||
"""
|
||||
args = parser.parse_args(["user-list"])
|
||||
assert args.action == Action.List
|
||||
assert args.architecture == [""]
|
||||
assert args.lock is None
|
||||
assert args.no_report
|
||||
assert args.password is not None
|
||||
assert args.quiet
|
||||
assert args.unsafe
|
||||
|
||||
|
||||
def test_subparsers_user_list_architecture(parser: argparse.ArgumentParser) -> None:
|
||||
"""
|
||||
user-list command must correctly parse architecture list
|
||||
"""
|
||||
args = parser.parse_args(["-a", "x86_64", "user-list"])
|
||||
assert args.architecture == [""]
|
||||
|
||||
|
||||
def test_subparsers_user_list_option_role(parser: argparse.ArgumentParser) -> None:
|
||||
"""
|
||||
user-list command must convert role option to useraccess instance
|
||||
"""
|
||||
args = parser.parse_args(["user-list"])
|
||||
assert isinstance(args.role, UserAccess)
|
||||
args = parser.parse_args(["user-list", "--role", "write"])
|
||||
assert isinstance(args.role, UserAccess)
|
||||
|
||||
|
||||
def test_subparsers_user_remove(parser: argparse.ArgumentParser) -> None:
|
||||
"""
|
||||
user-remove command must imply action, architecture, lock, no-report, password, quiet, role and unsafe
|
||||
|
@ -3,14 +3,16 @@ import pytest
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any, Type, TypeVar
|
||||
from typing import Any, Dict, Type, TypeVar
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.auth.auth import Auth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.spawn import Spawn
|
||||
from ahriman.core.status.watcher import Watcher
|
||||
from ahriman.models.aur_package import AURPackage
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.package_description import PackageDescription
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
@ -48,6 +50,26 @@ def anyvar(cls: Type[T], strict: bool = False) -> T:
|
||||
return AnyVar()
|
||||
|
||||
|
||||
@pytest.helpers.register
|
||||
def get_package_status(package: Package) -> Dict[str, Any]:
|
||||
"""
|
||||
helper to extract package status from package
|
||||
:param package: package object
|
||||
:return: simplified package status map (with only status and view)
|
||||
"""
|
||||
return {"status": BuildStatusEnum.Unknown.value, "package": package.view()}
|
||||
|
||||
|
||||
@pytest.helpers.register
|
||||
def get_package_status_extended(package: Package) -> Dict[str, Any]:
|
||||
"""
|
||||
helper to extract package status from package
|
||||
:param package: package object
|
||||
:return: full package status map (with timestamped build status and view)
|
||||
"""
|
||||
return {"status": BuildStatus().view(), "package": package.view()}
|
||||
|
||||
|
||||
# generic fixtures
|
||||
@pytest.fixture
|
||||
def aur_package_ahriman() -> AURPackage:
|
||||
@ -123,6 +145,18 @@ def configuration(resource_path_root: Path) -> Configuration:
|
||||
return Configuration.from_path(path=path, architecture="x86_64", quiet=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def database(configuration: Configuration) -> SQLite:
|
||||
"""
|
||||
database fixture
|
||||
:param: configuration: configuration fixture
|
||||
:return: database test instance
|
||||
"""
|
||||
database = SQLite.load(configuration)
|
||||
yield database
|
||||
database.path.unlink()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def package_ahriman(package_description_ahriman: PackageDescription) -> Package:
|
||||
"""
|
||||
@ -267,12 +301,13 @@ def user() -> User:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def watcher(configuration: Configuration, mocker: MockerFixture) -> Watcher:
|
||||
def watcher(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Watcher:
|
||||
"""
|
||||
package status watcher fixture
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: package status watcher test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
return Watcher("x86_64", configuration)
|
||||
return Watcher("x86_64", configuration, database)
|
||||
|
@ -3,24 +3,27 @@ import pytest
|
||||
from ahriman.core.auth.mapping import Mapping
|
||||
from ahriman.core.auth.oauth import OAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mapping(configuration: Configuration) -> Mapping:
|
||||
def mapping(configuration: Configuration, database: SQLite) -> Mapping:
|
||||
"""
|
||||
auth provider fixture
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:return: auth service instance
|
||||
"""
|
||||
return Mapping(configuration)
|
||||
return Mapping(configuration, database)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def oauth(configuration: Configuration) -> OAuth:
|
||||
def oauth(configuration: Configuration, database: SQLite) -> OAuth:
|
||||
"""
|
||||
OAuth provider fixture
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:return: OAuth2 service instance
|
||||
"""
|
||||
configuration.set("web", "address", "https://example.com")
|
||||
return OAuth(configuration)
|
||||
return OAuth(configuration, database)
|
||||
|
@ -1,10 +1,8 @@
|
||||
import pytest
|
||||
|
||||
from ahriman.core.auth.auth import Auth
|
||||
from ahriman.core.auth.mapping import Mapping
|
||||
from ahriman.core.auth.oauth import OAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.exceptions import DuplicateUser
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.user import User
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
@ -17,88 +15,42 @@ def test_auth_control(auth: Auth) -> None:
|
||||
assert "button" in auth.auth_control # I think it should be button
|
||||
|
||||
|
||||
def test_load_dummy(configuration: Configuration) -> None:
|
||||
def test_load_dummy(configuration: Configuration, database: SQLite) -> None:
|
||||
"""
|
||||
must load dummy validator if authorization is not enabled
|
||||
"""
|
||||
configuration.set_option("auth", "target", "disabled")
|
||||
auth = Auth.load(configuration)
|
||||
auth = Auth.load(configuration, database)
|
||||
assert isinstance(auth, Auth)
|
||||
|
||||
|
||||
def test_load_dummy_empty(configuration: Configuration) -> None:
|
||||
def test_load_dummy_empty(configuration: Configuration, database: SQLite) -> None:
|
||||
"""
|
||||
must load dummy validator if no option set
|
||||
"""
|
||||
auth = Auth.load(configuration)
|
||||
auth = Auth.load(configuration, database)
|
||||
assert isinstance(auth, Auth)
|
||||
|
||||
|
||||
def test_load_mapping(configuration: Configuration) -> None:
|
||||
def test_load_mapping(configuration: Configuration, database: SQLite) -> None:
|
||||
"""
|
||||
must load mapping validator if option set
|
||||
"""
|
||||
configuration.set_option("auth", "target", "configuration")
|
||||
auth = Auth.load(configuration)
|
||||
auth = Auth.load(configuration, database)
|
||||
assert isinstance(auth, Mapping)
|
||||
|
||||
|
||||
def test_load_oauth(configuration: Configuration) -> None:
|
||||
def test_load_oauth(configuration: Configuration, database: SQLite) -> None:
|
||||
"""
|
||||
must load OAuth2 validator if option set
|
||||
"""
|
||||
configuration.set_option("auth", "target", "oauth")
|
||||
configuration.set_option("web", "address", "https://example.com")
|
||||
auth = Auth.load(configuration)
|
||||
auth = Auth.load(configuration, database)
|
||||
assert isinstance(auth, OAuth)
|
||||
|
||||
|
||||
def test_get_users(mapping: Auth, configuration: Configuration) -> None:
|
||||
"""
|
||||
must return valid user list
|
||||
"""
|
||||
user_write = User("user_write", "pwd_write", UserAccess.Write)
|
||||
write_section = Configuration.section_name("auth", user_write.access.value)
|
||||
configuration.set_option(write_section, user_write.username, user_write.password)
|
||||
user_read = User("user_read", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user_read.access.value)
|
||||
configuration.set_option(read_section, user_read.username, user_read.password)
|
||||
user_read = User("user_read", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user_read.access.value)
|
||||
configuration.set_option(read_section, user_read.username, user_read.password)
|
||||
|
||||
users = mapping.get_users(configuration)
|
||||
expected = {user_write.username: user_write, user_read.username: user_read}
|
||||
assert users == expected
|
||||
|
||||
|
||||
def test_get_users_normalized(mapping: Auth, configuration: Configuration) -> None:
|
||||
"""
|
||||
must return user list with normalized usernames in keys
|
||||
"""
|
||||
user = User("UsEr", "pwd_read", UserAccess.Read)
|
||||
read_section = Configuration.section_name("auth", user.access.value)
|
||||
configuration.set_option(read_section, user.username, user.password)
|
||||
|
||||
users = mapping.get_users(configuration)
|
||||
expected = user.username.lower()
|
||||
assert expected in users
|
||||
assert users[expected].username == expected
|
||||
|
||||
|
||||
def test_get_users_duplicate(mapping: Auth, configuration: Configuration, user: User) -> None:
|
||||
"""
|
||||
must raise exception on duplicate username
|
||||
"""
|
||||
write_section = Configuration.section_name("auth", UserAccess.Write.value)
|
||||
configuration.set_option(write_section, user.username, user.password)
|
||||
read_section = Configuration.section_name("auth", UserAccess.Read.value)
|
||||
configuration.set_option(read_section, user.username, user.password)
|
||||
|
||||
with pytest.raises(DuplicateUser):
|
||||
mapping.get_users(configuration)
|
||||
|
||||
|
||||
async def test_check_credentials(auth: Auth, user: User) -> None:
|
||||
"""
|
||||
must pass any credentials
|
||||
|
@ -1,15 +1,17 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.auth.mapping import Mapping
|
||||
from ahriman.models.user import User
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
|
||||
async def test_check_credentials(mapping: Mapping, user: User) -> None:
|
||||
async def test_check_credentials(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return true for valid credentials
|
||||
"""
|
||||
current_password = user.password
|
||||
user.password = user.hash_password(mapping.salt)
|
||||
mapping._users[user.username] = user
|
||||
user = user.hash_password(mapping.salt)
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert await mapping.check_credentials(user.username, current_password)
|
||||
# here password is hashed so it is invalid
|
||||
assert not await mapping.check_credentials(user.username, user.password)
|
||||
@ -31,19 +33,19 @@ async def test_check_credentials_unknown(mapping: Mapping, user: User) -> None:
|
||||
assert not await mapping.check_credentials(user.username, user.password)
|
||||
|
||||
|
||||
def test_get_user(mapping: Mapping, user: User) -> None:
|
||||
def test_get_user(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return user from storage by username
|
||||
"""
|
||||
mapping._users[user.username] = user
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert mapping.get_user(user.username) == user
|
||||
|
||||
|
||||
def test_get_user_normalized(mapping: Mapping, user: User) -> None:
|
||||
def test_get_user_normalized(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return user from storage by username case-insensitive
|
||||
"""
|
||||
mapping._users[user.username] = user
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert mapping.get_user(user.username.upper()) == user
|
||||
|
||||
|
||||
@ -54,20 +56,27 @@ def test_get_user_unknown(mapping: Mapping, user: User) -> None:
|
||||
assert mapping.get_user(user.username) is None
|
||||
|
||||
|
||||
async def test_known_username(mapping: Mapping, user: User) -> None:
|
||||
async def test_known_username(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow only known users
|
||||
"""
|
||||
mapping._users[user.username] = user
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert await mapping.known_username(user.username)
|
||||
|
||||
|
||||
async def test_known_username_unknown(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not allow only known users
|
||||
"""
|
||||
assert not await mapping.known_username(None)
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=None)
|
||||
assert not await mapping.known_username(user.password)
|
||||
|
||||
|
||||
async def test_verify_access(mapping: Mapping, user: User) -> None:
|
||||
async def test_verify_access(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must verify user access
|
||||
"""
|
||||
mapping._users[user.username] = user
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert await mapping.verify_access(user.username, user.access, None)
|
||||
assert not await mapping.verify_access(user.username, UserAccess.Write, None)
|
||||
|
@ -22,16 +22,25 @@ def test_add(mocker: MockerFixture) -> None:
|
||||
exception=None, cwd=local, logger=pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_add_skip(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip addition of files to index if no fiels found
|
||||
"""
|
||||
mocker.patch("pathlib.Path.glob", return_value=[])
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.sources.Sources._check_output")
|
||||
|
||||
Sources.add(Path("local"), "pattern1")
|
||||
check_output_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_diff(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must calculate diff
|
||||
"""
|
||||
write_mock = mocker.patch("pathlib.Path.write_text")
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.sources.Sources._check_output")
|
||||
|
||||
local = Path("local")
|
||||
Sources.diff(local, Path("patch"))
|
||||
write_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
assert Sources.diff(local)
|
||||
check_output_mock.assert_called_once_with("git", "diff",
|
||||
exception=None, cwd=local, logger=pytest.helpers.anyvar(int))
|
||||
|
||||
@ -142,51 +151,34 @@ def test_load(mocker: MockerFixture) -> None:
|
||||
fetch_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.fetch")
|
||||
patch_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.patch_apply")
|
||||
|
||||
Sources.load(Path("local"), "remote", Path("patches"))
|
||||
Sources.load(Path("local"), "remote", "patch")
|
||||
fetch_mock.assert_called_once_with(Path("local"), "remote")
|
||||
patch_mock.assert_called_once_with(Path("local"), Path("patches"))
|
||||
patch_mock.assert_called_once_with(Path("local"), "patch")
|
||||
|
||||
|
||||
def test_load_no_patch(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load packages sources correctly without patches
|
||||
"""
|
||||
mocker.patch("ahriman.core.build_tools.sources.Sources.fetch")
|
||||
patch_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.patch_apply")
|
||||
|
||||
Sources.load(Path("local"), "remote", None)
|
||||
patch_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_patch_apply(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must apply patches if any
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
glob_mock = mocker.patch("pathlib.Path.glob", return_value=[Path("01.patch"), Path("02.patch")])
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.sources.Sources._check_output")
|
||||
|
||||
local = Path("local")
|
||||
Sources.patch_apply(local, Path("patches"))
|
||||
glob_mock.assert_called_once_with("*.patch")
|
||||
check_output_mock.assert_has_calls([
|
||||
mock.call("git", "apply", "--ignore-space-change", "--ignore-whitespace", "01.patch",
|
||||
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)),
|
||||
mock.call("git", "apply", "--ignore-space-change", "--ignore-whitespace", "02.patch",
|
||||
exception=None, cwd=local, logger=pytest.helpers.anyvar(int)),
|
||||
])
|
||||
|
||||
|
||||
def test_patch_apply_no_dir(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail if no patches directory exists
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=False)
|
||||
glob_mock = mocker.patch("pathlib.Path.glob")
|
||||
|
||||
Sources.patch_apply(Path("local"), Path("patches"))
|
||||
glob_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_patch_apply_no_patches(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail if no patches exist
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
mocker.patch("pathlib.Path.glob", return_value=[])
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.sources.Sources._check_output")
|
||||
|
||||
Sources.patch_apply(Path("local"), Path("patches"))
|
||||
check_output_mock.assert_not_called()
|
||||
Sources.patch_apply(local, "patches")
|
||||
check_output_mock.assert_called_once_with(
|
||||
"git", "apply", "--ignore-space-change", "--ignore-whitespace",
|
||||
exception=None, cwd=local, input_data="patches", logger=pytest.helpers.anyvar(int)
|
||||
)
|
||||
|
||||
|
||||
def test_patch_create(mocker: MockerFixture) -> None:
|
||||
@ -196,6 +188,15 @@ def test_patch_create(mocker: MockerFixture) -> None:
|
||||
add_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.add")
|
||||
diff_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.diff")
|
||||
|
||||
Sources.patch_create(Path("local"), Path("patch"), "glob")
|
||||
Sources.patch_create(Path("local"), "glob")
|
||||
add_mock.assert_called_once_with(Path("local"), "glob")
|
||||
diff_mock.assert_called_once_with(Path("local"), Path("patch"))
|
||||
diff_mock.assert_called_once_with(Path("local"))
|
||||
|
||||
|
||||
def test_patch_create_with_newline(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
created patch must have new line at the end
|
||||
"""
|
||||
mocker.patch("ahriman.core.build_tools.sources.Sources.add")
|
||||
mocker.patch("ahriman.core.build_tools.sources.Sources.diff", return_value="diff")
|
||||
assert Sources.patch_create(Path("local"), "glob").endswith("\n")
|
||||
|
@ -1,6 +1,8 @@
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.build_tools.task import Task
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
def test_build(task_ahriman: Task, mocker: MockerFixture) -> None:
|
||||
@ -8,11 +10,11 @@ def test_build(task_ahriman: Task, mocker: MockerFixture) -> None:
|
||||
must build package
|
||||
"""
|
||||
check_output_mock = mocker.patch("ahriman.core.build_tools.task.Task._check_output")
|
||||
task_ahriman.build()
|
||||
task_ahriman.build(Path("ahriman"))
|
||||
check_output_mock.assert_called()
|
||||
|
||||
|
||||
def test_init_with_cache(task_ahriman: Task, mocker: MockerFixture) -> None:
|
||||
def test_init_with_cache(task_ahriman: Task, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must copy tree instead of fetch
|
||||
"""
|
||||
@ -20,5 +22,5 @@ def test_init_with_cache(task_ahriman: Task, mocker: MockerFixture) -> None:
|
||||
mocker.patch("ahriman.core.build_tools.sources.Sources.load")
|
||||
copytree_mock = mocker.patch("shutil.copytree")
|
||||
|
||||
task_ahriman.init(None)
|
||||
task_ahriman.init(Path("ahriman"), database)
|
||||
copytree_mock.assert_called_once() # we do not check full command here, sorry
|
||||
|
13
tests/ahriman/core/database/conftest.py
Normal file
13
tests/ahriman/core/database/conftest.py
Normal file
@ -0,0 +1,13 @@
|
||||
import pytest
|
||||
|
||||
from sqlite3 import Connection
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def connection() -> Connection:
|
||||
"""
|
||||
mock object for sqlite3 connection
|
||||
:return: sqlite3 connection test instance
|
||||
"""
|
||||
return MagicMock()
|
33
tests/ahriman/core/database/data/test_data_init.py
Normal file
33
tests/ahriman/core/database/data/test_data_init.py
Normal file
@ -0,0 +1,33 @@
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlite3 import Connection
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.data import migrate_data
|
||||
from ahriman.models.migration_result import MigrationResult
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
def test_migrate_data_initial(connection: Connection, configuration: Configuration,
|
||||
repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must perform initial migration
|
||||
"""
|
||||
packages = mocker.patch("ahriman.core.database.data.migrate_package_statuses")
|
||||
users = mocker.patch("ahriman.core.database.data.migrate_users_data")
|
||||
|
||||
migrate_data(MigrationResult(old_version=0, new_version=900), connection, configuration, repository_paths)
|
||||
packages.assert_called_once_with(connection, repository_paths)
|
||||
users.assert_called_once_with(connection, configuration)
|
||||
|
||||
|
||||
def test_migrate_data_skip(connection: Connection, configuration: Configuration,
|
||||
repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not migrate data if version is up-to-date
|
||||
"""
|
||||
packages = mocker.patch("ahriman.core.database.data.migrate_package_statuses")
|
||||
users = mocker.patch("ahriman.core.database.data.migrate_users_data")
|
||||
|
||||
migrate_data(MigrationResult(old_version=900, new_version=900), connection, configuration, repository_paths)
|
||||
packages.assert_not_called()
|
||||
users.assert_not_called()
|
43
tests/ahriman/core/database/data/test_package_statuses.py
Normal file
43
tests/ahriman/core/database/data/test_package_statuses.py
Normal file
@ -0,0 +1,43 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlite3 import Connection
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.database.data import migrate_package_statuses
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
def test_migrate_package_statuses(connection: Connection, package_ahriman: Package, repository_paths: RepositoryPaths,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must migrate packages to database
|
||||
"""
|
||||
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)
|
||||
unlink_mock = mocker.patch("pathlib.Path.unlink")
|
||||
|
||||
migrate_package_statuses(connection, repository_paths)
|
||||
unlink_mock.assert_called_once_with()
|
||||
connection.execute.assert_has_calls([
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int)),
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int)),
|
||||
])
|
||||
connection.executemany.assert_has_calls([
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int)),
|
||||
])
|
||||
connection.commit.assert_called_once_with()
|
||||
|
||||
|
||||
def test_migrate_package_statuses_skip(connection: Connection, repository_paths: RepositoryPaths,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip packages migration if no cache file found
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_file", return_value=False)
|
||||
migrate_package_statuses(connection, repository_paths)
|
||||
connection.commit.assert_not_called()
|
53
tests/ahriman/core/database/data/test_patches.py
Normal file
53
tests/ahriman/core/database/data/test_patches.py
Normal file
@ -0,0 +1,53 @@
|
||||
import pytest
|
||||
|
||||
from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlite3 import Connection
|
||||
|
||||
from ahriman.core.database.data import migrate_patches
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
def test_migrate_patches(connection: Connection, repository_paths: RepositoryPaths,
|
||||
package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must perform migration for patches
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
mocker.patch("pathlib.Path.is_file", return_value=True)
|
||||
iterdir_mock = mocker.patch("pathlib.Path.iterdir", return_value=[Path(package_ahriman.base)])
|
||||
read_mock = mocker.patch("pathlib.Path.read_text", return_value="patch")
|
||||
|
||||
migrate_patches(connection, repository_paths)
|
||||
iterdir_mock.assert_called_once_with()
|
||||
read_mock.assert_called_once_with(encoding="utf8")
|
||||
connection.execute.assert_called_once_with(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int))
|
||||
connection.commit.assert_called_once_with()
|
||||
|
||||
|
||||
def test_migrate_patches_skip(connection: Connection, repository_paths: RepositoryPaths,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip patches migration in case if no root found
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=False)
|
||||
iterdir_mock = mocker.patch("pathlib.Path.iterdir")
|
||||
|
||||
migrate_patches(connection, repository_paths)
|
||||
iterdir_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_migrate_patches_no_patch(connection: Connection, repository_paths: RepositoryPaths,
|
||||
package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip package if no match found
|
||||
"""
|
||||
mocker.patch("pathlib.Path.is_dir", return_value=True)
|
||||
mocker.patch("pathlib.Path.is_file", return_value=False)
|
||||
iterdir_mock = mocker.patch("pathlib.Path.iterdir", return_value=[Path(package_ahriman.base)])
|
||||
read_mock = mocker.patch("pathlib.Path.read_text")
|
||||
|
||||
migrate_patches(connection, repository_paths)
|
||||
iterdir_mock.assert_called_once_with()
|
||||
read_mock.assert_not_called()
|
22
tests/ahriman/core/database/data/test_users.py
Normal file
22
tests/ahriman/core/database/data/test_users.py
Normal file
@ -0,0 +1,22 @@
|
||||
import pytest
|
||||
|
||||
from sqlite3 import Connection
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.data import migrate_users_data
|
||||
|
||||
|
||||
def test_migrate_users_data(connection: Connection, configuration: Configuration) -> None:
|
||||
"""
|
||||
must users to database
|
||||
"""
|
||||
configuration.set_option("auth:read", "user1", "password1")
|
||||
configuration.set_option("auth:write", "user2", "password2")
|
||||
|
||||
migrate_users_data(connection, configuration)
|
||||
connection.execute.assert_has_calls([
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int)),
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int)),
|
||||
])
|
||||
connection.commit.assert_called_once_with()
|
15
tests/ahriman/core/database/migrations/conftest.py
Normal file
15
tests/ahriman/core/database/migrations/conftest.py
Normal file
@ -0,0 +1,15 @@
|
||||
import pytest
|
||||
|
||||
from sqlite3 import Connection
|
||||
|
||||
from ahriman.core.database.migrations import Migrations
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def migrations(connection: Connection) -> Migrations:
|
||||
"""
|
||||
fixture for migrations object
|
||||
:param connection: sqlite connection fixture
|
||||
:return: migrations test instance
|
||||
"""
|
||||
return Migrations(connection)
|
@ -0,0 +1,8 @@
|
||||
from ahriman.core.database.migrations.m000_initial import steps
|
||||
|
||||
|
||||
def test_migration_initial() -> None:
|
||||
"""
|
||||
migration must not be empty
|
||||
"""
|
||||
assert steps
|
109
tests/ahriman/core/database/migrations/test_migrations_init.py
Normal file
109
tests/ahriman/core/database/migrations/test_migrations_init.py
Normal file
@ -0,0 +1,109 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlite3 import Connection
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.database.migrations import Migrations
|
||||
from ahriman.models.migration import Migration
|
||||
from ahriman.models.migration_result import MigrationResult
|
||||
|
||||
|
||||
def test_migrate(connection: Connection, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must perform migrations
|
||||
"""
|
||||
run_mock = mocker.patch("ahriman.core.database.migrations.Migrations.run")
|
||||
Migrations.migrate(connection)
|
||||
run_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_migrations(migrations: Migrations) -> None:
|
||||
"""
|
||||
must retrieve migrations
|
||||
"""
|
||||
assert migrations.migrations()
|
||||
|
||||
|
||||
def test_run_skip(migrations: Migrations, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must skip migration if version is the same
|
||||
"""
|
||||
mocker.patch.object(MigrationResult, "is_outdated", False)
|
||||
|
||||
migrations.run()
|
||||
migrations.connection.cursor.assert_not_called()
|
||||
|
||||
|
||||
def test_run(migrations: Migrations, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run migration
|
||||
"""
|
||||
cursor = MagicMock()
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.migrations",
|
||||
return_value=[Migration(0, "test", ["select 1"])])
|
||||
migrations.connection.cursor.return_value = cursor
|
||||
validate_mock = mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
|
||||
|
||||
migrations.run()
|
||||
validate_mock.assert_called_once_with()
|
||||
cursor.execute.assert_has_calls([
|
||||
mock.call("begin exclusive"),
|
||||
mock.call("select 1"),
|
||||
mock.call("pragma user_version = 1"),
|
||||
mock.call("commit"),
|
||||
])
|
||||
cursor.close.assert_called_once_with()
|
||||
|
||||
|
||||
def test_run_migration_exception(migrations: Migrations, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must rollback and close cursor on exception during migration
|
||||
"""
|
||||
cursor = MagicMock()
|
||||
mocker.patch("logging.Logger.info", side_effect=Exception())
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.migrations",
|
||||
return_value=[Migration(0, "test", ["select 1"])])
|
||||
mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
|
||||
migrations.connection.cursor.return_value = cursor
|
||||
|
||||
with pytest.raises(Exception):
|
||||
migrations.run()
|
||||
cursor.execute.assert_has_calls([
|
||||
mock.call("begin exclusive"),
|
||||
mock.call("rollback"),
|
||||
])
|
||||
cursor.close.assert_called_once_with()
|
||||
|
||||
|
||||
def test_run_sql_exception(migrations: Migrations, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must close cursor on general migration error
|
||||
"""
|
||||
cursor = MagicMock()
|
||||
cursor.execute.side_effect = Exception()
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
|
||||
mocker.patch("ahriman.core.database.migrations.Migrations.migrations",
|
||||
return_value=[Migration(0, "test", ["select 1"])])
|
||||
mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
|
||||
migrations.connection.cursor.return_value = cursor
|
||||
|
||||
with pytest.raises(Exception):
|
||||
migrations.run()
|
||||
cursor.close.assert_called_once_with()
|
||||
|
||||
|
||||
def test_user_version(migrations: Migrations) -> None:
|
||||
"""
|
||||
must correctly extract current migration version
|
||||
"""
|
||||
cursor = MagicMock()
|
||||
cursor.fetchone.return_value = {"user_version": 42}
|
||||
migrations.connection.execute.return_value = cursor
|
||||
|
||||
version = migrations.user_version()
|
||||
migrations.connection.execute.assert_called_once_with("pragma user_version")
|
||||
assert version == 42
|
@ -0,0 +1,97 @@
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.user import User
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
|
||||
def test_user_get_update(database: SQLite, user: User) -> None:
|
||||
"""
|
||||
must retrieve user from the database
|
||||
"""
|
||||
database.user_update(user)
|
||||
assert database.user_get(user.username) == user
|
||||
|
||||
|
||||
def test_user_list(database: SQLite, user: User) -> None:
|
||||
"""
|
||||
must return all users
|
||||
"""
|
||||
database.user_update(user)
|
||||
database.user_update(User(user.password, user.username, user.access))
|
||||
|
||||
users = database.user_list(None, None)
|
||||
assert len(users) == 2
|
||||
assert user in users
|
||||
assert User(user.password, user.username, user.access) in users
|
||||
|
||||
|
||||
def test_user_list_filter_by_username(database: SQLite) -> None:
|
||||
"""
|
||||
must return users filtered by its id
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
database.user_update(second)
|
||||
database.user_update(third)
|
||||
|
||||
assert database.user_list("1", None) == [first]
|
||||
assert database.user_list("2", None) == [second]
|
||||
assert database.user_list("3", None) == [third]
|
||||
|
||||
|
||||
def test_user_list_filter_by_access(database: SQLite) -> None:
|
||||
"""
|
||||
must return users filtered by its access
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
database.user_update(second)
|
||||
database.user_update(third)
|
||||
|
||||
users = database.user_list(None, UserAccess.Read)
|
||||
assert len(users) == 2
|
||||
assert first in users
|
||||
assert third in users
|
||||
|
||||
|
||||
def test_user_list_filter_by_username_access(database: SQLite) -> None:
|
||||
"""
|
||||
must return users filtered by its access and username
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
database.user_update(second)
|
||||
database.user_update(third)
|
||||
|
||||
assert database.user_list("1", UserAccess.Read) == [first]
|
||||
assert not database.user_list("1", UserAccess.Write)
|
||||
|
||||
|
||||
def test_user_remove_update(database: SQLite, user: User) -> None:
|
||||
"""
|
||||
must remove user from the database
|
||||
"""
|
||||
database.user_update(user)
|
||||
database.user_remove(user.username)
|
||||
assert database.user_get(user.username) is None
|
||||
|
||||
|
||||
def test_user_update(database: SQLite, user: User) -> None:
|
||||
"""
|
||||
must update user in the database
|
||||
"""
|
||||
database.user_update(user)
|
||||
assert database.user_get(user.username) == user
|
||||
|
||||
new_user = user.hash_password("salt")
|
||||
new_user.access = UserAccess.Write
|
||||
database.user_update(new_user)
|
||||
assert database.user_get(new_user.username) == new_user
|
@ -0,0 +1,45 @@
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_build_queue_insert_clear(database: SQLite, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must clear all packages from queue
|
||||
"""
|
||||
database.build_queue_insert(package_ahriman)
|
||||
database.build_queue_insert(package_python_schedule)
|
||||
|
||||
database.build_queue_clear(None)
|
||||
assert not database.build_queue_get()
|
||||
|
||||
|
||||
def test_build_queue_insert_clear_specific(database: SQLite, package_ahriman: Package,
|
||||
package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must remove only specified package from the queue
|
||||
"""
|
||||
database.build_queue_insert(package_ahriman)
|
||||
database.build_queue_insert(package_python_schedule)
|
||||
|
||||
database.build_queue_clear(package_python_schedule.base)
|
||||
assert database.build_queue_get() == [package_ahriman]
|
||||
|
||||
|
||||
def test_build_queue_insert_get(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert and get package from the database
|
||||
"""
|
||||
database.build_queue_insert(package_ahriman)
|
||||
assert database.build_queue_get() == [package_ahriman]
|
||||
|
||||
|
||||
def test_build_queue_insert(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must update user in the database
|
||||
"""
|
||||
database.build_queue_insert(package_ahriman)
|
||||
assert database.build_queue_get() == [package_ahriman]
|
||||
|
||||
package_ahriman.version = "42"
|
||||
database.build_queue_insert(package_ahriman)
|
||||
assert database.build_queue_get() == [package_ahriman]
|
39
tests/ahriman/core/database/operations/test_operations.py
Normal file
39
tests/ahriman/core/database/operations/test_operations.py
Normal file
@ -0,0 +1,39 @@
|
||||
import sqlite3
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
def test_factory(database: SQLite) -> None:
|
||||
"""
|
||||
must convert response to dictionary
|
||||
"""
|
||||
result = database.with_connection(lambda conn: conn.execute("select 1 as result").fetchone())
|
||||
assert isinstance(result, dict)
|
||||
assert result["result"] == 1
|
||||
|
||||
|
||||
def test_with_connection(database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run query inside connection
|
||||
"""
|
||||
connection_mock = MagicMock()
|
||||
connect_mock = mocker.patch("sqlite3.connect", return_value=connection_mock)
|
||||
|
||||
database.with_connection(lambda conn: conn.execute("select 1"))
|
||||
connect_mock.assert_called_once_with(database.path, detect_types=sqlite3.PARSE_DECLTYPES)
|
||||
connection_mock.__enter__().commit.assert_not_called()
|
||||
|
||||
|
||||
def test_with_connection_with_commit(database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run query inside connection and commit after
|
||||
"""
|
||||
connection_mock = MagicMock()
|
||||
connection_mock.commit.return_value = 42
|
||||
mocker.patch("sqlite3.connect", return_value=connection_mock)
|
||||
|
||||
database.with_connection(lambda conn: conn.execute("select 1"), commit=True)
|
||||
connection_mock.__enter__().commit.assert_called_once_with()
|
@ -0,0 +1,168 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
from sqlite3 import Connection
|
||||
from unittest import mock
|
||||
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_package_remove_package_base(database: SQLite, connection: Connection) -> None:
|
||||
"""
|
||||
must remove package base
|
||||
"""
|
||||
database._package_remove_package_base(connection, "package")
|
||||
connection.execute.assert_has_calls([
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), {"package_base": "package"}),
|
||||
mock.call(pytest.helpers.anyvar(str, strict=True), {"package_base": "package"}),
|
||||
])
|
||||
|
||||
|
||||
def test_package_remove_packages(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must remove packages belong to base
|
||||
"""
|
||||
database._package_remove_packages(connection, package_ahriman.base, package_ahriman.packages.keys())
|
||||
connection.execute.assert_called_once_with(
|
||||
pytest.helpers.anyvar(str, strict=True), {"package_base": package_ahriman.base})
|
||||
connection.executemany.assert_called_once_with(pytest.helpers.anyvar(str, strict=True), [])
|
||||
|
||||
|
||||
def test_package_update_insert_base(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert base package
|
||||
"""
|
||||
database._package_update_insert_base(connection, package_ahriman)
|
||||
connection.execute.assert_called_once_with(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_package_update_insert_packages(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert single packages
|
||||
"""
|
||||
database._package_update_insert_packages(connection, package_ahriman)
|
||||
connection.executemany(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_package_update_insert_status(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert single package status
|
||||
"""
|
||||
database._package_update_insert_status(connection, package_ahriman.base, BuildStatus())
|
||||
connection.execute(pytest.helpers.anyvar(str, strict=True), pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
def test_packages_get_select_package_bases(database: SQLite, connection: Connection) -> None:
|
||||
"""
|
||||
must select all bases
|
||||
"""
|
||||
database._packages_get_select_package_bases(connection)
|
||||
connection.execute(pytest.helpers.anyvar(str, strict=True))
|
||||
|
||||
|
||||
def test_packages_get_select_packages(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must select all packages
|
||||
"""
|
||||
database._packages_get_select_packages(connection, {package_ahriman.base: package_ahriman})
|
||||
connection.execute(pytest.helpers.anyvar(str, strict=True))
|
||||
|
||||
|
||||
def test_packages_get_select_packages_skip(database: SQLite, connection: Connection, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must skip unknown packages
|
||||
"""
|
||||
view = {"package_base": package_ahriman.base}
|
||||
for package, properties in package_ahriman.packages.items():
|
||||
view.update({"package": package})
|
||||
view.update(properties.view())
|
||||
connection.execute.return_value = [{"package_base": "random name"}, view]
|
||||
|
||||
database._packages_get_select_packages(connection, {package_ahriman.base: package_ahriman})
|
||||
|
||||
|
||||
def test_packages_get_select_statuses(database: SQLite, connection: Connection) -> None:
|
||||
"""
|
||||
must select all statuses
|
||||
"""
|
||||
database._packages_get_select_statuses(connection)
|
||||
connection.execute(pytest.helpers.anyvar(str, strict=True))
|
||||
|
||||
|
||||
def test_package_remove(database: SQLite, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must totally remove package from the database
|
||||
"""
|
||||
remove_package_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_remove_package_base")
|
||||
remove_packages_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_remove_packages")
|
||||
|
||||
database.package_remove(package_ahriman.base)
|
||||
remove_package_mock.assert_called_once_with(pytest.helpers.anyvar(int), package_ahriman.base)
|
||||
remove_packages_mock.assert_called_once_with(pytest.helpers.anyvar(int), package_ahriman.base, [])
|
||||
|
||||
|
||||
def test_package_update(database: SQLite, package_ahriman: Package, mocker: MockerFixture):
|
||||
"""
|
||||
must update package status
|
||||
"""
|
||||
status = BuildStatus()
|
||||
insert_base_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_update_insert_base")
|
||||
insert_status_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_update_insert_status")
|
||||
insert_packages_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_update_insert_packages")
|
||||
remove_packages_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._package_remove_packages")
|
||||
|
||||
database.package_update(package_ahriman, status)
|
||||
insert_base_mock.assert_called_once_with(pytest.helpers.anyvar(int), package_ahriman)
|
||||
insert_status_mock.assert_called_once_with(pytest.helpers.anyvar(int), package_ahriman.base, status)
|
||||
insert_packages_mock.assert_called_once_with(pytest.helpers.anyvar(int), package_ahriman)
|
||||
remove_packages_mock.assert_called_once_with(
|
||||
pytest.helpers.anyvar(int), package_ahriman.base, package_ahriman.packages.keys())
|
||||
|
||||
|
||||
def test_packages_get(database: SQLite, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return all packages
|
||||
"""
|
||||
select_bases_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._packages_get_select_package_bases",
|
||||
return_value={package_ahriman.base: package_ahriman})
|
||||
select_packages_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._packages_get_select_packages")
|
||||
select_statuses_mock = mocker.patch("ahriman.core.database.sqlite.SQLite._packages_get_select_statuses")
|
||||
|
||||
database.packages_get()
|
||||
select_bases_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
select_statuses_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
select_packages_mock.assert_called_once_with(pytest.helpers.anyvar(int), {package_ahriman.base: package_ahriman})
|
||||
|
||||
|
||||
def test_package_update_get(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert and retrieve package
|
||||
"""
|
||||
status = BuildStatus()
|
||||
database.package_update(package_ahriman, status)
|
||||
assert next((db_package, db_status)
|
||||
for db_package, db_status in database.packages_get()
|
||||
if db_package.base == package_ahriman.base) == (package_ahriman, status)
|
||||
|
||||
|
||||
def test_package_update_remove_get(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must insert, remove and retrieve package
|
||||
"""
|
||||
status = BuildStatus()
|
||||
database.package_update(package_ahriman, status)
|
||||
database.package_remove(package_ahriman.base)
|
||||
assert not database.packages_get()
|
||||
|
||||
|
||||
def test_package_update_update(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must perform update for existing package
|
||||
"""
|
||||
database.package_update(package_ahriman, BuildStatus())
|
||||
database.package_update(package_ahriman, BuildStatus(BuildStatusEnum.Failed))
|
||||
assert next(db_status.status
|
||||
for db_package, db_status in database.packages_get()
|
||||
if db_package.base == package_ahriman.base) == BuildStatusEnum.Failed
|
@ -0,0 +1,55 @@
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_patches_get_insert(database: SQLite, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must insert patch to database
|
||||
"""
|
||||
database.patches_insert(package_ahriman.base, "patch_1")
|
||||
database.patches_insert(package_python_schedule.base, "patch_2")
|
||||
assert database.patches_get(package_ahriman.base) == "patch_1"
|
||||
assert not database.build_queue_get()
|
||||
|
||||
|
||||
def test_patches_list(database: SQLite, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must list all patches
|
||||
"""
|
||||
database.patches_insert(package_ahriman.base, "patch1")
|
||||
database.patches_insert(package_python_schedule.base, "patch2")
|
||||
assert database.patches_list(None) == {package_ahriman.base: "patch1", package_python_schedule.base: "patch2"}
|
||||
|
||||
|
||||
def test_patches_list_filter(database: SQLite, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must list all patches filtered by package name (same as get)
|
||||
"""
|
||||
database.patches_insert(package_ahriman.base, "patch1")
|
||||
database.patches_insert(package_python_schedule.base, "patch2")
|
||||
|
||||
assert database.patches_list(package_ahriman.base) == {package_ahriman.base: "patch1"}
|
||||
assert database.patches_list(package_python_schedule.base) == {package_python_schedule.base: "patch2"}
|
||||
|
||||
|
||||
def test_patches_insert_remove(database: SQLite, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
"""
|
||||
must remove patch from database
|
||||
"""
|
||||
database.patches_insert(package_ahriman.base, "patch_1")
|
||||
database.patches_insert(package_python_schedule.base, "patch_2")
|
||||
database.patches_remove(package_ahriman.base)
|
||||
|
||||
assert database.patches_get(package_ahriman.base) is None
|
||||
database.patches_insert(package_python_schedule.base, "patch_2")
|
||||
|
||||
|
||||
def test_patches_insert_insert(database: SQLite, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must update patch in database
|
||||
"""
|
||||
database.patches_insert(package_ahriman.base, "patch_1")
|
||||
assert database.patches_get(package_ahriman.base) == "patch_1"
|
||||
|
||||
database.patches_insert(package_ahriman.base, "patch_2")
|
||||
assert database.patches_get(package_ahriman.base) == "patch_2"
|
28
tests/ahriman/core/database/test_sqlite.py
Normal file
28
tests/ahriman/core/database/test_sqlite.py
Normal file
@ -0,0 +1,28 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
|
||||
|
||||
def test_load(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must correctly load instance
|
||||
"""
|
||||
init_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.init")
|
||||
SQLite.load(configuration)
|
||||
init_mock.assert_called_once_with(configuration)
|
||||
|
||||
|
||||
def test_init(database: SQLite, configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must run migrations on init
|
||||
"""
|
||||
migrate_schema_mock = mocker.patch("ahriman.core.database.migrations.Migrations.migrate")
|
||||
migrate_data_mock = mocker.patch("ahriman.core.database.sqlite.migrate_data")
|
||||
|
||||
database.init(configuration)
|
||||
migrate_schema_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
migrate_data_mock.assert_called_once_with(
|
||||
pytest.helpers.anyvar(int), pytest.helpers.anyvar(int), configuration, configuration.repository_paths)
|
@ -6,9 +6,11 @@ from ahriman.core.formatters.package_printer import PackagePrinter
|
||||
from ahriman.core.formatters.status_printer import StatusPrinter
|
||||
from ahriman.core.formatters.string_printer import StringPrinter
|
||||
from ahriman.core.formatters.update_printer import UpdatePrinter
|
||||
from ahriman.core.formatters.user_printer import UserPrinter
|
||||
from ahriman.models.aur_package import AURPackage
|
||||
from ahriman.models.build_status import BuildStatus
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.user import User
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@ -65,3 +67,13 @@ def update_printer(package_ahriman: Package) -> UpdatePrinter:
|
||||
:return: build status printer test instance
|
||||
"""
|
||||
return UpdatePrinter(package_ahriman, None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_printer(user: User) -> UserPrinter:
|
||||
"""
|
||||
fixture for user printer
|
||||
:param user: user fixture
|
||||
:return: user printer test instance
|
||||
"""
|
||||
return UserPrinter(user)
|
||||
|
15
tests/ahriman/core/formatters/test_user_printer.py
Normal file
15
tests/ahriman/core/formatters/test_user_printer.py
Normal file
@ -0,0 +1,15 @@
|
||||
from ahriman.core.formatters.user_printer import UserPrinter
|
||||
|
||||
|
||||
def test_properties(user_printer: UserPrinter) -> None:
|
||||
"""
|
||||
must return non empty properties list
|
||||
"""
|
||||
assert user_printer.properties()
|
||||
|
||||
|
||||
def test_title(user_printer: UserPrinter) -> None:
|
||||
"""
|
||||
must return non empty title
|
||||
"""
|
||||
assert user_printer.title() is not None
|
@ -3,6 +3,7 @@ import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.repository import Repository
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
from ahriman.core.repository.executor import Executor
|
||||
@ -11,68 +12,71 @@ from ahriman.core.repository.update_handler import UpdateHandler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cleaner(configuration: Configuration, mocker: MockerFixture) -> Cleaner:
|
||||
def cleaner(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Cleaner:
|
||||
"""
|
||||
fixture for cleaner
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: cleaner test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
return Cleaner("x86_64", configuration, no_report=True, unsafe=False)
|
||||
return Cleaner("x86_64", configuration, database, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def executor(configuration: Configuration, mocker: MockerFixture) -> Executor:
|
||||
def executor(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Executor:
|
||||
"""
|
||||
fixture for executor
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: executor test instance
|
||||
"""
|
||||
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")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_queue")
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
return Executor("x86_64", configuration, no_report=True, unsafe=False)
|
||||
return Executor("x86_64", configuration, database, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repository(configuration: Configuration, mocker: MockerFixture) -> Repository:
|
||||
def repository(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> Repository:
|
||||
"""
|
||||
fixture for repository
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: repository test instance
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
return Repository("x86_64", configuration, no_report=True, unsafe=False)
|
||||
return Repository("x86_64", configuration, database, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def properties(configuration: Configuration) -> Properties:
|
||||
def properties(configuration: Configuration, database: SQLite) -> Properties:
|
||||
"""
|
||||
fixture for properties
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:return: properties test instance
|
||||
"""
|
||||
return Properties("x86_64", configuration, no_report=True, unsafe=False)
|
||||
return Properties("x86_64", configuration, database, no_report=True, unsafe=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def update_handler(configuration: Configuration, mocker: MockerFixture) -> UpdateHandler:
|
||||
def update_handler(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> UpdateHandler:
|
||||
"""
|
||||
fixture for update handler
|
||||
:param configuration: configuration fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: update handler test instance
|
||||
"""
|
||||
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")
|
||||
mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_queue")
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
return UpdateHandler("x86_64", configuration, no_report=True, unsafe=False)
|
||||
return UpdateHandler("x86_64", configuration, database, no_report=True, unsafe=False)
|
||||
|
@ -36,15 +36,6 @@ def test_packages_built(cleaner: Cleaner) -> None:
|
||||
cleaner.packages_built()
|
||||
|
||||
|
||||
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
|
||||
@ -63,15 +54,6 @@ def test_clear_chroot(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
_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
|
||||
@ -84,10 +66,10 @@ def test_clear_packages(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
Path.unlink.assert_has_calls([mock.call(), mock.call(), mock.call()])
|
||||
|
||||
|
||||
def test_clear_patches(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
def test_clear_queue(cleaner: Cleaner, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must clear directory with patches
|
||||
must clear queued packages from the database
|
||||
"""
|
||||
_mock_clear(mocker)
|
||||
cleaner.clear_patches()
|
||||
_mock_clear_check()
|
||||
clear_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_clear")
|
||||
cleaner.clear_queue()
|
||||
clear_mock.assert_called_once_with(None)
|
||||
|
@ -41,9 +41,6 @@ def test_process_build(executor: Executor, package_ahriman: Package, mocker: Moc
|
||||
move_mock.assert_called_once_with(Path(package_ahriman.base), executor.paths.packages / package_ahriman.base)
|
||||
# must update status
|
||||
status_client_mock.assert_called_once_with(package_ahriman.base)
|
||||
# must clear directory
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
Cleaner.clear_build.assert_called_once_with()
|
||||
|
||||
|
||||
def test_process_build_failure(executor: Executor, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
|
@ -1,51 +1,52 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.exceptions import UnsafeRun
|
||||
from ahriman.core.repository.properties import Properties
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
|
||||
|
||||
def test_create_tree_on_load(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_create_tree_on_load(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create tree on load
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.properties.check_user")
|
||||
tree_create_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
Properties("x86_64", configuration, True, False)
|
||||
Properties("x86_64", configuration, database, True, False)
|
||||
|
||||
tree_create_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_create_tree_on_load_unsafe(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_create_tree_on_load_unsafe(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not create tree on load in case if user differs from the root owner
|
||||
"""
|
||||
mocker.patch("ahriman.core.repository.properties.check_user", side_effect=UnsafeRun(0, 1))
|
||||
tree_create_mock = mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
Properties("x86_64", configuration, True, False)
|
||||
Properties("x86_64", configuration, database, True, False)
|
||||
|
||||
tree_create_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_create_dummy_report_client(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_create_dummy_report_client(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create dummy report client if report is disabled
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
load_mock = mocker.patch("ahriman.core.status.client.Client.load")
|
||||
properties = Properties("x86_64", configuration, True, False)
|
||||
properties = Properties("x86_64", configuration, database, True, False)
|
||||
|
||||
load_mock.assert_not_called()
|
||||
assert not isinstance(properties.reporter, WebClient)
|
||||
|
||||
|
||||
def test_create_full_report_client(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_create_full_report_client(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must create load report client if report is enabled
|
||||
"""
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
load_mock = mocker.patch("ahriman.core.status.client.Client.load")
|
||||
Properties("x86_64", configuration, False, False)
|
||||
Properties("x86_64", configuration, database, False, False)
|
||||
|
||||
load_mock.assert_called_once_with(configuration)
|
||||
|
@ -168,7 +168,7 @@ def test_updates_manual_clear(update_handler: UpdateHandler, mocker: MockerFixtu
|
||||
update_handler.updates_manual()
|
||||
|
||||
from ahriman.core.repository.cleaner import Cleaner
|
||||
Cleaner.clear_manual.assert_called_once_with()
|
||||
Cleaner.clear_queue.assert_called_once_with()
|
||||
|
||||
|
||||
def test_updates_manual_status_known(update_handler: UpdateHandler, package_ahriman: Package,
|
||||
@ -176,9 +176,8 @@ def test_updates_manual_status_known(update_handler: UpdateHandler, package_ahri
|
||||
"""
|
||||
must create record for known package via reporter
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_get", return_value=[package_ahriman])
|
||||
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()
|
||||
@ -190,9 +189,8 @@ def test_updates_manual_status_unknown(update_handler: UpdateHandler, package_ah
|
||||
"""
|
||||
must create record for unknown package via reporter
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[package_ahriman.base])
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_get", return_value=[package_ahriman])
|
||||
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()
|
||||
@ -204,8 +202,6 @@ def test_updates_manual_with_failures(update_handler: UpdateHandler, package_ahr
|
||||
"""
|
||||
must process manual 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())
|
||||
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.build_queue_get", side_effect=Exception())
|
||||
mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.packages", return_value=[package_ahriman])
|
||||
assert update_handler.updates_manual() == []
|
||||
|
@ -1,33 +1,8 @@
|
||||
import pytest
|
||||
|
||||
from typing import Any, Dict
|
||||
|
||||
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 BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
# helpers
|
||||
@pytest.helpers.register
|
||||
def get_package_status(package: Package) -> Dict[str, Any]:
|
||||
"""
|
||||
helper to extract package status from package
|
||||
:param package: package object
|
||||
:return: simplified package status map (with only status and view)
|
||||
"""
|
||||
return {"status": BuildStatusEnum.Unknown.value, "package": package.view()}
|
||||
|
||||
|
||||
@pytest.helpers.register
|
||||
def get_package_status_extended(package: Package) -> Dict[str, Any]:
|
||||
"""
|
||||
helper to extract package status from package
|
||||
:param package: package object
|
||||
:return: full package status map (with timestamped build status and view)
|
||||
"""
|
||||
return {"status": BuildStatus().view(), "package": package.view()}
|
||||
|
||||
|
||||
# fixtures
|
||||
@ -47,5 +22,5 @@ def web_client(configuration: Configuration) -> WebClient:
|
||||
:param configuration: configuration fixture
|
||||
:return: web client test instance
|
||||
"""
|
||||
configuration.set("web", "port", 8080)
|
||||
configuration.set("web", "port", "8080")
|
||||
return WebClient(configuration)
|
||||
|
@ -61,13 +61,6 @@ def test_get_self(client: Client) -> None:
|
||||
assert client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_reload_auth(client: Client) -> None:
|
||||
"""
|
||||
must process auth reload without errors
|
||||
"""
|
||||
client.reload_auth()
|
||||
|
||||
|
||||
def test_remove(client: Client, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must process remove without errors
|
||||
|
@ -6,6 +6,7 @@ from pytest_mock import MockerFixture
|
||||
from unittest.mock import PropertyMock
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.exceptions import UnknownPackage
|
||||
from ahriman.core.status.watcher import Watcher
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
@ -13,7 +14,7 @@ from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.package import Package
|
||||
|
||||
|
||||
def test_force_no_report(configuration: Configuration, mocker: MockerFixture) -> None:
|
||||
def test_force_no_report(configuration: Configuration, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must force dummy report client
|
||||
"""
|
||||
@ -21,122 +22,12 @@ def test_force_no_report(configuration: Configuration, mocker: MockerFixture) ->
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
|
||||
load_mock = mocker.patch("ahriman.core.status.client.Client.load")
|
||||
watcher = Watcher("x86_64", configuration)
|
||||
watcher = Watcher("x86_64", configuration, database)
|
||||
|
||||
load_mock.assert_not_called()
|
||||
assert not isinstance(watcher.repository.reporter, WebClient)
|
||||
|
||||
|
||||
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_package_load_error(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not fail on json errors
|
||||
"""
|
||||
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("ahriman.models.package.Package.from_json", side_effect=Exception())
|
||||
mocker.patch("json.load", return_value=response)
|
||||
|
||||
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_with(pytest.helpers.anyvar(int), pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
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()) # nosec
|
||||
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
|
||||
@ -160,7 +51,7 @@ def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture)
|
||||
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")
|
||||
cache_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.packages_get")
|
||||
|
||||
watcher.load()
|
||||
cache_mock.assert_called_once_with()
|
||||
@ -173,9 +64,10 @@ def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFi
|
||||
"""
|
||||
must correctly load packages with known statuses
|
||||
"""
|
||||
status = BuildStatus(BuildStatusEnum.Success)
|
||||
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))}
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.packages_get", return_value=[(package_ahriman, status)])
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, status)}
|
||||
|
||||
watcher.load()
|
||||
_, status = watcher.known[package_ahriman.base]
|
||||
@ -186,32 +78,32 @@ def test_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixtur
|
||||
"""
|
||||
must remove package base
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
cache_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.package_remove")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
assert not watcher.known
|
||||
cache_mock.assert_called_once_with()
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
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")
|
||||
cache_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.package_remove")
|
||||
|
||||
watcher.remove(package_ahriman.base)
|
||||
cache_mock.assert_called_once_with()
|
||||
cache_mock.assert_called_once_with(package_ahriman.base)
|
||||
|
||||
|
||||
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")
|
||||
cache_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.package_update")
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
|
||||
cache_mock.assert_called_once_with()
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
package, status = watcher.known[package_ahriman.base]
|
||||
assert package == package_ahriman
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
@ -221,25 +113,22 @@ def test_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerF
|
||||
"""
|
||||
must update package status only for known package
|
||||
"""
|
||||
cache_mock = mocker.patch("ahriman.core.status.watcher.Watcher._cache_save")
|
||||
cache_mock = mocker.patch("ahriman.core.database.sqlite.SQLite.package_update")
|
||||
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
|
||||
|
||||
watcher.update(package_ahriman.base, BuildStatusEnum.Success, None)
|
||||
cache_mock.assert_called_once_with()
|
||||
cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
|
||||
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:
|
||||
def test_update_unknown(watcher: Watcher, package_ahriman: Package) -> 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_with()
|
||||
|
||||
|
||||
def test_update_self(watcher: Watcher) -> None:
|
||||
|
@ -230,32 +230,6 @@ def test_get_self_failed_http_error(web_client: WebClient, mocker: MockerFixture
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_reload_auth(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process auth reload
|
||||
"""
|
||||
requests_mock = mocker.patch("requests.Session.post")
|
||||
|
||||
web_client.reload_auth()
|
||||
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True))
|
||||
|
||||
|
||||
def test_reload_auth_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during auth reload
|
||||
"""
|
||||
mocker.patch("requests.Session.post", side_effect=Exception())
|
||||
web_client.reload_auth()
|
||||
|
||||
|
||||
def test_reload_auth_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during removal
|
||||
"""
|
||||
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
|
||||
web_client.reload_auth()
|
||||
|
||||
|
||||
def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must process package removal
|
||||
|
@ -8,6 +8,14 @@ from unittest import mock
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.exceptions import InitializeException
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
def test_repository_paths(configuration: Configuration, repository_paths: RepositoryPaths) -> None:
|
||||
"""
|
||||
must return repository paths
|
||||
"""
|
||||
assert configuration.repository_paths == repository_paths
|
||||
|
||||
|
||||
def test_from_path(mocker: MockerFixture) -> None:
|
||||
@ -40,6 +48,33 @@ def test_from_path_file_missing(mocker: MockerFixture) -> None:
|
||||
read_mock.assert_called_once_with(configuration.SYSTEM_CONFIGURATION_PATH)
|
||||
|
||||
|
||||
def test_check_loaded(configuration: Configuration) -> None:
|
||||
"""
|
||||
must return valid path and architecture
|
||||
"""
|
||||
path, architecture = configuration.check_loaded()
|
||||
assert path == configuration.path
|
||||
assert architecture == configuration.architecture
|
||||
|
||||
|
||||
def test_check_loaded_path(configuration: Configuration) -> None:
|
||||
"""
|
||||
must raise exception if path is none
|
||||
"""
|
||||
configuration.path = None
|
||||
with pytest.raises(InitializeException):
|
||||
configuration.check_loaded()
|
||||
|
||||
|
||||
def test_check_loaded_architecture(configuration: Configuration) -> None:
|
||||
"""
|
||||
must raise exception if architecture is none
|
||||
"""
|
||||
configuration.architecture = None
|
||||
with pytest.raises(InitializeException):
|
||||
configuration.check_loaded()
|
||||
|
||||
|
||||
def test_dump(configuration: Configuration) -> None:
|
||||
"""
|
||||
dump must not be empty
|
||||
|
@ -2,9 +2,9 @@ import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.tree import Leaf, Tree
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
|
||||
def test_leaf_is_root_empty(leaf_ahriman: Leaf) -> None:
|
||||
@ -37,7 +37,7 @@ def test_leaf_is_root_true(leaf_ahriman: Leaf, leaf_python_schedule: Leaf) -> No
|
||||
assert not leaf_ahriman.is_root([leaf_python_schedule])
|
||||
|
||||
|
||||
def test_leaf_load(package_ahriman: Package, repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
|
||||
def test_leaf_load(package_ahriman: Package, database: SQLite, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must load with dependencies
|
||||
"""
|
||||
@ -46,12 +46,12 @@ def test_leaf_load(package_ahriman: Package, repository_paths: RepositoryPaths,
|
||||
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies", return_value={"ahriman-dependency"})
|
||||
rmtree_mock = mocker.patch("shutil.rmtree")
|
||||
|
||||
leaf = Leaf.load(package_ahriman, repository_paths)
|
||||
leaf = Leaf.load(package_ahriman, database)
|
||||
assert leaf.package == package_ahriman
|
||||
assert leaf.dependencies == {"ahriman-dependency"}
|
||||
tempdir_mock.assert_called_once_with()
|
||||
load_mock.assert_called_once_with(
|
||||
pytest.helpers.anyvar(int), package_ahriman.git_url, repository_paths.patches_for(package_ahriman.base))
|
||||
pytest.helpers.anyvar(int), package_ahriman.git_url, database.patches_get(package_ahriman.base))
|
||||
dependencies_mock.assert_called_once_with(pytest.helpers.anyvar(int))
|
||||
rmtree_mock.assert_called_once_with(pytest.helpers.anyvar(int), ignore_errors=True)
|
||||
|
||||
@ -69,8 +69,8 @@ def test_tree_levels(leaf_ahriman: Leaf, leaf_python_schedule: Leaf) -> None:
|
||||
assert second == [leaf_ahriman.package]
|
||||
|
||||
|
||||
def test_tree_load(package_ahriman: Package, package_python_schedule: Package,
|
||||
repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
|
||||
def test_tree_load(package_ahriman: Package, package_python_schedule: Package, database: SQLite,
|
||||
mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must package list
|
||||
"""
|
||||
@ -79,5 +79,5 @@ def test_tree_load(package_ahriman: Package, package_python_schedule: Package,
|
||||
mocker.patch("ahriman.models.package.Package.dependencies")
|
||||
mocker.patch("shutil.rmtree")
|
||||
|
||||
tree = Tree.load([package_ahriman, package_python_schedule], repository_paths)
|
||||
tree = Tree.load([package_ahriman, package_python_schedule], database)
|
||||
assert len(tree.leaves) == 2
|
||||
|
@ -7,7 +7,7 @@ from pathlib import Path
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.exceptions import InvalidOption, UnsafeRun
|
||||
from ahriman.core.util import check_output, check_user, filter_json, package_like, pretty_datetime, pretty_size, walk
|
||||
from ahriman.core.util import check_output, check_user, filter_json, package_like, pretty_datetime, pretty_size, tmpdir, walk
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.repository_paths import RepositoryPaths
|
||||
|
||||
@ -207,6 +207,25 @@ def test_pretty_size_empty() -> None:
|
||||
assert pretty_size(None) == ""
|
||||
|
||||
|
||||
def test_tmpdir() -> None:
|
||||
"""
|
||||
must create temporary directory and remove it after
|
||||
"""
|
||||
with tmpdir() as directory:
|
||||
assert directory.is_dir()
|
||||
assert not directory.exists()
|
||||
|
||||
|
||||
def test_tmpdir_failure() -> None:
|
||||
"""
|
||||
must create temporary directory and remove it even after exception
|
||||
"""
|
||||
with pytest.raises(Exception):
|
||||
with tmpdir() as directory:
|
||||
raise Exception()
|
||||
assert not directory.exists()
|
||||
|
||||
|
||||
def test_walk(resource_path_root: Path) -> None:
|
||||
"""
|
||||
must traverse directory recursively
|
||||
|
34
tests/ahriman/models/test_migration.py
Normal file
34
tests/ahriman/models/test_migration.py
Normal file
@ -0,0 +1,34 @@
|
||||
import pytest
|
||||
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.core.exceptions import MigrationError
|
||||
from ahriman.models.migration_result import MigrationResult
|
||||
|
||||
|
||||
def test_is_outdated() -> None:
|
||||
"""
|
||||
must return False for outdated schema
|
||||
"""
|
||||
assert MigrationResult(old_version=0, new_version=1).is_outdated
|
||||
assert not MigrationResult(old_version=1, new_version=1).is_outdated
|
||||
|
||||
|
||||
def test_is_outdated_validation(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call validation before version check
|
||||
"""
|
||||
validate_mock = mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
|
||||
assert MigrationResult(old_version=0, new_version=1).is_outdated
|
||||
validate_mock.assert_called_once_with()
|
||||
|
||||
|
||||
def test_validate() -> None:
|
||||
"""
|
||||
must raise exception on invalid migration versions
|
||||
"""
|
||||
with pytest.raises(MigrationError):
|
||||
MigrationResult(old_version=-1, new_version=0).validate()
|
||||
|
||||
with pytest.raises(MigrationError):
|
||||
MigrationResult(old_version=1, new_version=0).validate()
|
0
tests/ahriman/models/test_migration_result.py
Normal file
0
tests/ahriman/models/test_migration_result.py
Normal file
@ -44,3 +44,10 @@ def test_from_package(package_description_ahriman: PackageDescription,
|
||||
package_description = PackageDescription.from_package(pyalpm_package_description_ahriman,
|
||||
package_description_ahriman.filepath)
|
||||
assert package_description_ahriman == package_description
|
||||
|
||||
|
||||
def test_from_json_view(package_description_ahriman: PackageDescription) -> None:
|
||||
"""
|
||||
must generate same description from json view
|
||||
"""
|
||||
assert PackageDescription.from_json(package_description_ahriman.view()) == package_description_ahriman
|
||||
|
@ -111,33 +111,6 @@ def test_chown_invalid_path(repository_paths: RepositoryPaths) -> None:
|
||||
repository_paths.chown(repository_paths.root.parent)
|
||||
|
||||
|
||||
def test_manual_for(repository_paths: RepositoryPaths, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return correct path for manual directory
|
||||
"""
|
||||
path = repository_paths.manual_for(package_ahriman.base)
|
||||
assert path.name == package_ahriman.base
|
||||
assert path.parent == repository_paths.manual
|
||||
|
||||
|
||||
def test_patches_for(repository_paths: RepositoryPaths, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return correct path for patches directory
|
||||
"""
|
||||
path = repository_paths.patches_for(package_ahriman.base)
|
||||
assert path.name == package_ahriman.base
|
||||
assert path.parent == repository_paths.patches
|
||||
|
||||
|
||||
def test_sources_for(repository_paths: RepositoryPaths, package_ahriman: Package) -> None:
|
||||
"""
|
||||
must return correct path for sources directory
|
||||
"""
|
||||
path = repository_paths.sources_for(package_ahriman.base)
|
||||
assert path.name == package_ahriman.base
|
||||
assert path.parent == repository_paths.sources
|
||||
|
||||
|
||||
def test_tree_clear(repository_paths: RepositoryPaths, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must remove any package related files
|
||||
|
@ -27,7 +27,7 @@ def test_check_credentials_hash_password(user: User) -> None:
|
||||
must generate and validate user password
|
||||
"""
|
||||
current_password = user.password
|
||||
user.password = user.hash_password("salt")
|
||||
user = user.hash_password("salt")
|
||||
assert user.check_credentials(current_password, "salt")
|
||||
assert not user.check_credentials(current_password, "salt1")
|
||||
assert not user.check_credentials(user.password, "salt")
|
||||
@ -48,9 +48,9 @@ def test_hash_password_empty_hash(user: User) -> None:
|
||||
must return empty string after hash in case if password not set
|
||||
"""
|
||||
user.password = ""
|
||||
assert user.hash_password("salt") == ""
|
||||
assert user.hash_password("salt") == user
|
||||
user.password = None
|
||||
assert user.hash_password("salt") == ""
|
||||
assert user.hash_password("salt") == user
|
||||
|
||||
|
||||
def test_generate_password() -> None:
|
||||
|
@ -8,6 +8,7 @@ from typing import Any
|
||||
import ahriman.core.auth.helpers
|
||||
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.core.spawn import Spawn
|
||||
from ahriman.models.user import User
|
||||
from ahriman.web.web import setup_service
|
||||
@ -31,53 +32,60 @@ def request(app: web.Application, path: str, method: str, json: Any = None, data
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application(configuration: Configuration, spawner: Spawn, mocker: MockerFixture) -> web.Application:
|
||||
def application(configuration: Configuration, spawner: Spawn, database: SQLite,
|
||||
mocker: MockerFixture) -> web.Application:
|
||||
"""
|
||||
application fixture
|
||||
:param configuration: configuration fixture
|
||||
:param spawner: spawner fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", False)
|
||||
return setup_service("x86_64", configuration, spawner)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_with_auth(configuration: Configuration, user: User, spawner: Spawn,
|
||||
def application_with_auth(configuration: Configuration, user: User, spawner: Spawn, database: SQLite,
|
||||
mocker: MockerFixture) -> web.Application:
|
||||
"""
|
||||
application fixture with auth enabled
|
||||
:param configuration: configuration fixture
|
||||
:param user: user descriptor fixture
|
||||
:param spawner: spawner fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
configuration.set_option("auth", "target", "configuration")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", True)
|
||||
application = setup_service("x86_64", configuration, spawner)
|
||||
|
||||
generated = User(user.username, user.hash_password(application["validator"].salt), user.access)
|
||||
application["validator"]._users[generated.username] = generated
|
||||
generated = user.hash_password(application["validator"].salt)
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=generated)
|
||||
|
||||
return application
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_with_debug(configuration: Configuration, user: User, spawner: Spawn,
|
||||
def application_with_debug(configuration: Configuration, user: User, spawner: Spawn, database: SQLite,
|
||||
mocker: MockerFixture) -> web.Application:
|
||||
"""
|
||||
application fixture with debug enabled
|
||||
:param configuration: configuration fixture
|
||||
:param user: user descriptor fixture
|
||||
:param spawner: spawner fixture
|
||||
:param database: database fixture
|
||||
:param mocker: mocker object
|
||||
:return: application test instance
|
||||
"""
|
||||
configuration.set_option("web", "debug", "yes")
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.load", return_value=database)
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", False)
|
||||
return setup_service("x86_64", configuration, spawner)
|
||||
|
@ -2,18 +2,18 @@ import pytest
|
||||
|
||||
from ahriman.core.auth.auth import Auth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database.sqlite import SQLite
|
||||
from ahriman.models.user import User
|
||||
from ahriman.web.middlewares.auth_handler import AuthorizationPolicy
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authorization_policy(configuration: Configuration, user: User) -> AuthorizationPolicy:
|
||||
def authorization_policy(configuration: Configuration, database: SQLite, user: User) -> AuthorizationPolicy:
|
||||
"""
|
||||
fixture for authorization policy
|
||||
:return: authorization policy fixture
|
||||
"""
|
||||
configuration.set_option("auth", "target", "configuration")
|
||||
validator = Auth.load(configuration)
|
||||
validator = Auth.load(configuration, database)
|
||||
policy = AuthorizationPolicy(validator)
|
||||
policy.validator._users = {user.username: user}
|
||||
return policy
|
||||
|
@ -20,11 +20,18 @@ def _identity(username: str) -> str:
|
||||
return f"{username} {UserIdentity.expire_when(60)}"
|
||||
|
||||
|
||||
async def test_authorized_userid(authorization_policy: AuthorizationPolicy, user: User) -> None:
|
||||
async def test_authorized_userid(authorization_policy: AuthorizationPolicy, user: User, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return authorized user id
|
||||
"""
|
||||
mocker.patch("ahriman.core.database.sqlite.SQLite.user_get", return_value=user)
|
||||
assert await authorization_policy.authorized_userid(_identity(user.username)) == user.username
|
||||
|
||||
|
||||
async def test_authorized_userid_unknown(authorization_policy: AuthorizationPolicy, user: User) -> None:
|
||||
"""
|
||||
must not allow unknown user id for authorization
|
||||
"""
|
||||
assert await authorization_policy.authorized_userid(_identity("somerandomname")) is None
|
||||
assert await authorization_policy.authorized_userid("somerandomname") is None
|
||||
|
||||
|
@ -2,11 +2,12 @@ import pytest
|
||||
|
||||
from aiohttp import web
|
||||
from asyncio import BaseEventLoop
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.auth.oauth import OAuth
|
||||
from ahriman.web.views.base import BaseView
|
||||
|
||||
|
||||
@ -21,30 +22,46 @@ def base(application: web.Application) -> BaseView:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: web.Application, loop: BaseEventLoop,
|
||||
def client(application: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture
|
||||
:param application: application fixture
|
||||
:param loop: context event loop
|
||||
:param event_loop: context event loop
|
||||
:param aiohttp_client: aiohttp client fixture
|
||||
:param mocker: mocker object
|
||||
:return: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return loop.run_until_complete(aiohttp_client(application))
|
||||
return event_loop.run_until_complete(aiohttp_client(application))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_auth(application_with_auth: web.Application, loop: BaseEventLoop,
|
||||
def client_with_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
:param application_with_auth: application fixture
|
||||
:param loop: context event loop
|
||||
:param event_loop: context event loop
|
||||
:param aiohttp_client: aiohttp client fixture
|
||||
:param mocker: mocker object
|
||||
:return: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_oauth_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
:param application_with_auth: application fixture
|
||||
:param event_loop: context event loop
|
||||
:param aiohttp_client: aiohttp client fixture
|
||||
:param mocker: mocker object
|
||||
:return: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
application_with_auth["validator"] = MagicMock(spec=OAuth)
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
@ -1,41 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.models.user_access import UserAccess
|
||||
from ahriman.web.views.service.reload_auth import ReloadAuthView
|
||||
|
||||
|
||||
async def test_get_permission() -> None:
|
||||
"""
|
||||
must return correct permission for the request
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await ReloadAuthView.get_permission(request) == UserAccess.Write
|
||||
|
||||
|
||||
async def test_post(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call post request correctly
|
||||
"""
|
||||
mocker.patch("aiohttp_security.check_permission", return_value=True)
|
||||
reload_mock = mocker.patch("ahriman.core.configuration.Configuration.reload")
|
||||
load_mock = mocker.patch("ahriman.core.auth.auth.Auth.load")
|
||||
response = await client_with_auth.post("/service-api/v1/reload-auth")
|
||||
|
||||
assert response.ok
|
||||
reload_mock.assert_called_once_with()
|
||||
load_mock.assert_called_once_with(client_with_auth.app["configuration"])
|
||||
|
||||
|
||||
async def test_post_no_auth(client: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call return 500 if no authorization module loaded
|
||||
"""
|
||||
reload_mock = mocker.patch("ahriman.core.configuration.Configuration.reload")
|
||||
response = await client.post("/service-api/v1/reload-auth")
|
||||
|
||||
assert response.status == 500
|
||||
reload_mock.assert_called_once_with()
|
@ -12,6 +12,13 @@ def test_configuration(base: BaseView) -> None:
|
||||
assert base.configuration
|
||||
|
||||
|
||||
def test_database(base: BaseView) -> None:
|
||||
"""
|
||||
must return database
|
||||
"""
|
||||
assert base.database
|
||||
|
||||
|
||||
def test_service(base: BaseView) -> None:
|
||||
"""
|
||||
must return service
|
||||
|
@ -52,7 +52,7 @@ async def test_get_static(client: TestClient) -> None:
|
||||
|
||||
async def test_get_static_with_auth(client_with_auth: TestClient) -> None:
|
||||
"""
|
||||
must return static files
|
||||
must return static files with authorization enabled
|
||||
"""
|
||||
response = await client_with_auth.get("/static/favicon.ico")
|
||||
assert response.ok
|
||||
|
@ -27,42 +27,42 @@ async def test_get_default_validator(client_with_auth: TestClient) -> None:
|
||||
assert get_response.status == 405
|
||||
|
||||
|
||||
async def test_get_redirect_to_oauth(client_with_auth: TestClient) -> None:
|
||||
async def test_get_redirect_to_oauth(client_with_oauth_auth: TestClient) -> None:
|
||||
"""
|
||||
must redirect to OAuth service provider in case if no code is supplied
|
||||
"""
|
||||
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
||||
oauth = client_with_oauth_auth.app["validator"]
|
||||
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
||||
|
||||
get_response = await client_with_auth.get("/user-api/v1/login")
|
||||
get_response = await client_with_oauth_auth.get("/user-api/v1/login")
|
||||
assert get_response.ok
|
||||
oauth.get_oauth_url.assert_called_once_with()
|
||||
|
||||
|
||||
async def test_get_redirect_to_oauth_empty_code(client_with_auth: TestClient) -> None:
|
||||
async def test_get_redirect_to_oauth_empty_code(client_with_oauth_auth: TestClient) -> None:
|
||||
"""
|
||||
must redirect to OAuth service provider in case if empty code is supplied
|
||||
"""
|
||||
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
||||
oauth = client_with_oauth_auth.app["validator"]
|
||||
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
||||
|
||||
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": ""})
|
||||
get_response = await client_with_oauth_auth.get("/user-api/v1/login", params={"code": ""})
|
||||
assert get_response.ok
|
||||
oauth.get_oauth_url.assert_called_once_with()
|
||||
|
||||
|
||||
async def test_get(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
async def test_get(client_with_oauth_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must login user correctly from OAuth
|
||||
"""
|
||||
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
||||
oauth = client_with_oauth_auth.app["validator"]
|
||||
oauth.get_oauth_username.return_value = "user"
|
||||
oauth.known_username.return_value = True
|
||||
oauth.enabled = False # lol
|
||||
oauth.max_age = 60
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
|
||||
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": "code"})
|
||||
get_response = await client_with_oauth_auth.get("/user-api/v1/login", params={"code": "code"})
|
||||
|
||||
assert get_response.ok
|
||||
oauth.get_oauth_username.assert_called_once_with("code")
|
||||
@ -71,16 +71,16 @@ async def test_get(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
pytest.helpers.anyvar(int), pytest.helpers.anyvar(int), pytest.helpers.anyvar(int))
|
||||
|
||||
|
||||
async def test_get_unauthorized(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
async def test_get_unauthorized(client_with_oauth_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return unauthorized from OAuth
|
||||
"""
|
||||
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
||||
oauth = client_with_oauth_auth.app["validator"]
|
||||
oauth.known_username.return_value = False
|
||||
oauth.max_age = 60
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
|
||||
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": "code"})
|
||||
get_response = await client_with_oauth_auth.get("/user-api/v1/login", params={"code": "code"})
|
||||
|
||||
assert get_response.status == 401
|
||||
remember_mock.assert_not_called()
|
||||
|
@ -1,6 +1,7 @@
|
||||
[settings]
|
||||
include = .
|
||||
logging = logging.ini
|
||||
database = ../../../ahriman-test.db
|
||||
|
||||
[alpm]
|
||||
aur_url = https://aur.archlinux.org
|
||||
|
Reference in New Issue
Block a user