port part of settings to database (#54)

This commit is contained in:
2022-03-31 01:48:06 +03:00
committed by GitHub
parent d4eadf0013
commit 83931f5cf4
117 changed files with 2768 additions and 1044 deletions

View File

@ -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)

View File

@ -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,

View File

@ -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()

View File

@ -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)

View File

@ -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)

View File

@ -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)

View File

@ -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)

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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)

View File

@ -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)

View File

@ -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

View File

@ -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)

View File

@ -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")

View File

@ -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

View 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()

View 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()

View 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()

View 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()

View 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()

View 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)

View File

@ -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

View 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

View File

@ -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

View File

@ -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]

View 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()

View File

@ -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

View File

@ -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"

View 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)

View File

@ -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)

View 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

View File

@ -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)

View File

@ -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)

View File

@ -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:

View File

@ -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)

View File

@ -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() == []

View File

@ -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)

View File

@ -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

View File

@ -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:

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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

View 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()

View 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

View File

@ -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

View File

@ -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:

View File

@ -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)

View File

@ -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

View File

@ -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

View File

@ -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))

View File

@ -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()

View File

@ -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

View File

@ -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

View File

@ -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()

View File

@ -1,6 +1,7 @@
[settings]
include = .
logging = logging.ini
database = ../../../ahriman-test.db
[alpm]
aur_url = https://aur.archlinux.org