arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-06-28 06:41:43 +00:00
Code Issues Projects Releases Wiki Activity

Remote call trigger support (#105)

Browse Source 
* add support of remote task tracking
* add remote call trigger implementation
* docs update
* add cross-service upload
* add notes about user
* add more ability to control upload
* multipart upload with signatures as well as safe file save
* configuration reference update
* rename watcher methods
* erase logs based on current package version

Old implementation has used process id instead, but it leads to log
removal in case of remote process trigger

* add --server flag for setup command
* restore behavior of the httploghandler
This commit is contained in:
Evgenii Alekseev
2023-08-20 03:44:31 +03:00
committed by GitHub
parent 5b172ad20b
commit ad1c0051c4
111 changed files with 2774 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/ahriman/application/conftest.py
View File

@ -39,7 +39,7 @@ def args() -> argparse.Namespace:
Returns:
argparse.Namespace: command line arguments test instance
"""
return argparse.Namespace(architecture=None, lock=None, force=False, unsafe=False, report=False)
return argparse.Namespace(architecture=None, lock=None, force=False, unsafe=False, report=False, wait_timeout=-1)
@pytest.fixture

39
tests/ahriman/application/handlers/test_handler_setup.py
View File

@ -32,6 +32,7 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
args.multilib = True
args.packager = "John Doe <john@doe.com>"
args.repository = "aur-clone"
args.server = None
args.sign_key = "key"
args.sign_target = [SignSettings.Packages]
args.web_port = 8080
@ -57,13 +58,34 @@ def test_run(args: argparse.Namespace, configuration: Configuration, repository:
ahriman_configuration_mock.assert_called_once_with(args, "x86_64", args.repository, configuration)
devtools_configuration_mock.assert_called_once_with(
args.build_command, "x86_64", args.from_configuration, args.mirror, args.multilib, args.repository,
repository_paths)
f"file://{repository_paths.repository}")
makepkg_configuration_mock.assert_called_once_with(args.packager, args.makeflags_jobs, repository_paths)
sudo_configuration_mock.assert_called_once_with(repository_paths, args.build_command, "x86_64")
executable_mock.assert_called_once_with(repository_paths, args.build_command, "x86_64")
init_mock.assert_called_once_with()
def test_run_with_server(args: argparse.Namespace, configuration: Configuration, repository: Repository,
mocker: MockerFixture) -> None:
"""
must run command with server specified
"""
args = _default_args(args)
args.server = "server"
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
mocker.patch("ahriman.application.handlers.Setup.configuration_create_ahriman")
mocker.patch("ahriman.application.handlers.Setup.configuration_create_makepkg")
mocker.patch("ahriman.application.handlers.Setup.configuration_create_sudo")
mocker.patch("ahriman.application.handlers.Setup.executable_create")
mocker.patch("ahriman.core.alpm.repo.Repo.init")
devtools_configuration_mock = mocker.patch("ahriman.application.handlers.Setup.configuration_create_devtools")
Setup.run(args, "x86_64", configuration, report=False)
devtools_configuration_mock.assert_called_once_with(
args.build_command, "x86_64", args.from_configuration, args.mirror, args.multilib, args.repository,
"server")
def test_build_command(args: argparse.Namespace) -> None:
"""
must generate correct build command name
@ -120,8 +142,7 @@ def test_configuration_create_ahriman_no_multilib(args: argparse.Namespace, conf
]) # non-strict check called intentionally
def test_configuration_create_devtools(args: argparse.Namespace, repository_paths: RepositoryPaths,
mocker: MockerFixture) -> None:
def test_configuration_create_devtools(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must create configuration for the devtools
"""
@ -132,13 +153,12 @@ def test_configuration_create_devtools(args: argparse.Namespace, repository_path
write_mock = mocker.patch("ahriman.core.configuration.Configuration.write")
Setup.configuration_create_devtools(args.build_command, "x86_64", args.from_configuration,
None, args.multilib, args.repository, repository_paths)
None, args.multilib, args.repository, "server")
add_section_mock.assert_has_calls([MockCall("multilib"), MockCall(args.repository)])
write_mock.assert_called_once_with(pytest.helpers.anyvar(int))
def test_configuration_create_devtools_mirror(args: argparse.Namespace, repository_paths: RepositoryPaths,
mocker: MockerFixture) -> None:
def test_configuration_create_devtools_mirror(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must create configuration for the devtools with mirror set explicitly
"""
@ -157,14 +177,13 @@ def test_configuration_create_devtools_mirror(args: argparse.Namespace, reposito
set_option_mock = mocker.patch("ahriman.core.configuration.Configuration.set_option")
Setup.configuration_create_devtools(args.build_command, "x86_64", args.from_configuration,
args.mirror, False, args.repository, repository_paths)
args.mirror, False, args.repository, "server")
get_mock.assert_has_calls([MockCall("core", "Include", fallback=None), MockCall("extra", "Include", fallback=None)])
remove_option_mock.assert_called_once_with("core", "Include")
set_option_mock.assert_has_calls([MockCall("core", "Server", args.mirror)]) # non-strict check called intentionally
def test_configuration_create_devtools_no_multilib(args: argparse.Namespace, repository_paths: RepositoryPaths,
mocker: MockerFixture) -> None:
def test_configuration_create_devtools_no_multilib(args: argparse.Namespace, mocker: MockerFixture) -> None:
"""
must create configuration for the devtools without multilib
"""
@ -174,7 +193,7 @@ def test_configuration_create_devtools_no_multilib(args: argparse.Namespace, rep
write_mock = mocker.patch("ahriman.core.configuration.Configuration.write")
Setup.configuration_create_devtools(args.build_command, "x86_64", args.from_configuration,
None, False, args.repository, repository_paths)
None, False, args.repository, "server")
write_mock.assert_called_once_with(pytest.helpers.anyvar(int))

14
tests/ahriman/application/handlers/test_handler_status.py
View File

@ -36,8 +36,8 @@ def test_run(args: argparse.Namespace, configuration: Configuration, repository:
"""
args = _default_args(args)
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
application_mock = mocker.patch("ahriman.core.status.client.Client.get_internal")
packages_mock = mocker.patch("ahriman.core.status.client.Client.get",
application_mock = mocker.patch("ahriman.core.status.client.Client.status_get")
packages_mock = mocker.patch("ahriman.core.status.client.Client.package_get",
return_value=[(package_ahriman, BuildStatus(BuildStatusEnum.Success)),
(package_python_schedule, BuildStatus(BuildStatusEnum.Failed))])
check_mock = mocker.patch("ahriman.application.handlers.Handler.check_if_empty")
@ -58,8 +58,8 @@ def test_run_empty_exception(args: argparse.Namespace, configuration: Configurat
args = _default_args(args)
args.exit_code = True
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
mocker.patch("ahriman.core.status.client.Client.get_internal")
mocker.patch("ahriman.core.status.client.Client.get", return_value=[])
mocker.patch("ahriman.core.status.client.Client.status_get")
mocker.patch("ahriman.core.status.client.Client.package_get", return_value=[])
check_mock = mocker.patch("ahriman.application.handlers.Handler.check_if_empty")
Status.run(args, "x86_64", configuration, report=False)
@ -74,7 +74,7 @@ def test_run_verbose(args: argparse.Namespace, configuration: Configuration, rep
args = _default_args(args)
args.info = True
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
mocker.patch("ahriman.core.status.client.Client.get",
mocker.patch("ahriman.core.status.client.Client.package_get",
return_value=[(package_ahriman, BuildStatus(BuildStatusEnum.Success))])
print_mock = mocker.patch("ahriman.core.formatters.Printer.print")
@ -90,7 +90,7 @@ def test_run_with_package_filter(args: argparse.Namespace, configuration: Config
args = _default_args(args)
args.package = [package_ahriman.base]
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
packages_mock = mocker.patch("ahriman.core.status.client.Client.get",
packages_mock = mocker.patch("ahriman.core.status.client.Client.package_get",
return_value=[(package_ahriman, BuildStatus(BuildStatusEnum.Success))])
Status.run(args, "x86_64", configuration, report=False)
@ -104,7 +104,7 @@ def test_run_by_status(args: argparse.Namespace, configuration: Configuration, r
"""
args = _default_args(args)
args.status = BuildStatusEnum.Failed
mocker.patch("ahriman.core.status.client.Client.get",
mocker.patch("ahriman.core.status.client.Client.package_get",
return_value=[(package_ahriman, BuildStatus(BuildStatusEnum.Success)),
(package_python_schedule, BuildStatus(BuildStatusEnum.Failed))])
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)

6
tests/ahriman/application/handlers/test_handler_status_update.py
View File

@ -34,7 +34,7 @@ def test_run(args: argparse.Namespace, configuration: Configuration, repository:
"""
args = _default_args(args)
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
update_self_mock = mocker.patch("ahriman.core.status.client.Client.update_self")
update_self_mock = mocker.patch("ahriman.core.status.client.Client.status_update")
StatusUpdate.run(args, "x86_64", configuration, report=False)
update_self_mock.assert_called_once_with(args.status)
@ -48,7 +48,7 @@ def test_run_packages(args: argparse.Namespace, configuration: Configuration, re
args = _default_args(args)
args.package = [package_ahriman.base]
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
update_mock = mocker.patch("ahriman.core.status.client.Client.package_update")
StatusUpdate.run(args, "x86_64", configuration, report=False)
update_mock.assert_called_once_with(package_ahriman.base, args.status)
@ -63,7 +63,7 @@ def test_run_remove(args: argparse.Namespace, configuration: Configuration, repo
args.package = [package_ahriman.base]
args.action = Action.Remove
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
update_mock = mocker.patch("ahriman.core.status.client.Client.remove")
update_mock = mocker.patch("ahriman.core.status.client.Client.package_remove")
StatusUpdate.run(args, "x86_64", configuration, report=False)
update_mock.assert_called_once_with(package_ahriman.base)

2
tests/ahriman/application/handlers/test_handler_validate.py
View File

@ -67,8 +67,10 @@ def test_schema(configuration: Configuration) -> None:
assert schema.pop("keyring-generator")
assert schema.pop("mirrorlist")
assert schema.pop("mirrorlist-generator")
assert schema.pop("remote-call")
assert schema.pop("remote-pull")
assert schema.pop("remote-push")
assert schema.pop("remote-service")
assert schema.pop("report")
assert schema.pop("rsync")
assert schema.pop("s3")

5
tests/ahriman/application/handlers/test_handler_web.py
View File

@ -77,6 +77,10 @@ def test_extract_arguments(args: argparse.Namespace, configuration: Configuratio
expected.extend(["--unsafe"])
assert list(Web.extract_arguments(probe, "x86_64", configuration)) == expected
configuration.set_option("web", "wait_timeout", "60")
expected.extend(["--wait-timeout", "60"])
assert list(Web.extract_arguments(probe, "x86_64", configuration)) == expected
def test_extract_arguments_full(parser: argparse.ArgumentParser, configuration: Configuration):
"""
@ -91,6 +95,7 @@ def test_extract_arguments_full(parser: argparse.ArgumentParser, configuration:
value = action.const or \
next(iter(action.choices or []), None) or \
(not action.default if isinstance(action.default, bool) else None) or \
(42 if action.type == int else None) or \
"random string"
if action.type is not None:
value = action.type(value)

10
tests/ahriman/application/test_ahriman.py
View File

@ -47,6 +47,16 @@ def test_parser_option_log_handler(parser: argparse.ArgumentParser) -> None:
assert isinstance(args.log_handler, LogHandler)
def test_parser_option_wait_timeout(parser: argparse.ArgumentParser) -> None:
"""
must convert wait-timeout option to int instance
"""
args = parser.parse_args(["service-config"])
assert isinstance(args.wait_timeout, int)
args = parser.parse_args(["--wait-timeout", "60", "service-config"])
assert isinstance(args.wait_timeout, int)
def test_multiple_architectures(parser: argparse.ArgumentParser) -> None:
"""
must accept multiple architectures

29
tests/ahriman/application/test_lock.py
View File

@ -32,7 +32,7 @@ def test_check_version(lock: Lock, mocker: MockerFixture) -> None:
"""
must check version correctly
"""
mocker.patch("ahriman.core.status.client.Client.get_internal",
mocker.patch("ahriman.core.status.client.Client.status_get",
return_value=InternalStatus(status=BuildStatus(), version=__version__))
logging_mock = mocker.patch("logging.Logger.warning")
@ -44,7 +44,7 @@ def test_check_version_mismatch(lock: Lock, mocker: MockerFixture) -> None:
"""
must check mismatched version correctly
"""
mocker.patch("ahriman.core.status.client.Client.get_internal",
mocker.patch("ahriman.core.status.client.Client.status_get",
return_value=InternalStatus(status=BuildStatus(), version="version"))
logging_mock = mocker.patch("logging.Logger.warning")
@ -154,15 +154,35 @@ def test_create_unsafe(lock: Lock) -> None:
lock.path.unlink()
def test_watch(lock: Lock, mocker: MockerFixture) -> None:
"""
must check if lock file exists
"""
wait_mock = mocker.patch("ahriman.models.waiter.Waiter.wait")
lock.path = Path(tempfile.mktemp()) # nosec
lock.watch()
wait_mock.assert_called_once_with(lock.path.is_file)
def test_watch_skip(lock: Lock, mocker: MockerFixture) -> None:
"""
must skip watch on empty path
"""
mocker.patch("pathlib.Path.is_file", return_value=True)
lock.watch()
def test_enter(lock: Lock, mocker: MockerFixture) -> None:
"""
must process with context manager
"""
check_user_mock = mocker.patch("ahriman.application.lock.Lock.check_user")
check_version_mock = mocker.patch("ahriman.application.lock.Lock.check_version")
watch_mock = mocker.patch("ahriman.application.lock.Lock.watch")
clear_mock = mocker.patch("ahriman.application.lock.Lock.clear")
create_mock = mocker.patch("ahriman.application.lock.Lock.create")
update_status_mock = mocker.patch("ahriman.core.status.client.Client.update_self")
update_status_mock = mocker.patch("ahriman.core.status.client.Client.status_update")
with lock:
pass
@ -170,6 +190,7 @@ def test_enter(lock: Lock, mocker: MockerFixture) -> None:
clear_mock.assert_called_once_with()
create_mock.assert_called_once_with()
check_version_mock.assert_called_once_with()
watch_mock.assert_called_once_with()
update_status_mock.assert_has_calls([MockCall(BuildStatusEnum.Building), MockCall(BuildStatusEnum.Success)])
@ -180,7 +201,7 @@ def test_exit_with_exception(lock: Lock, mocker: MockerFixture) -> None:
mocker.patch("ahriman.application.lock.Lock.check_user")
mocker.patch("ahriman.application.lock.Lock.clear")
mocker.patch("ahriman.application.lock.Lock.create")
update_status_mock = mocker.patch("ahriman.core.status.client.Client.update_self")
update_status_mock = mocker.patch("ahriman.core.status.client.Client.status_update")
with pytest.raises(Exception):
with lock:

2
tests/ahriman/core/database/migrations/test_m009_local_source.py
View File

@ -1,7 +1,7 @@
from ahriman.core.database.migrations.m009_local_source import steps
def test_migration_packagers() -> None:
def test_migration_local_source() -> None:
"""
migration must not be empty
"""

8
tests/ahriman/core/database/migrations/test_m010_version_based_logs_removal.py Normal file
View File

@ -0,0 +1,8 @@
from ahriman.core.database.migrations.m010_version_based_logs_removal import steps
def test_migration_version_based_logs_removal() -> None:
"""
migration must not be empty
"""
assert steps

18
tests/ahriman/core/database/operations/test_logs_operations.py
View File

@ -8,11 +8,11 @@ def test_logs_insert_remove_process(database: SQLite, package_ahriman: Package,
"""
must clear process specific package logs
"""
database.logs_insert(LogRecordId(package_ahriman.base, 1), 42.0, "message 1")
database.logs_insert(LogRecordId(package_ahriman.base, 2), 43.0, "message 2")
database.logs_insert(LogRecordId(package_python_schedule.base, 1), 42.0, "message 3")
database.logs_insert(LogRecordId(package_ahriman.base, "1"), 42.0, "message 1")
database.logs_insert(LogRecordId(package_ahriman.base, "2"), 43.0, "message 2")
database.logs_insert(LogRecordId(package_python_schedule.base, "1"), 42.0, "message 3")
database.logs_remove(package_ahriman.base, 1)
database.logs_remove(package_ahriman.base, "1")
assert database.logs_get(package_ahriman.base) == "[1970-01-01 00:00:42] message 1"
assert database.logs_get(package_python_schedule.base) == "[1970-01-01 00:00:42] message 3"
@ -21,9 +21,9 @@ def test_logs_insert_remove_full(database: SQLite, package_ahriman: Package, pac
"""
must clear full package logs
"""
database.logs_insert(LogRecordId(package_ahriman.base, 1), 42.0, "message 1")
database.logs_insert(LogRecordId(package_ahriman.base, 2), 43.0, "message 2")
database.logs_insert(LogRecordId(package_python_schedule.base, 1), 42.0, "message 3")
database.logs_insert(LogRecordId(package_ahriman.base, "1"), 42.0, "message 1")
database.logs_insert(LogRecordId(package_ahriman.base, "2"), 43.0, "message 2")
database.logs_insert(LogRecordId(package_python_schedule.base, "1"), 42.0, "message 3")
database.logs_remove(package_ahriman.base, None)
assert not database.logs_get(package_ahriman.base)
@ -34,6 +34,6 @@ def test_logs_insert_get(database: SQLite, package_ahriman: Package) -> None:
"""
must insert and get package logs
"""
database.logs_insert(LogRecordId(package_ahriman.base, 1), 43.0, "message 2")
database.logs_insert(LogRecordId(package_ahriman.base, 1), 42.0, "message 1")
database.logs_insert(LogRecordId(package_ahriman.base, "1"), 43.0, "message 2")
database.logs_insert(LogRecordId(package_ahriman.base, "1"), 42.0, "message 1")
assert database.logs_get(package_ahriman.base) == "[1970-01-01 00:00:42] message 1\n[1970-01-01 00:00:43] message 2"

27
tests/ahriman/core/log/test_filtered_access_logger.py
View File

@ -43,6 +43,33 @@ def test_is_logs_post() -> None:
assert not FilteredAccessLogger.is_logs_post(request)
def test_is_process_get() -> None:
"""
must correctly define if request belongs to process get
"""
request = MagicMock()
request.method = "GET"
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00"
assert FilteredAccessLogger.is_process_get(request)
request.method = "POST"
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00"
assert not FilteredAccessLogger.is_process_get(request)
request.method = "GET"
request.path = "/api/v1/service/process/e7d67119-264a-48f4-b7e4-07bc96a7de00/some/random/path"
assert not FilteredAccessLogger.is_process_get(request)
request.method = "GET"
request.path = "/api/v1/service/process"
assert not FilteredAccessLogger.is_process_get(request)
request.method = "GET"
request.path = "/api/v1/service/process/"
assert not FilteredAccessLogger.is_process_get(request)
def test_log(filtered_access_logger: FilteredAccessLogger, mocker: MockerFixture) -> None:
"""
must emit log record

17
tests/ahriman/core/log/test_http_log_handler.py
View File

@ -4,6 +4,7 @@ from pytest_mock import MockerFixture
from ahriman.core.configuration import Configuration
from ahriman.core.log.http_log_handler import HttpLogHandler
from ahriman.models.log_record_id import LogRecordId
from ahriman.models.package import Package
@ -39,13 +40,13 @@ def test_emit(configuration: Configuration, log_record: logging.LogRecord, packa
"""
must emit log record to reporter
"""
log_record.package_base = package_ahriman.base
log_mock = mocker.patch("ahriman.core.status.client.Client.logs")
log_record_id = log_record.package_id = LogRecordId(package_ahriman.base, package_ahriman.version)
log_mock = mocker.patch("ahriman.core.status.client.Client.package_logs")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
handler.emit(log_record)
log_mock.assert_called_once_with(package_ahriman.base, log_record)
log_mock.assert_called_once_with(log_record_id, log_record)
def test_emit_failed(configuration: Configuration, log_record: logging.LogRecord, package_ahriman: Package,
@ -53,8 +54,8 @@ def test_emit_failed(configuration: Configuration, log_record: logging.LogRecord
"""
must call handle error on exception
"""
log_record.package_base = package_ahriman.base
mocker.patch("ahriman.core.status.client.Client.logs", side_effect=Exception())
log_record.package_id = LogRecordId(package_ahriman.base, package_ahriman.version)
mocker.patch("ahriman.core.status.client.Client.package_logs", side_effect=Exception())
handle_error_mock = mocker.patch("logging.Handler.handleError")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
@ -67,8 +68,8 @@ def test_emit_suppress_failed(configuration: Configuration, log_record: logging.
"""
must not call handle error on exception if suppress flag is set
"""
log_record.package_base = package_ahriman.base
mocker.patch("ahriman.core.status.client.Client.logs", side_effect=Exception())
log_record.package_id = LogRecordId(package_ahriman.base, package_ahriman.version)
mocker.patch("ahriman.core.status.client.Client.package_logs", side_effect=Exception())
handle_error_mock = mocker.patch("logging.Handler.handleError")
handler = HttpLogHandler(configuration, report=False, suppress_errors=True)
@ -80,7 +81,7 @@ def test_emit_skip(configuration: Configuration, log_record: logging.LogRecord,
"""
must skip log record posting if no package base set
"""
log_mock = mocker.patch("ahriman.core.status.client.Client.logs")
log_mock = mocker.patch("ahriman.core.status.client.Client.package_logs")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
handler.emit(log_record)

29
tests/ahriman/core/log/test_lazy_logging.py
View File

@ -5,6 +5,7 @@ from pytest_mock import MockerFixture
from ahriman.core.alpm.repo import Repo
from ahriman.core.database import SQLite
from ahriman.models.log_record_id import LogRecordId
from ahriman.models.package import Package
@ -20,16 +21,16 @@ def test_package_logger_set_reset(database: SQLite) -> None:
"""
must set and reset package base attribute
"""
package_base = "package base"
log_record_id = LogRecordId("base", "version")
database._package_logger_set(package_base)
database._package_logger_set(log_record_id.package_base, log_record_id.version)
record = logging.makeLogRecord({})
assert record.package_base == package_base
assert record.package_id == log_record_id
database._package_logger_reset()
record = logging.makeLogRecord({})
with pytest.raises(AttributeError):
record.package_base
record.package_id
def test_in_package_context(database: SQLite, package_ahriman: Package, mocker: MockerFixture) -> None:
@ -39,10 +40,24 @@ def test_in_package_context(database: SQLite, package_ahriman: Package, mocker:
set_mock = mocker.patch("ahriman.core.log.LazyLogging._package_logger_set")
reset_mock = mocker.patch("ahriman.core.log.LazyLogging._package_logger_reset")
with database.in_package_context(package_ahriman.base):
with database.in_package_context(package_ahriman.base, package_ahriman.version):
pass
set_mock.assert_called_once_with(package_ahriman.base)
set_mock.assert_called_once_with(package_ahriman.base, package_ahriman.version)
reset_mock.assert_called_once_with()
def test_in_package_context_empty_version(database: SQLite, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must set package log context
"""
set_mock = mocker.patch("ahriman.core.log.LazyLogging._package_logger_set")
reset_mock = mocker.patch("ahriman.core.log.LazyLogging._package_logger_reset")
with database.in_package_context(package_ahriman.base, None):
pass
set_mock.assert_called_once_with(package_ahriman.base, None)
reset_mock.assert_called_once_with()
@ -54,7 +69,7 @@ def test_in_package_context_failed(database: SQLite, package_ahriman: Package, m
reset_mock = mocker.patch("ahriman.core.log.LazyLogging._package_logger_reset")
with pytest.raises(Exception):
with database.in_package_context(package_ahriman.base):
with database.in_package_context(package_ahriman.base, ""):
raise Exception()
reset_mock.assert_called_once_with()

20
tests/ahriman/core/report/conftest.py Normal file
View File

@ -0,0 +1,20 @@
import pytest
from ahriman.core.configuration import Configuration
from ahriman.core.report.remote_call import RemoteCall
@pytest.fixture
def remote_call(configuration: Configuration) -> RemoteCall:
"""
fixture for remote update trigger
Args:
configuration(Configuration): configuration fixture
Returns:
RemoteCall: remote update trigger test instance
"""
configuration.set_option("web", "host", "localhost")
configuration.set_option("web", "port", "8080")
return RemoteCall("x86_64", configuration, "remote-call")

95
tests/ahriman/core/report/test_remote_call.py Normal file
View File

@ -0,0 +1,95 @@
import pytest
import requests
from pytest_mock import MockerFixture
from ahriman.core.report.remote_call import RemoteCall
from ahriman.models.result import Result
def test_generate(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must correctly call client
"""
update_mock = mocker.patch("ahriman.core.report.remote_call.RemoteCall.remote_update", return_value="id")
wait_mock = mocker.patch("ahriman.core.report.remote_call.RemoteCall.remote_wait")
remote_call.generate([], Result())
update_mock.assert_called_once_with()
wait_mock.assert_called_once_with("id")
def test_is_process_alive(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must correctly define if process is alive
"""
response_obj = requests.Response()
response_obj._content = """{"is_alive": true}""".encode("utf8")
response_obj.status_code = 200
request_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request", return_value=response_obj)
assert remote_call.is_process_alive("id")
request_mock.assert_called_once_with("GET", "/api/v1/service/process/id")
def test_is_process_alive_unknown(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must correctly define if process is unknown
"""
response = requests.Response()
response.status_code = 404
mocker.patch("ahriman.core.status.web_client.WebClient.make_request",
side_effect=requests.RequestException(response=response))
assert not remote_call.is_process_alive("id")
def test_is_process_alive_error(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must reraise exception on process request
"""
mocker.patch("ahriman.core.status.web_client.WebClient.make_request", side_effect=Exception)
with pytest.raises(Exception):
remote_call.is_process_alive("id")
def test_is_process_alive_http_error(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must reraise http exception on process request
"""
response = requests.Response()
response.status_code = 500
mocker.patch("ahriman.core.status.web_client.WebClient.make_request",
side_effect=requests.RequestException(response=response))
with pytest.raises(requests.RequestException):
remote_call.is_process_alive("id")
def test_remote_update(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must call remote server for update process
"""
response_obj = requests.Response()
response_obj._content = """{"process_id": "id"}""".encode("utf8")
response_obj.status_code = 200
request_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request", return_value=response_obj)
assert remote_call.remote_update() == "id"
request_mock.assert_called_once_with("POST", "/api/v1/service/update", json={
"aur": False,
"local": False,
"manual": True,
})
def test_remote_wait(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must wait for remote process to success
"""
wait_mock = mocker.patch("ahriman.models.waiter.Waiter.wait")
remote_call.remote_wait("id")
wait_mock.assert_called_once_with(pytest.helpers.anyvar(int), "id")

13
tests/ahriman/core/report/test_report.py
View File

@ -24,6 +24,7 @@ def test_report_dummy(configuration: Configuration, result: Result, mocker: Mock
"""
mocker.patch("ahriman.models.report_settings.ReportSettings.from_option", return_value=ReportSettings.Disabled)
report_mock = mocker.patch("ahriman.core.report.report.Report.generate")
Report.load("x86_64", configuration, "disabled").run(result, [])
report_mock.assert_called_once_with([], result)
@ -55,6 +56,18 @@ def test_report_html(configuration: Configuration, result: Result, mocker: Mocke
report_mock.assert_called_once_with([], result)
def test_report_remote_call(configuration: Configuration, result: Result, mocker: MockerFixture) -> None:
"""
must instantiate remote call trigger
"""
configuration.set_option("web", "host", "localhost")
configuration.set_option("web", "port", "8080")
report_mock = mocker.patch("ahriman.core.report.remote_call.RemoteCall.generate")
Report.load("x86_64", configuration, "remote-call").run(result, [])
report_mock.assert_called_once_with([], result)
def test_report_telegram(configuration: Configuration, result: Result, mocker: MockerFixture) -> None:
"""
must generate telegram report

8
tests/ahriman/core/repository/test_executor.py
View File

@ -85,7 +85,7 @@ def test_process_remove_base(executor: Executor, package_ahriman: Package, mocke
build_queue_mock = mocker.patch("ahriman.core.database.SQLite.build_queue_clear")
patches_mock = mocker.patch("ahriman.core.database.SQLite.patches_remove")
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.package_remove")
executor.process_remove([package_ahriman.base])
# must remove via alpm wrapper
@ -106,7 +106,7 @@ def test_process_remove_base_multiple(executor: Executor, package_python_schedul
"""
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_python_schedule])
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.package_remove")
executor.process_remove([package_python_schedule.base])
# must remove via alpm wrapper
@ -125,7 +125,7 @@ def test_process_remove_base_single(executor: Executor, package_python_schedule:
"""
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[package_python_schedule])
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.package_remove")
executor.process_remove(["python2-schedule"])
# must remove via alpm wrapper
@ -171,7 +171,7 @@ def test_process_remove_unknown(executor: Executor, package_ahriman: Package, mo
"""
mocker.patch("ahriman.core.repository.executor.Executor.packages", return_value=[])
repo_remove_mock = mocker.patch("ahriman.core.alpm.repo.Repo.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.remove")
status_client_mock = mocker.patch("ahriman.core.status.client.Client.package_remove")
executor.process_remove([package_ahriman.base])
repo_remove_mock.assert_not_called()

19
tests/ahriman/core/sign/test_gpg.py
View File

@ -76,6 +76,13 @@ def test_sign_options(configuration: Configuration) -> None:
assert default_key == "default-key"
def test_signature() -> None:
"""
must correctly generate the signature path
"""
assert GPG.signature(Path("path") / "to" / "package.tar.xz") == Path("path") / "to" / "package.tar.xz.sig"
def test_key_download(gpg: GPG, mocker: MockerFixture) -> None:
"""
must download the key from public server
@ -222,6 +229,18 @@ def test_process_sign_package_skip_4(gpg: GPG, mocker: MockerFixture) -> None:
process_mock.assert_not_called()
def test_process_sign_package_skip_already_signed(gpg_with_key: GPG, mocker: MockerFixture) -> None:
"""
must not sign package if it was already signed
"""
result = [Path("a"), Path("a.sig")]
mocker.patch("pathlib.Path.is_file", return_value=True)
process_mock = mocker.patch("ahriman.core.sign.gpg.GPG.process")
assert gpg_with_key.process_sign_package(Path("a"), gpg_with_key.default_key) == result
process_mock.assert_not_called()
def test_process_sign_repository_1(gpg_with_key: GPG, mocker: MockerFixture) -> None:
"""
must sign repository

67
tests/ahriman/core/status/test_client.py
View File

@ -7,6 +7,7 @@ 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.internal_status import InternalStatus
from ahriman.models.log_record_id import LogRecordId
from ahriman.models.package import Package
@ -51,64 +52,47 @@ def test_load_full_client_from_unix_socket(configuration: Configuration) -> None
assert isinstance(Client.load(configuration, report=True), WebClient)
def test_add(client: Client, package_ahriman: Package) -> None:
def test_package_add(client: Client, package_ahriman: Package) -> None:
"""
must process package addition without errors
"""
client.add(package_ahriman, BuildStatusEnum.Unknown)
client.package_add(package_ahriman, BuildStatusEnum.Unknown)
def test_get(client: Client, package_ahriman: Package) -> None:
def test_package_get(client: Client, package_ahriman: Package) -> None:
"""
must return empty package list
"""
assert client.get(package_ahriman.base) == []
assert client.get(None) == []
assert client.package_get(package_ahriman.base) == []
assert client.package_get(None) == []
def test_get_internal(client: Client) -> None:
"""
must return dummy status for web service
"""
actual = client.get_internal()
expected = InternalStatus(status=BuildStatus(timestamp=actual.status.timestamp))
assert actual == expected
def test_log(client: Client, package_ahriman: Package, log_record: logging.LogRecord) -> None:
def test_package_logs(client: Client, package_ahriman: Package, log_record: logging.LogRecord) -> None:
"""
must process log record without errors
"""
client.logs(package_ahriman.base, log_record)
client.package_logs(LogRecordId(package_ahriman.base, package_ahriman.version), log_record)
def test_remove(client: Client, package_ahriman: Package) -> None:
def test_package_remove(client: Client, package_ahriman: Package) -> None:
"""
must process remove without errors
"""
client.remove(package_ahriman.base)
client.package_remove(package_ahriman.base)
def test_update(client: Client, package_ahriman: Package) -> None:
def test_package_update(client: Client, package_ahriman: Package) -> None:
"""
must update package status without errors
"""
client.update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_update_self(client: Client) -> None:
"""
must update self status without errors
"""
client.update_self(BuildStatusEnum.Unknown)
client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_set_building(client: Client, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must set building status to the package
"""
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
update_mock = mocker.patch("ahriman.core.status.client.Client.package_update")
client.set_building(package_ahriman.base)
update_mock.assert_called_once_with(package_ahriman.base, BuildStatusEnum.Building)
@ -118,7 +102,7 @@ def test_set_failed(client: Client, package_ahriman: Package, mocker: MockerFixt
"""
must set failed status to the package
"""
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
update_mock = mocker.patch("ahriman.core.status.client.Client.package_update")
client.set_failed(package_ahriman.base)
update_mock.assert_called_once_with(package_ahriman.base, BuildStatusEnum.Failed)
@ -128,7 +112,7 @@ def test_set_pending(client: Client, package_ahriman: Package, mocker: MockerFix
"""
must set building status to the package
"""
update_mock = mocker.patch("ahriman.core.status.client.Client.update")
update_mock = mocker.patch("ahriman.core.status.client.Client.package_update")
client.set_pending(package_ahriman.base)
update_mock.assert_called_once_with(package_ahriman.base, BuildStatusEnum.Pending)
@ -138,7 +122,7 @@ def test_set_success(client: Client, package_ahriman: Package, mocker: MockerFix
"""
must set success status to the package
"""
add_mock = mocker.patch("ahriman.core.status.client.Client.add")
add_mock = mocker.patch("ahriman.core.status.client.Client.package_add")
client.set_success(package_ahriman)
add_mock.assert_called_once_with(package_ahriman, BuildStatusEnum.Success)
@ -148,7 +132,24 @@ def test_set_unknown(client: Client, package_ahriman: Package, mocker: MockerFix
"""
must add new package with unknown status
"""
add_mock = mocker.patch("ahriman.core.status.client.Client.add")
add_mock = mocker.patch("ahriman.core.status.client.Client.package_add")
client.set_unknown(package_ahriman)
add_mock.assert_called_once_with(package_ahriman, BuildStatusEnum.Unknown)
def test_status_get(client: Client) -> None:
"""
must return dummy status for web service
"""
actual = client.status_get()
expected = InternalStatus(status=BuildStatus(timestamp=actual.status.timestamp))
assert actual == expected
def test_status_update(client: Client) -> None:
"""
must update self status without errors
"""
client.status_update(BuildStatusEnum.Unknown)

162
tests/ahriman/core/status/test_watcher.py
View File

@ -22,33 +22,6 @@ def test_force_no_report(configuration: Configuration, database: SQLite, mocker:
load_mock.assert_called_once_with("x86_64", configuration, database, report=False)
def test_get(watcher: Watcher, package_ahriman: Package) -> None:
"""
must return package status
"""
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
package, status = watcher.get(package_ahriman.base)
assert package == package_ahriman
assert status.status == BuildStatusEnum.Unknown
def test_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
"""
must fail on unknown package
"""
with pytest.raises(UnknownPackageError):
watcher.get(package_ahriman.base)
def test_get_logs(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return package logs
"""
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_get")
watcher.get_logs(package_ahriman.base)
logs_mock.assert_called_once_with(package_ahriman.base)
def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must correctly load packages
@ -77,109 +50,136 @@ def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFi
assert status.status == BuildStatusEnum.Success
def test_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_logs_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return package logs
"""
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_get")
watcher.logs_get(package_ahriman.base)
logs_mock.assert_called_once_with(package_ahriman.base)
def test_logs_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must remove package logs
"""
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_remove")
watcher.logs_remove(package_ahriman.base, "42")
logs_mock.assert_called_once_with(package_ahriman.base, "42")
def test_logs_update_new(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must create package logs record for new package
"""
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.version)
assert watcher._last_log_record_id != log_record_id
watcher.logs_update(log_record_id, 42.01, "log record")
delete_mock.assert_called_once_with(package_ahriman.base, log_record_id.version)
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
assert watcher._last_log_record_id == log_record_id
def test_logs_update_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must create package logs record for current package
"""
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.version)
watcher._last_log_record_id = log_record_id
watcher.logs_update(log_record_id, 42.01, "log record")
delete_mock.assert_not_called()
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
def test_package_get(watcher: Watcher, package_ahriman: Package) -> None:
"""
must return package status
"""
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
package, status = watcher.package_get(package_ahriman.base)
assert package == package_ahriman
assert status.status == BuildStatusEnum.Unknown
def test_package_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
"""
must fail on unknown package
"""
with pytest.raises(UnknownPackageError):
watcher.package_get(package_ahriman.base)
def test_package_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must remove package base
"""
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
logs_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
logs_mock = mocker.patch("ahriman.core.status.watcher.Watcher.logs_remove")
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
watcher.remove(package_ahriman.base)
watcher.package_remove(package_ahriman.base)
assert not watcher.known
cache_mock.assert_called_once_with(package_ahriman.base)
logs_mock.assert_called_once_with(package_ahriman.base, None)
def test_remove_logs(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must remove package logs
"""
logs_mock = mocker.patch("ahriman.core.database.SQLite.logs_remove")
watcher.remove_logs(package_ahriman.base, 42)
logs_mock.assert_called_once_with(package_ahriman.base, 42)
def test_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must not fail on unknown base removal
"""
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_remove")
watcher.remove(package_ahriman.base)
watcher.package_remove(package_ahriman.base)
cache_mock.assert_called_once_with(package_ahriman.base)
def test_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must update package status
"""
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
watcher.package_update(package_ahriman.base, BuildStatusEnum.Unknown, package_ahriman)
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
def test_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_update_ping(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must update package status only for known package
"""
cache_mock = mocker.patch("ahriman.core.database.SQLite.package_update")
watcher.known = {package_ahriman.base: (package_ahriman, BuildStatus())}
watcher.update(package_ahriman.base, BuildStatusEnum.Success, None)
watcher.package_update(package_ahriman.base, BuildStatusEnum.Success, None)
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) -> None:
def test_package_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
"""
must fail on unknown package status update only
"""
with pytest.raises(UnknownPackageError):
watcher.update(package_ahriman.base, BuildStatusEnum.Unknown, None)
watcher.package_update(package_ahriman.base, BuildStatusEnum.Unknown, None)
def test_update_logs_new(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must create package logs record for new package
"""
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.process_id)
assert watcher._last_log_record_id != log_record_id
watcher.update_logs(log_record_id, 42.01, "log record")
delete_mock.assert_called_once_with(package_ahriman.base, log_record_id.process_id)
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
assert watcher._last_log_record_id == log_record_id
def test_update_logs_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must create package logs record for current package
"""
delete_mock = mocker.patch("ahriman.core.status.watcher.Watcher.remove_logs")
insert_mock = mocker.patch("ahriman.core.database.SQLite.logs_insert")
log_record_id = LogRecordId(package_ahriman.base, watcher._last_log_record_id.process_id)
watcher._last_log_record_id = log_record_id
watcher.update_logs(log_record_id, 42.01, "log record")
delete_mock.assert_not_called()
insert_mock.assert_called_once_with(log_record_id, 42.01, "log record")
def test_update_self(watcher: Watcher) -> None:
def test_status_update(watcher: Watcher) -> None:
"""
must update service status
"""
watcher.update_self(BuildStatusEnum.Success)
watcher.status_update(BuildStatusEnum.Success)
assert watcher.status.status == BuildStatusEnum.Success

366
tests/ahriman/core/status/test_web_client.py
View File

@ -5,12 +5,13 @@ import requests
import requests_unixsocket
from pytest_mock import MockerFixture
from requests import Response
from unittest.mock import call as MockCall
from ahriman.core.configuration import Configuration
from ahriman.core.status.web_client import WebClient
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
from ahriman.models.internal_status import InternalStatus
from ahriman.models.log_record_id import LogRecordId
from ahriman.models.package import Package
from ahriman.models.user import User
@ -19,7 +20,6 @@ def test_login_url(web_client: WebClient) -> None:
"""
must generate login url correctly
"""
assert web_client._login_url.startswith(web_client.address)
assert web_client._login_url.endswith("/api/v1/login")
@ -27,10 +27,24 @@ def test_status_url(web_client: WebClient) -> None:
"""
must generate package status url correctly
"""
assert web_client._status_url.startswith(web_client.address)
assert web_client._status_url.endswith("/api/v1/status")
def test_logs_url(web_client: WebClient, package_ahriman: Package) -> None:
"""
must generate logs url correctly
"""
assert web_client._logs_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}/logs")
def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
"""
must generate package status url correctly
"""
assert web_client._package_url("").endswith("/api/v1/packages")
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
def test_parse_address(configuration: Configuration) -> None:
"""
must extract address correctly
@ -74,14 +88,15 @@ def test_login(web_client: WebClient, user: User, mocker: MockerFixture) -> None
must login user
"""
web_client.user = user
requests_mock = mocker.patch("requests.Session.post")
requests_mock = mocker.patch("requests.Session.request")
payload = {
"username": user.username,
"password": user.password
}
web_client._login(requests.Session())
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json=payload)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True),
params=None, json=payload, files=None)
def test_login_failed(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
@ -89,7 +104,7 @@ def test_login_failed(web_client: WebClient, user: User, mocker: MockerFixture)
must suppress any exception happened during login
"""
web_client.user = user
mocker.patch("requests.Session.post", side_effect=Exception())
mocker.patch("requests.Session.request", side_effect=Exception())
web_client._login(requests.Session())
@ -98,7 +113,7 @@ def test_login_failed_http_error(web_client: WebClient, user: User, mocker: Mock
must suppress HTTP exception happened during login
"""
web_client.user = user
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client._login(requests.Session())
@ -106,265 +121,310 @@ def test_login_skip(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must skip login if no user set
"""
requests_mock = mocker.patch("requests.Session.post")
requests_mock = mocker.patch("requests.Session.request")
web_client._login(requests.Session())
requests_mock.assert_not_called()
def test_logs_url(web_client: WebClient, package_ahriman: Package) -> None:
def test_make_request(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must generate logs url correctly
must make HTTP request
"""
assert web_client._logs_url(package_ahriman.base).startswith(web_client.address)
assert web_client._logs_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}/logs")
request_mock = mocker.patch("requests.Session.request")
assert web_client.make_request("GET", "/url1") is not None
assert web_client.make_request("GET", "/url2", params=[("param", "value")]) is not None
assert web_client.make_request("POST", "/url3") is not None
assert web_client.make_request("POST", "/url4", json={"param": "value"}) is not None
# we don't want to put full descriptor here
assert web_client.make_request("POST", "/url5", files={"file": "tuple"}) is not None
assert web_client.make_request("DELETE", "/url6") is not None
request_mock.assert_has_calls([
MockCall("GET", f"{web_client.address}/url1", params=None, json=None, files=None),
MockCall().raise_for_status(),
MockCall("GET", f"{web_client.address}/url2", params=[("param", "value")], json=None, files=None),
MockCall().raise_for_status(),
MockCall("POST", f"{web_client.address}/url3", params=None, json=None, files=None),
MockCall().raise_for_status(),
MockCall("POST", f"{web_client.address}/url4", params=None, json={"param": "value"}, files=None),
MockCall().raise_for_status(),
MockCall("POST", f"{web_client.address}/url5", params=None, json=None, files={"file": "tuple"}),
MockCall().raise_for_status(),
MockCall("DELETE", f"{web_client.address}/url6", params=None, json=None, files=None),
MockCall().raise_for_status(),
])
def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
def test_make_request_failed(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must generate package status url correctly
must make HTTP request
"""
assert web_client._package_url("").startswith(web_client.address)
assert web_client._package_url("").endswith(f"/api/v1/packages")
assert web_client._package_url(package_ahriman.base).startswith(web_client.address)
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
mocker.patch("requests.Session.request", side_effect=Exception())
with pytest.raises(Exception):
web_client.make_request("GET", "url")
def test_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package addition
"""
requests_mock = mocker.patch("requests.Session.post")
requests_mock = mocker.patch("requests.Session.request")
payload = pytest.helpers.get_package_status(package_ahriman)
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json=payload)
web_client.package_add(package_ahriman, BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True),
params=None, json=payload, files=None)
def test_add_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_add_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during addition
"""
mocker.patch("requests.Session.post", side_effect=Exception())
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.package_add(package_ahriman, BuildStatusEnum.Unknown)
def test_add_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_add_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during addition
"""
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.package_add(package_ahriman, BuildStatusEnum.Unknown)
def test_add_failed_suppress(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_add_failed_suppress(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during addition and don't log
"""
web_client.suppress_errors = True
mocker.patch("requests.Session.post", side_effect=Exception())
mocker.patch("requests.Session.request", side_effect=Exception())
logging_mock = mocker.patch("logging.exception")
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
web_client.package_add(package_ahriman, BuildStatusEnum.Unknown)
logging_mock.assert_not_called()
def test_add_failed_http_error_suppress(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_add_failed_http_error_suppress(web_client: WebClient, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during addition and don't log
"""
web_client.suppress_errors = True
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
logging_mock = mocker.patch("logging.exception")
web_client.add(package_ahriman, BuildStatusEnum.Unknown)
web_client.package_add(package_ahriman, BuildStatusEnum.Unknown)
logging_mock.assert_not_called()
def test_get_all(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_get_all(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return all packages status
"""
response = [pytest.helpers.get_package_status_extended(package_ahriman)]
response_obj = Response()
response_obj = requests.Response()
response_obj._content = json.dumps(response).encode("utf8")
response_obj.status_code = 200
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
requests_mock = mocker.patch("requests.Session.request", return_value=response_obj)
result = web_client.get(None)
requests_mock.assert_called_once_with(web_client._package_url())
result = web_client.package_get(None)
requests_mock.assert_called_once_with("GET", f"{web_client.address}{web_client._package_url()}",
params=None, json=None, files=None)
assert len(result) == len(response)
assert (package_ahriman, BuildStatusEnum.Unknown) in [(package, status.status) for package, status in result]
def test_get_failed(web_client: WebClient, mocker: MockerFixture) -> None:
def test_package_get_failed(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during status getting
"""
mocker.patch("requests.Session.get", side_effect=Exception())
assert web_client.get(None) == []
mocker.patch("requests.Session.request", side_effect=Exception())
assert web_client.package_get(None) == []
def test_get_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
def test_package_get_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during status getting
"""
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
assert web_client.get(None) == []
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
assert web_client.package_get(None) == []
def test_get_single(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_get_single(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return single package status
"""
response = [pytest.helpers.get_package_status_extended(package_ahriman)]
response_obj = Response()
response_obj = requests.Response()
response_obj._content = json.dumps(response).encode("utf8")
response_obj.status_code = 200
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
requests_mock = mocker.patch("requests.Session.request", return_value=response_obj)
result = web_client.get(package_ahriman.base)
requests_mock.assert_called_once_with(web_client._package_url(package_ahriman.base))
result = web_client.package_get(package_ahriman.base)
requests_mock.assert_called_once_with("GET",
f"{web_client.address}{web_client._package_url(package_ahriman.base)}",
params=None, json=None, files=None)
assert len(result) == len(response)
assert (package_ahriman, BuildStatusEnum.Unknown) in [(package, status.status) for package, status in result]
def test_get_internal(web_client: WebClient, mocker: MockerFixture) -> None:
def test_package_logs(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must process log record
"""
requests_mock = mocker.patch("requests.Session.request")
payload = {
"created": log_record.created,
"message": log_record.getMessage(),
"version": package_ahriman.version,
}
web_client.package_logs(LogRecordId(package_ahriman.base, package_ahriman.version), log_record)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True),
params=None, json=payload, files=None)
def test_package_logs_failed(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must pass exception during log post
"""
mocker.patch("requests.Session.request", side_effect=Exception())
log_record.package_base = package_ahriman.base
with pytest.raises(Exception):
web_client.package_logs(LogRecordId(package_ahriman.base, package_ahriman.version), log_record)
def test_package_logs_failed_http_error(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must pass exception during log post
"""
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
log_record.package_base = package_ahriman.base
with pytest.raises(Exception):
web_client.package_logs(LogRecordId(package_ahriman.base, package_ahriman.version), log_record)
def test_package_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package removal
"""
requests_mock = mocker.patch("requests.Session.request")
web_client.package_remove(package_ahriman.base)
requests_mock.assert_called_once_with("DELETE", pytest.helpers.anyvar(str, True),
params=None, json=None, files=None)
def test_package_remove_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during removal
"""
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.package_remove(package_ahriman.base)
def test_package_remove_failed_http_error(web_client: WebClient, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during removal
"""
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.package_remove(package_ahriman.base)
def test_package_update(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package update
"""
requests_mock = mocker.patch("requests.Session.request")
web_client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True), params=None, json={
"status": BuildStatusEnum.Unknown.value
}, files=None)
def test_package_update_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during update
"""
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_package_update_failed_http_error(web_client: WebClient, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during update
"""
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_status_get(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must return web service status
"""
status = InternalStatus(status=BuildStatus(), architecture="x86_64")
response_obj = Response()
response_obj = requests.Response()
response_obj._content = json.dumps(status.view()).encode("utf8")
response_obj.status_code = 200
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
requests_mock = mocker.patch("requests.Session.request", return_value=response_obj)
result = web_client.get_internal()
requests_mock.assert_called_once_with(web_client._status_url)
result = web_client.status_get()
requests_mock.assert_called_once_with("GET", f"{web_client.address}{web_client._status_url}",
params=None, json=None, files=None)
assert result.architecture == "x86_64"
def test_get_internal_failed(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_get_failed(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during web service status getting
"""
mocker.patch("requests.Session.get", side_effect=Exception())
assert web_client.get_internal().architecture is None
mocker.patch("requests.Session.request", side_effect=Exception())
assert web_client.status_get().architecture is None
def test_get_internal_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_get_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during web service status getting
"""
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
assert web_client.get_internal().architecture is None
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
assert web_client.status_get().architecture is None
def test_logs(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must process log record
"""
requests_mock = mocker.patch("requests.Session.post")
payload = {
"created": log_record.created,
"message": log_record.getMessage(),
"process_id": log_record.process,
}
web_client.logs(package_ahriman.base, log_record)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json=payload)
def test_log_failed(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must pass exception during log post
"""
mocker.patch("requests.Session.post", side_effect=Exception())
log_record.package_base = package_ahriman.base
with pytest.raises(Exception):
web_client.logs(package_ahriman.base, log_record)
def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package removal
"""
requests_mock = mocker.patch("requests.Session.delete")
web_client.remove(package_ahriman.base)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True))
def test_remove_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during removal
"""
mocker.patch("requests.Session.delete", side_effect=Exception())
web_client.remove(package_ahriman.base)
def test_remove_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during removal
"""
mocker.patch("requests.Session.delete", side_effect=requests.exceptions.HTTPError())
web_client.remove(package_ahriman.base)
def test_update(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package update
"""
requests_mock = mocker.patch("requests.Session.post")
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json={
"status": BuildStatusEnum.Unknown.value})
def test_update_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during update
"""
mocker.patch("requests.Session.post", side_effect=Exception())
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_update_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during update
"""
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_update_self(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_update(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must process service update
"""
requests_mock = mocker.patch("requests.Session.post")
requests_mock = mocker.patch("requests.Session.request")
web_client.update_self(BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json={
"status": BuildStatusEnum.Unknown.value})
web_client.status_update(BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True), params=None, json={
"status": BuildStatusEnum.Unknown.value
}, files=None)
def test_update_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_update_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress any exception happened during service update
"""
mocker.patch("requests.Session.post", side_effect=Exception())
web_client.update_self(BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.status_update(BuildStatusEnum.Unknown)
def test_update_self_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_update_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must suppress HTTP exception happened during service update
"""
mocker.patch("requests.Session.post", side_effect=requests.exceptions.HTTPError())
web_client.update_self(BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.status_update(BuildStatusEnum.Unknown)

72
tests/ahriman/core/test_spawn.py
View File

@ -1,9 +1,17 @@
from pytest_mock import MockerFixture
from unittest.mock import MagicMock
from unittest.mock import MagicMock, call as MockCall
from ahriman.core.spawn import Spawn
def test_boolean_action_argument() -> None:
"""
must correctly convert argument to boolean flag
"""
assert Spawn.boolean_action_argument("option", True) == "option"
assert Spawn.boolean_action_argument("option", False) == "no-option"
def test_process(spawner: Spawn) -> None:
"""
must process external process run correctly
@ -15,9 +23,10 @@ def test_process(spawner: Spawn) -> None:
spawner.process(callback, args, spawner.architecture, "id", spawner.queue)
callback.assert_called_once_with(args, spawner.architecture)
(uuid, status) = spawner.queue.get()
(uuid, status, time) = spawner.queue.get()
assert uuid == "id"
assert status
assert time >= 0
assert spawner.queue.empty()
@ -30,9 +39,10 @@ def test_process_error(spawner: Spawn) -> None:
spawner.process(callback, MagicMock(), spawner.architecture, "id", spawner.queue)
(uuid, status) = spawner.queue.get()
(uuid, status, time) = spawner.queue.get()
assert uuid == "id"
assert not status
assert time >= 0
assert spawner.queue.empty()
@ -42,7 +52,7 @@ def test_spawn_process(spawner: Spawn, mocker: MockerFixture) -> None:
"""
start_mock = mocker.patch("multiprocessing.Process.start")
spawner._spawn_process("add", "ahriman", now="", maybe="?", none=None)
assert spawner._spawn_process("add", "ahriman", now="", maybe="?", none=None)
start_mock.assert_called_once_with()
spawner.args_parser.parse_args.assert_called_once_with(
spawner.command_arguments + [
@ -51,12 +61,22 @@ def test_spawn_process(spawner: Spawn, mocker: MockerFixture) -> None:
)
def test_has_process(spawner: Spawn) -> None:
"""
must correctly determine if there is a process
"""
assert not spawner.has_process("id")
spawner.active["id"] = MagicMock()
assert spawner.has_process("id")
def test_key_import(spawner: Spawn, mocker: MockerFixture) -> None:
"""
must call key import
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.key_import("0xdeadbeaf", None)
assert spawner.key_import("0xdeadbeaf", None)
spawn_mock.assert_called_once_with("service-key-import", "0xdeadbeaf")
@ -65,7 +85,7 @@ def test_key_import_with_server(spawner: Spawn, mocker: MockerFixture) -> None:
must call key import with server specified
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.key_import("0xdeadbeaf", "keyserver.ubuntu.com")
assert spawner.key_import("0xdeadbeaf", "keyserver.ubuntu.com")
spawn_mock.assert_called_once_with("service-key-import", "0xdeadbeaf", **{"key-server": "keyserver.ubuntu.com"})
@ -74,7 +94,7 @@ def test_packages_add(spawner: Spawn, mocker: MockerFixture) -> None:
must call package addition
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_add(["ahriman", "linux"], None, now=False)
assert spawner.packages_add(["ahriman", "linux"], None, now=False)
spawn_mock.assert_called_once_with("package-add", "ahriman", "linux", username=None)
@ -83,7 +103,7 @@ def test_packages_add_with_build(spawner: Spawn, mocker: MockerFixture) -> None:
must call package addition with update
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_add(["ahriman", "linux"], None, now=True)
assert spawner.packages_add(["ahriman", "linux"], None, now=True)
spawn_mock.assert_called_once_with("package-add", "ahriman", "linux", username=None, now="")
@ -92,7 +112,7 @@ def test_packages_add_with_username(spawner: Spawn, mocker: MockerFixture) -> No
must call package addition with username
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_add(["ahriman", "linux"], "username", now=False)
assert spawner.packages_add(["ahriman", "linux"], "username", now=False)
spawn_mock.assert_called_once_with("package-add", "ahriman", "linux", username="username")
@ -101,7 +121,7 @@ def test_packages_rebuild(spawner: Spawn, mocker: MockerFixture) -> None:
must call package rebuild
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_rebuild("python", "packager")
assert spawner.packages_rebuild("python", "packager")
spawn_mock.assert_called_once_with("repo-rebuild", **{"depends-on": "python", "username": "packager"})
@ -110,7 +130,7 @@ def test_packages_remove(spawner: Spawn, mocker: MockerFixture) -> None:
must call package removal
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_remove(["ahriman", "linux"])
assert spawner.packages_remove(["ahriman", "linux"])
spawn_mock.assert_called_once_with("package-remove", "ahriman", "linux")
@ -119,8 +139,26 @@ def test_packages_update(spawner: Spawn, mocker: MockerFixture) -> None:
must call repo update
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn._spawn_process")
spawner.packages_update("packager")
spawn_mock.assert_called_once_with("repo-update", username="packager")
assert spawner.packages_update("packager", aur=True, local=True, manual=True)
args = {"username": "packager", "aur": "", "local": "", "manual": ""}
spawn_mock.assert_called_once_with("repo-update", **args)
spawn_mock.reset_mock()
assert spawner.packages_update("packager", aur=False, local=True, manual=True)
args = {"username": "packager", "no-aur": "", "local": "", "manual": ""}
spawn_mock.assert_called_once_with("repo-update", **args)
spawn_mock.reset_mock()
assert spawner.packages_update("packager", aur=True, local=False, manual=True)
args = {"username": "packager", "aur": "", "no-local": "", "manual": ""}
spawn_mock.assert_called_once_with("repo-update", **args)
spawn_mock.reset_mock()
assert spawner.packages_update("packager", aur=True, local=True, manual=False)
args = {"username": "packager", "aur": "", "local": "", "no-manual": ""}
spawn_mock.assert_called_once_with("repo-update", **args)
spawn_mock.reset_mock()
def test_run(spawner: Spawn, mocker: MockerFixture) -> None:
@ -129,8 +167,8 @@ def test_run(spawner: Spawn, mocker: MockerFixture) -> None:
"""
logging_mock = mocker.patch("logging.Logger.info")
spawner.queue.put(("1", False))
spawner.queue.put(("2", True))
spawner.queue.put(("1", False, 1))
spawner.queue.put(("2", True, 1))
spawner.queue.put(None) # terminate
spawner.run()
@ -144,8 +182,8 @@ def test_run_pop(spawner: Spawn) -> None:
first = spawner.active["1"] = MagicMock()
second = spawner.active["2"] = MagicMock()
spawner.queue.put(("1", False))
spawner.queue.put(("2", True))
spawner.queue.put(("1", False, 1))
spawner.queue.put(("2", True, 1))
spawner.queue.put(None) # terminate
spawner.run()

9
tests/ahriman/core/test_util.py
View File

@ -273,6 +273,15 @@ def test_package_like(package_ahriman: Package) -> None:
assert package_like(package_ahriman.packages[package_ahriman.base].filepath)
def test_package_like_hidden(package_ahriman: Package) -> None:
"""
package_like must return false for hidden files
"""
package_file = package_ahriman.packages[package_ahriman.base].filepath
hidden_file = package_file.parent / f".{package_file.name}"
assert not package_like(hidden_file)
def test_package_like_sig(package_ahriman: Package) -> None:
"""
package_like must return false for signature files

17
tests/ahriman/core/upload/conftest.py
View File

@ -5,6 +5,7 @@ from unittest.mock import MagicMock
from ahriman.core.configuration import Configuration
from ahriman.core.upload.github import Github
from ahriman.core.upload.remote_service import RemoteService
from ahriman.core.upload.rsync import Rsync
from ahriman.core.upload.s3 import S3
@ -45,6 +46,22 @@ def github_release() -> dict[str, Any]:
}
@pytest.fixture
def remote_service(configuration: Configuration) -> RemoteService:
"""
fixture for remote service synchronization
Args:
configuration(Configuration): configuration fixture
Returns:
RemoteService: remote service test instance
"""
configuration.set_option("web", "host", "localhost")
configuration.set_option("web", "port", "8080")
return RemoteService("x86_64", configuration, "remote-service")
@pytest.fixture
def rsync(configuration: Configuration) -> Rsync:
"""

4
tests/ahriman/core/upload/test_http_upload.py
View File

@ -47,7 +47,7 @@ def test_request(github: Github, mocker: MockerFixture) -> None:
must call request method
"""
response_mock = MagicMock()
request_mock = mocker.patch("requests.request", return_value=response_mock)
request_mock = mocker.patch("requests.Session.request", return_value=response_mock)
github._request("GET", "url", arg="arg")
request_mock.assert_called_once_with("GET", "url", auth=github.auth, timeout=github.timeout, arg="arg")
@ -58,6 +58,6 @@ def test_request_exception(github: Github, mocker: MockerFixture) -> None:
"""
must call request method and log HTTPError exception
"""
mocker.patch("requests.request", side_effect=requests.HTTPError())
mocker.patch("requests.Session.request", side_effect=requests.HTTPError())
with pytest.raises(requests.HTTPError):
github._request("GET", "url", arg="arg")

82
tests/ahriman/core/upload/test_remote_service.py Normal file
View File

@ -0,0 +1,82 @@
import pytest
from pathlib import Path
from pytest_mock import MockerFixture
from unittest.mock import MagicMock, call as MockCall
from ahriman.core.upload.remote_service import RemoteService
from ahriman.models.package import Package
def test_session(remote_service: RemoteService, mocker: MockerFixture) -> None:
"""
must generate ahriman session
"""
upload_mock = mocker.patch("ahriman.core.status.web_client.WebClient._create_session")
assert remote_service.session
upload_mock.assert_called_once_with(use_unix_socket=False)
def test_package_upload(remote_service: RemoteService, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must upload package to remote host
"""
mocker.patch("pathlib.Path.is_file", return_value=False)
file_mock = MagicMock()
open_mock = mocker.patch("pathlib.Path.open", return_value=file_mock)
upload_mock = mocker.patch("ahriman.core.upload.http_upload.HttpUpload._request")
filename = package_ahriman.packages[package_ahriman.base].filename
remote_service.sync(Path("local"), [package_ahriman])
open_mock.assert_called_once_with("rb")
file_mock.close.assert_called_once()
upload_mock.assert_called_once_with("POST", f"{remote_service.client.address}/api/v1/service/upload", files={
"package": (filename, pytest.helpers.anyvar(int), "application/octet-stream", {})
})
def test_package_upload_with_signature(remote_service: RemoteService, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must upload package to remote host with signatures
"""
mocker.patch("pathlib.Path.is_file", return_value=True)
file_mock = MagicMock()
open_mock = mocker.patch("pathlib.Path.open", return_value=file_mock)
upload_mock = mocker.patch("ahriman.core.upload.http_upload.HttpUpload._request")
filename = package_ahriman.packages[package_ahriman.base].filename
remote_service.sync(Path("local"), [package_ahriman])
open_mock.assert_has_calls([MockCall("rb"), MockCall("rb")])
file_mock.close.assert_has_calls([MockCall(), MockCall()])
upload_mock.assert_called_once_with(
"POST", f"{remote_service.client.address}/api/v1/service/upload", files={
"package": (filename, pytest.helpers.anyvar(int), "application/octet-stream", {}),
"signature": (f"{filename}.sig", pytest.helpers.anyvar(int), "application/octet-stream", {})
}
)
def test_package_upload_no_filename(remote_service: RemoteService, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must skip upload if no filename set
"""
open_mock = mocker.patch("pathlib.Path.open")
upload_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request")
package_ahriman.packages[package_ahriman.base].filename = None
remote_service.sync(Path("local"), [package_ahriman])
open_mock.assert_not_called()
upload_mock.assert_not_called()
def test_sync(remote_service: RemoteService, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must run sync command
"""
upload_mock = mocker.patch("ahriman.core.upload.remote_service.RemoteService.package_upload")
local = Path("local")
remote_service.sync(local, [package_ahriman])
upload_mock.assert_called_once_with(local, package_ahriman)

12
tests/ahriman/core/upload/test_upload.py
View File

@ -53,3 +53,15 @@ def test_upload_github(configuration: Configuration, mocker: MockerFixture) -> N
upload_mock = mocker.patch("ahriman.core.upload.github.Github.sync")
Upload.load("x86_64", configuration, "github").run(Path("path"), [])
upload_mock.assert_called_once_with(Path("path"), [])
def test_upload_ahriman(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must upload via ahriman
"""
upload_mock = mocker.patch("ahriman.core.upload.remote_service.RemoteService.sync")
configuration.set_option("web", "host", "localhost")
configuration.set_option("web", "port", "8080")
Upload.load("x86_64", configuration, "remote-service").run(Path("path"), [])
upload_mock.assert_called_once_with(Path("path"), [])

5
tests/ahriman/models/test_report_settings.py
View File

@ -23,3 +23,8 @@ def test_from_option_valid() -> None:
assert ReportSettings.from_option("telegram") == ReportSettings.Telegram
assert ReportSettings.from_option("TElegraM") == ReportSettings.Telegram
assert ReportSettings.from_option("remote-call") == ReportSettings.RemoteCall
assert ReportSettings.from_option("reMOte-cALL") == ReportSettings.RemoteCall
assert ReportSettings.from_option("ahriman") == ReportSettings.RemoteCall
assert ReportSettings.from_option("AhRiMAN") == ReportSettings.RemoteCall

5
tests/ahriman/models/test_upload_settings.py
View File

@ -20,3 +20,8 @@ def test_from_option_valid() -> None:
assert UploadSettings.from_option("github") == UploadSettings.Github
assert UploadSettings.from_option("GitHub") == UploadSettings.Github
assert UploadSettings.from_option("remote-service") == UploadSettings.RemoteService
assert UploadSettings.from_option("Remote-Service") == UploadSettings.RemoteService
assert UploadSettings.from_option("ahriman") == UploadSettings.RemoteService
assert UploadSettings.from_option("AhRiMAN") == UploadSettings.RemoteService

29
tests/ahriman/models/test_waiter.py Normal file
View File

@ -0,0 +1,29 @@
import time
from ahriman.models.waiter import Waiter
def test_is_timed_out() -> None:
"""
must correctly check if timer runs out
"""
assert Waiter(-1).is_timed_out()
assert Waiter(1, start_time=time.monotonic() - 10.0).is_timed_out()
assert not Waiter(1, start_time=time.monotonic() + 10.0).is_timed_out()
def test_is_timed_out_infinite() -> None:
"""
must treat 0 wait timeout as infinite
"""
assert not Waiter(0).is_timed_out()
assert not Waiter(0, start_time=time.monotonic() - 10.0).is_timed_out()
def test_wait() -> None:
"""
must wait until file will disappear
"""
results = iter([True, False])
waiter = Waiter(1, interval=1)
assert waiter.wait(lambda: next(results)) > 0

1
tests/ahriman/web/schemas/test_file_schema.py Normal file
View File

@ -0,0 +1 @@
# schema testing goes in view class tests

1
tests/ahriman/web/schemas/test_process_id_schema.py Normal file
View File

@ -0,0 +1 @@
# schema testing goes in view class tests

1
tests/ahriman/web/schemas/test_process_schema.py Normal file
View File

@ -0,0 +1 @@
# schema testing goes in view class tests

1
tests/ahriman/web/schemas/test_update_flags_schema.py Normal file
View File

@ -0,0 +1 @@
# schema testing goes in view class tests

7
tests/ahriman/web/views/service/test_views_service_add.py
View File

@ -21,11 +21,12 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(AddView.post)
response_schema = pytest.helpers.schema_response(AddView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
@ -33,6 +34,10 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
assert response.ok
add_mock.assert_called_once_with(["ahriman"], "username", now=True)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_empty(client: TestClient, mocker: MockerFixture) -> None:
"""

7
tests/ahriman/web/views/service/test_views_service_pgp.py
View File

@ -66,8 +66,9 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import", return_value="abc")
request_schema = pytest.helpers.schema_request(PGPView.post)
response_schema = pytest.helpers.schema_response(PGPView.post)
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
assert not request_schema.validate(payload)
@ -75,6 +76,10 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
assert response.ok
import_mock.assert_called_once_with("0xdeadbeaf", "keyserver.ubuntu.com")
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""

46
tests/ahriman/web/views/service/test_views_service_process.py Normal file
View File

@ -0,0 +1,46 @@
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.process import ProcessView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await ProcessView.get_permission(request) == UserAccess.Reporter
async def test_get(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
process = "abc"
process_mock = mocker.patch("ahriman.core.spawn.Spawn.has_process", return_value=True)
response_schema = pytest.helpers.schema_response(ProcessView.get)
response = await client.get(f"/api/v1/service/process/{process}")
assert response.ok
process_mock.assert_called_once_with(process)
json = await response.json()
assert json["is_alive"]
assert not response_schema.validate(json)
async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must call raise 404 on unknown process
"""
process = "abc"
mocker.patch("ahriman.core.spawn.Spawn.has_process", return_value=False)
response_schema = pytest.helpers.schema_response(ProcessView.get, code=404)
response = await client.get(f"/api/v1/service/process/{process}")
assert response.status == 404
assert not response_schema.validate(await response.json())

7
tests/ahriman/web/views/service/test_views_service_rebuild.py
View File

@ -21,11 +21,12 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RebuildView.post)
response_schema = pytest.helpers.schema_response(RebuildView.post)
payload = {"packages": ["python", "ahriman"]}
assert not request_schema.validate(payload)
@ -33,6 +34,10 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
assert response.ok
rebuild_mock.assert_called_once_with("python", "username")
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""

7
tests/ahriman/web/views/service/test_views_service_remove.py
View File

@ -20,8 +20,9 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove", return_value="abc")
request_schema = pytest.helpers.schema_request(RemoveView.post)
response_schema = pytest.helpers.schema_response(RemoveView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
@ -29,6 +30,10 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
assert response.ok
remove_mock.assert_called_once_with(["ahriman"])
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""

7
tests/ahriman/web/views/service/test_views_service_request.py
View File

@ -21,11 +21,12 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RequestView.post)
response_schema = pytest.helpers.schema_response(RequestView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
@ -33,6 +34,10 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
assert response.ok
add_mock.assert_called_once_with(["ahriman"], "username", now=False)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""

56
tests/ahriman/web/views/service/test_views_service_update.py
View File

@ -1,17 +1,65 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.models.user_access import UserAccess
from ahriman.web.views.service.update import UpdateView
async def test_post_update(client: TestClient, mocker: MockerFixture) -> None:
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await UpdateView.get_permission(request) == UserAccess.Full
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly for alias
"""
update_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_update")
update_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_update", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(UpdateView.post)
response_schema = pytest.helpers.schema_response(UpdateView.post)
defaults = {
"aur": True,
"local": True,
"manual": True,
}
for payload in (
{},
{"aur": False},
{"local": False},
{"manual": False},
):
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/update", json=payload)
assert response.ok
update_mock.assert_called_once_with("username", **(defaults | payload))
update_mock.reset_mock()
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must call raise 400 on invalid request
"""
mocker.patch("ahriman.web.views.base.BaseView.extract_data", side_effect=Exception())
update_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_update")
response_schema = pytest.helpers.schema_response(UpdateView.post, code=400)
response = await client.post("/api/v1/service/update")
assert response.ok
update_mock.assert_called_once_with("username")
assert response.status == 400
assert not response_schema.validate(await response.json())
update_mock.assert_not_called()

179
tests/ahriman/web/views/service/test_views_service_upload.py Normal file
View File

@ -0,0 +1,179 @@
import pytest
from aiohttp import FormData
from aiohttp.test_utils import TestClient
from aiohttp.web import HTTPBadRequest
from io import BytesIO
from pathlib import Path
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock, MagicMock, call as MockCall
from ahriman.models.repository_paths import RepositoryPaths
from ahriman.models.user_access import UserAccess
from ahriman.web.views.service.upload import UploadView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await UploadView.get_permission(request) == UserAccess.Full
async def test_save_file(mocker: MockerFixture) -> None:
"""
must correctly save file
"""
part_mock = MagicMock()
part_mock.filename = "filename"
part_mock.read_chunk = AsyncMock(side_effect=[b"content", None])
tempfile_mock = mocker.patch("ahriman.web.views.service.upload.NamedTemporaryFile")
file_mock = MagicMock()
tempfile_mock.return_value.__enter__.return_value = file_mock
open_mock = mocker.patch("pathlib.Path.open")
copy_mock = mocker.patch("shutil.copyfileobj")
local = Path("local")
assert await UploadView.save_file(part_mock, local, max_body_size=None) == \
(part_mock.filename, local / f".{part_mock.filename}")
file_mock.write.assert_called_once_with(b"content")
open_mock.assert_called_once_with("wb")
copy_mock.assert_called_once_with(pytest.helpers.anyvar(int), pytest.helpers.anyvar(int))
async def test_save_file_no_filename() -> None:
"""
must raise exception on missing filename
"""
part_mock = MagicMock()
part_mock.filename = None
with pytest.raises(HTTPBadRequest):
await UploadView.save_file(part_mock, Path("local"), max_body_size=None)
async def test_save_file_invalid_filename() -> None:
"""
must raise exception on invalid filename
"""
part_mock = MagicMock()
part_mock.filename = ".."
with pytest.raises(HTTPBadRequest):
await UploadView.save_file(part_mock, Path("local"), max_body_size=None)
async def test_save_file_too_big() -> None:
"""
must raise exception on too big file
"""
part_mock = MagicMock()
part_mock.filename = "filename"
part_mock.read_chunk = AsyncMock(side_effect=[b"content", None])
with pytest.raises(HTTPBadRequest):
await UploadView.save_file(part_mock, Path("local"), max_body_size=0)
async def test_post(client: TestClient, repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
"""
must process file upload via http
"""
local = Path("local")
save_mock = mocker.patch("ahriman.web.views.service.upload.UploadView.save_file",
side_effect=AsyncMock(return_value=("filename", local / ".filename")))
rename_mock = mocker.patch("pathlib.Path.rename")
# no content validation here because it has invalid schema
data = FormData()
data.add_field("package", BytesIO(b"content"), filename="filename", content_type="application/octet-stream")
response = await client.post("/api/v1/service/upload", data=data)
assert response.ok
save_mock.assert_called_once_with(pytest.helpers.anyvar(int), repository_paths.packages, max_body_size=None)
rename_mock.assert_called_once_with(local / "filename")
async def test_post_with_sig(client: TestClient, repository_paths: RepositoryPaths, mocker: MockerFixture) -> None:
"""
must process file upload with signature via http
"""
local = Path("local")
save_mock = mocker.patch("ahriman.web.views.service.upload.UploadView.save_file",
side_effect=AsyncMock(side_effect=[
("filename", local / ".filename"),
("filename.sig", local / ".filename.sig"),
]))
rename_mock = mocker.patch("pathlib.Path.rename")
# no content validation here because it has invalid schema
data = FormData()
data.add_field("package", BytesIO(b"content"), filename="filename")
data.add_field("signature", BytesIO(b"sig"), filename="filename.sig")
response = await client.post("/api/v1/service/upload", data=data)
assert response.ok
save_mock.assert_has_calls([
MockCall(pytest.helpers.anyvar(int), repository_paths.packages, max_body_size=None),
MockCall(pytest.helpers.anyvar(int), repository_paths.packages, max_body_size=None),
])
rename_mock.assert_has_calls([
MockCall(local / "filename"),
MockCall(local / "filename.sig"),
])
async def test_post_not_found(client: TestClient, mocker: MockerFixture) -> None:
"""
must return 404 if request was disabled
"""
mocker.patch("ahriman.core.configuration.Configuration.getboolean", return_value=False)
data = FormData()
data.add_field("package", BytesIO(b"content"), filename="filename", content_type="application/octet-stream")
response_schema = pytest.helpers.schema_response(UploadView.post, code=404)
response = await client.post("/api/v1/service/upload", data=data)
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post_not_multipart(client: TestClient) -> None:
"""
must return 400 on invalid payload
"""
response_schema = pytest.helpers.schema_response(UploadView.post, code=400)
response = await client.post("/api/v1/service/upload")
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_post_not_body_part(client: TestClient, mocker: MockerFixture) -> None:
"""
must return 400 on invalid iterator in multipart
"""
response_schema = pytest.helpers.schema_response(UploadView.post, code=400)
mocker.patch("aiohttp.MultipartReader.next", return_value=42) # surprise, motherfucker
data = FormData()
data.add_field("package", BytesIO(b"content"), filename="filename", content_type="application/octet-stream")
response = await client.post("/api/v1/service/upload", data=data)
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_post_not_package(client: TestClient) -> None:
"""
must return 400 on invalid multipart key
"""
response_schema = pytest.helpers.schema_response(UploadView.post, code=400)
data = FormData()
data.add_field("random", BytesIO(b"content"), filename="filename", content_type="application/octet-stream")
response = await client.post("/api/v1/service/upload", data=data)
assert response.status == 400
assert not response_schema.validate(await response.json())

8
tests/ahriman/web/views/status/test_views_status_logs.py
View File

@ -30,9 +30,9 @@ async def test_delete(client: TestClient, package_ahriman: Package, package_pyth
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message", "process_id": 42})
json={"created": 42.0, "message": "message", "version": "42"})
await client.post(f"/api/v1/packages/{package_python_schedule.base}/logs",
json={"created": 42.0, "message": "message", "process_id": 42})
json={"created": 42.0, "message": "message", "version": "42"})
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}/logs")
assert response.status == 204
@ -53,7 +53,7 @@ async def test_get(client: TestClient, package_ahriman: Package) -> None:
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message", "process_id": 42})
json={"created": 42.0, "message": "message", "version": "42"})
response_schema = pytest.helpers.schema_response(LogsView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
@ -83,7 +83,7 @@ async def test_post(client: TestClient, package_ahriman: Package) -> None:
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(LogsView.post)
payload = {"created": 42.0, "message": "message", "process_id": 42}
payload = {"created": 42.0, "message": "message", "version": "42"}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/logs", json=payload)
assert response.status == 204

2
tests/ahriman/web/views/status/test_views_status_status.py
View File

@ -75,7 +75,7 @@ async def test_post_exception_inside(client: TestClient, mocker: MockerFixture)
exception handler must handle 500 errors
"""
payload = {"status": BuildStatusEnum.Success.value}
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
mocker.patch("ahriman.core.status.watcher.Watcher.status_update", side_effect=Exception())
response_schema = pytest.helpers.schema_response(StatusView.post, code=500)
response = await client.post("/api/v1/status", json=payload)

6
tests/testresources/core/ahriman.ini
View File

@ -74,6 +74,9 @@ homepage =
link_path =
template_path = ../web/templates/repo-index.jinja2
[remote-call]
manual = yes
[telegram]
api_key = apikey
chat_id = @ahrimantestchat
@ -103,10 +106,13 @@ password =
repository = ahriman
username = arcan1s
[remote-service]
[web]
debug = no
debug_check_host = no
debug_allowed_hosts =
enable_archive_upload = yes
host = 127.0.0.1
static_path = ../web/templates/static
templates = ../web/templates