arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-07-15 06:55:48 +00:00
Code Issues Projects Releases Wiki Activity

add remote call trigger implementation

Browse Source
This commit is contained in:
Evgenii Alekseev
2023-08-13 15:26:45 +03:00
parent 37d3b9fa83
commit af1803ed26
27 changed files with 705 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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)

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

@ -67,6 +67,7 @@ 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("report")

31
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")
@ -156,30 +156,13 @@ def test_create_unsafe(lock: Lock) -> None:
def test_watch(lock: Lock, mocker: MockerFixture) -> None:
"""
must check if lock file exists in cycle
must check if lock file exists
"""
mocker.patch("pathlib.Path.is_file", return_value=False)
lock.watch()
def test_watch_wait(lock: Lock, mocker: MockerFixture) -> None:
"""
must wait until file will disappear
"""
mocker.patch("pathlib.Path.is_file", side_effect=[True, False])
wait_mock = mocker.patch("ahriman.models.waiter.Waiter.wait")
lock.path = Path(tempfile.mktemp()) # nosec
lock.wait_timeout = 1
lock.watch(1)
def test_watch_empty_timeout(lock: Lock, mocker: MockerFixture) -> None:
"""
must skip watch on empty timeout
"""
mocker.patch("pathlib.Path.is_file", return_value=True)
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:
@ -199,7 +182,7 @@ def test_enter(lock: Lock, mocker: MockerFixture) -> None:
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
@ -218,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:

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

@ -40,48 +40,20 @@ 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_mock = mocker.patch("ahriman.core.status.client.Client.package_logs")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
handler = HttpLogHandler(configuration, report=False)
handler.emit(log_record)
log_mock.assert_called_once_with(package_ahriman.base, log_record)
def test_emit_failed(configuration: Configuration, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must call handle error on exception
"""
log_record.package_base = package_ahriman.base
mocker.patch("ahriman.core.status.client.Client.logs", side_effect=Exception())
handle_error_mock = mocker.patch("logging.Handler.handleError")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
handler.emit(log_record)
handle_error_mock.assert_called_once_with(log_record)
def test_emit_suppress_failed(configuration: Configuration, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
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())
handle_error_mock = mocker.patch("logging.Handler.handleError")
handler = HttpLogHandler(configuration, report=False, suppress_errors=True)
handler.emit(log_record)
handle_error_mock.assert_not_called()
def test_emit_skip(configuration: Configuration, log_record: logging.LogRecord, mocker: MockerFixture) -> None:
"""
must skip log record posting if no package base set
"""
log_mock = mocker.patch("ahriman.core.status.client.Client.logs")
handler = HttpLogHandler(configuration, report=False, suppress_errors=False)
log_mock = mocker.patch("ahriman.core.status.client.Client.package_logs")
handler = HttpLogHandler(configuration, report=False)
handler.emit(log_record)
log_mock.assert_not_called()

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

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

@ -0,0 +1,85 @@
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", f"{remote_call.client.address}/api/v1/service/process/id")
def test_is_process_alive_unknown(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must correctly define if process is unknown
"""
mocker.patch("ahriman.core.status.web_client.WebClient.make_request", return_value=None)
assert not 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", f"{remote_call.client.address}/api/v1/service/update", json={
"aur": False,
"local": False,
"manual": True,
})
def test_remote_update_failed(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must return empty process id in case of errors
"""
mocker.patch("ahriman.core.status.web_client.WebClient.make_request", return_value=None)
assert remote_call.generate([], Result()) is None
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")
def test_remote_wait_skip(remote_call: RemoteCall, mocker: MockerFixture) -> None:
"""
must skip wait if process id is unknown
"""
wait_mock = mocker.patch("ahriman.models.waiter.Waiter.wait")
remote_call.remote_wait(None)
wait_mock.assert_not_called()

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

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

@ -51,64 +51,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_log(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(package_ahriman.base, 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 +101,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 +111,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 +121,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 +131,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)

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

@ -5,7 +5,7 @@ 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
@ -74,14 +74,14 @@ 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)
def test_login_failed(web_client: WebClient, user: User, mocker: MockerFixture) -> None:
@ -89,7 +89,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 +98,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,7 +106,7 @@ 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()
@ -130,241 +130,292 @@ def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
def test_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_make_request(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must make HTTP request
"""
request_mock = mocker.patch("requests.Session.request")
assert web_client.make_request("GET", "url") is not None
assert web_client.make_request("GET", "url", params=[("param", "value")]) is not None
assert web_client.make_request("POST", "url") is not None
assert web_client.make_request("POST", "url", json={"param": "value"}) is not None
assert web_client.make_request("DELETE", "url") is not None
request_mock.assert_has_calls([
MockCall("GET", "url", params=None, json=None),
MockCall().raise_for_status(),
MockCall("GET", "url", params=[("param", "value")], json=None),
MockCall().raise_for_status(),
MockCall("POST", "url", params=None, json=None),
MockCall().raise_for_status(),
MockCall("POST", "url", params=None, json={"param": "value"}),
MockCall().raise_for_status(),
MockCall("DELETE", "url", params=None, json=None),
MockCall().raise_for_status(),
])
def test_make_request_failed(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must make HTTP request
"""
mocker.patch("requests.Session.request", side_effect=Exception())
assert web_client.make_request("GET", "url") is 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)
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", web_client._package_url(), params=None, json=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", web_client._package_url(package_ahriman.base),
params=None, json=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:
"""
must return web service status
"""
status = InternalStatus(status=BuildStatus(), architecture="x86_64")
response_obj = 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)
result = web_client.get_internal()
requests_mock.assert_called_once_with(web_client._status_url)
assert result.architecture == "x86_64"
def test_get_internal_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
def test_get_internal_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
def test_logs(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
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.post")
requests_mock = mocker.patch("requests.Session.request")
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)
web_client.package_logs(package_ahriman.base, log_record)
requests_mock.assert_called_once_with("POST", pytest.helpers.anyvar(str, True), params=None, json=payload)
def test_log_failed(web_client: WebClient, log_record: logging.LogRecord, package_ahriman: Package,
mocker: MockerFixture) -> 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.post", side_effect=Exception())
mocker.patch("requests.Session.request", side_effect=Exception())
log_record.package_base = package_ahriman.base
with pytest.raises(Exception):
web_client.logs(package_ahriman.base, log_record)
web_client.package_logs(package_ahriman.base, log_record)
def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
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
web_client.package_logs(package_ahriman.base, 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.delete")
requests_mock = mocker.patch("requests.Session.request")
web_client.remove(package_ahriman.base)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True))
web_client.package_remove(package_ahriman.base)
requests_mock.assert_called_once_with("DELETE", pytest.helpers.anyvar(str, True), params=None, json=None)
def test_remove_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> 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.delete", side_effect=Exception())
web_client.remove(package_ahriman.base)
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.package_remove(package_ahriman.base)
def test_remove_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
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.delete", side_effect=requests.exceptions.HTTPError())
web_client.remove(package_ahriman.base)
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.package_remove(package_ahriman.base)
def test_update(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
def test_package_update(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package update
"""
requests_mock = mocker.patch("requests.Session.post")
requests_mock = mocker.patch("requests.Session.request")
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
requests_mock.assert_called_once_with(pytest.helpers.anyvar(str, True), json={
"status": BuildStatusEnum.Unknown.value})
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
})
def test_update_failed(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> 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.post", side_effect=Exception())
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=Exception())
web_client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_update_failed_http_error(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
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.post", side_effect=requests.exceptions.HTTPError())
web_client.update(package_ahriman.base, BuildStatusEnum.Unknown)
mocker.patch("requests.Session.request", side_effect=requests.exceptions.HTTPError())
web_client.package_update(package_ahriman.base, BuildStatusEnum.Unknown)
def test_update_self(web_client: WebClient, mocker: MockerFixture) -> None:
def test_status_get(web_client: WebClient, mocker: MockerFixture) -> None:
"""
must return web service status
"""
status = InternalStatus(status=BuildStatus(), architecture="x86_64")
response_obj = requests.Response()
response_obj._content = json.dumps(status.view()).encode("utf8")
response_obj.status_code = 200
requests_mock = mocker.patch("requests.Session.request", return_value=response_obj)
result = web_client.status_get()
requests_mock.assert_called_once_with("GET", web_client._status_url, params=None, json=None)
assert result.architecture == "x86_64"
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.request", side_effect=Exception())
assert web_client.status_get().architecture is 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.request", side_effect=requests.exceptions.HTTPError())
assert web_client.status_get().architecture is 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
})
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)

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

@ -23,3 +23,6 @@ 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

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

3
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