arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-07-23 02:39:57 +00:00
Code Issues Projects Releases Wiki Activity

add external process spawner and update test cases

Browse Source
This commit is contained in:
Evgenii Alekseev
2021-09-07 03:15:50 +03:00
parent 18de70154e
commit a061ea96e6
15 changed files with 481 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,11 +6,23 @@ from ahriman.application.handlers import Web
from ahriman.core.configuration import Configuration
def _default_args(args: argparse.Namespace) -> argparse.Namespace:
"""
default arguments for these test cases
:param args: command line arguments fixture
:return: generated arguments for these test cases
"""
args.parser = True
return args
def test_run(args: argparse.Namespace, configuration: Configuration, mocker: MockerFixture) -> None:
"""
must run command
"""
args = _default_args(args)
mocker.patch("pathlib.Path.mkdir")
mocker.patch("ahriman.core.spawn.Spawn.start")
setup_mock = mocker.patch("ahriman.web.web.setup_service")
run_mock = mocker.patch("ahriman.web.web.run_server")

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

@ -260,11 +260,12 @@ def test_subparsers_update(parser: argparse.ArgumentParser) -> None:
def test_subparsers_web(parser: argparse.ArgumentParser) -> None:
"""
web command must imply lock and no_report
web command must imply lock, no_report and parser
"""
args = parser.parse_args(["-a", "x86_64", "web"])
assert args.lock is None
assert args.no_report
assert args.parser == parser
def test_run(args: argparse.Namespace, mocker: MockerFixture) -> None:

15
tests/ahriman/conftest.py
View File

@ -3,9 +3,11 @@ import pytest
from pathlib import Path
from pytest_mock import MockerFixture
from typing import Any, Type, TypeVar
from unittest.mock import MagicMock
from ahriman.core.auth.auth import Auth
from ahriman.core.configuration import Configuration
from ahriman.core.spawn import Spawn
from ahriman.core.status.watcher import Watcher
from ahriman.models.package import Package
from ahriman.models.package_description import PackageDescription
@ -13,6 +15,7 @@ from ahriman.models.repository_paths import RepositoryPaths
from ahriman.models.user import User
from ahriman.models.user_access import UserAccess
T = TypeVar("T")
@ -47,6 +50,7 @@ def anyvar(cls: Type[T], strict: bool = False) -> T:
def auth(configuration: Configuration) -> Auth:
"""
auth provider fixture
:param configuration: configuration fixture
:return: auth service instance
"""
return Auth(configuration)
@ -160,6 +164,7 @@ def package_description_python2_schedule() -> PackageDescription:
def repository_paths(configuration: Configuration) -> RepositoryPaths:
"""
repository paths fixture
:param configuration: configuration fixture
:return: repository paths test instance
"""
return RepositoryPaths(
@ -167,6 +172,16 @@ def repository_paths(configuration: Configuration) -> RepositoryPaths:
root=configuration.getpath("repository", "root"))
@pytest.fixture
def spawner(configuration: Configuration) -> Spawn:
"""
spawner fixture
:param configuration: configuration fixture
:return: spawner fixture
"""
return Spawn(MagicMock(), "x86_64", configuration)
@pytest.fixture
def user() -> User:
"""

121
tests/ahriman/core/test_spawn.py Normal file
View File

@ -0,0 +1,121 @@
from pytest_mock import MockerFixture
from unittest.mock import MagicMock
from ahriman.core.spawn import Spawn
def test_process(spawner: Spawn) -> None:
"""
must process external process run correctly
"""
args = MagicMock()
callback = MagicMock()
spawner.process(callback, args, spawner.architecture, spawner.configuration, "id", spawner.queue)
callback.assert_called_with(args, spawner.architecture, spawner.configuration)
(uuid, error) = spawner.queue.get()
assert uuid == "id"
assert error is None
assert spawner.queue.empty()
def test_process_error(spawner: Spawn) -> None:
"""
must process external run with error correctly
"""
callback = MagicMock()
callback.side_effect = Exception()
spawner.process(callback, MagicMock(), spawner.architecture, spawner.configuration, "id", spawner.queue)
(uuid, error) = spawner.queue.get()
assert uuid == "id"
assert isinstance(error, Exception)
assert spawner.queue.empty()
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"], now=False)
spawn_mock.assert_called_with("add", "ahriman", "linux")
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"], now=True)
spawn_mock.assert_called_with("add", "ahriman", "linux", now="")
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"])
spawn_mock.assert_called_with("remove", "ahriman", "linux")
def test_packages_update(spawner: Spawn, mocker: MockerFixture) -> None:
"""
must call package updates
"""
spawn_mock = mocker.patch("ahriman.core.spawn.Spawn.spawn_process")
spawner.packages_update(["ahriman", "linux"])
spawn_mock.assert_called_with("update", "ahriman", "linux")
def test_spawn_process(spawner: Spawn, mocker: MockerFixture) -> None:
"""
must correctly spawn child process
"""
start_mock = mocker.patch("multiprocessing.Process.start")
spawner.spawn_process("add", "ahriman", now="", maybe="?")
start_mock.assert_called_once()
spawner.args_parser.parse_args.assert_called_with([
"--architecture", spawner.architecture, "--configuration", str(spawner.configuration.path),
"add", "ahriman", "--now", "--maybe", "?"
])
def test_run(spawner: Spawn, mocker: MockerFixture) -> None:
"""
must implement run method
"""
logging_exception_mock = mocker.patch("logging.Logger.exception")
logging_info_mock = mocker.patch("logging.Logger.info")
spawner.queue.put(("1", None))
spawner.queue.put(("2", Exception()))
spawner.queue.put(None) # terminate
spawner.run()
logging_exception_mock.assert_called_once()
logging_info_mock.assert_called_once()
def test_run_pop(spawner: Spawn) -> None:
"""
must pop and terminate child process
"""
first = spawner.active["1"] = MagicMock()
second = spawner.active["2"] = MagicMock()
spawner.queue.put(("1", None))
spawner.queue.put(("2", Exception()))
spawner.queue.put(None) # terminate
spawner.run()
first.terminate.assert_called_once()
first.join.assert_called_once()
second.terminate.assert_called_once()
second.join.assert_called_once()
assert not spawner.active

31
tests/ahriman/web/conftest.py
View File

@ -1,41 +1,64 @@
import pytest
from aiohttp import web
from collections import namedtuple
from pytest_mock import MockerFixture
from typing import Any
import ahriman.core.auth.helpers
from ahriman.core.configuration import Configuration
from ahriman.core.spawn import Spawn
from ahriman.models.user import User
from ahriman.web.web import setup_service
_request = namedtuple("_request", ["app", "path", "method", "json", "post"])
@pytest.helpers.register
def request(app: web.Application, path: str, method: str, json: Any = None, data: Any = None) -> _request:
"""
request generator helper
:param app: application fixture
:param path: path for the request
:param method: method for the request
:param json: json payload of the request
:param data: form data payload of the request
:return: dummy request object
"""
return _request(app, path, method, json, data)
@pytest.fixture
def application(configuration: Configuration, mocker: MockerFixture) -> web.Application:
def application(configuration: Configuration, spawner: Spawn, mocker: MockerFixture) -> web.Application:
"""
application fixture
:param configuration: configuration fixture
:param spawner: spawner fixture
:param mocker: mocker object
:return: application test instance
"""
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", False)
mocker.patch("pathlib.Path.mkdir")
return setup_service("x86_64", configuration)
return setup_service("x86_64", configuration, spawner)
@pytest.fixture
def application_with_auth(configuration: Configuration, user: User, mocker: MockerFixture) -> web.Application:
def application_with_auth(configuration: Configuration, user: User, spawner: Spawn,
mocker: MockerFixture) -> web.Application:
"""
application fixture with auth enabled
:param configuration: configuration fixture
:param user: user descriptor fixture
:param spawner: spawner fixture
:param mocker: mocker object
:return: application test instance
"""
configuration.set_option("auth", "target", "configuration")
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", True)
mocker.patch("pathlib.Path.mkdir")
application = setup_service("x86_64", configuration)
application = setup_service("x86_64", configuration, spawner)
generated = User(user.username, user.hash_password(application["validator"].salt), user.access)
application["validator"]._users[generated.username] = generated

13
tests/ahriman/web/middlewares/conftest.py
View File

@ -1,23 +1,10 @@
import pytest
from collections import namedtuple
from ahriman.core.auth.auth import Auth
from ahriman.core.configuration import Configuration
from ahriman.models.user import User
from ahriman.web.middlewares.auth_handler import AuthorizationPolicy
_request = namedtuple("_request", ["path", "method"])
@pytest.fixture
def aiohttp_request() -> _request:
"""
fixture for aiohttp like object
:return: aiohttp like request test instance
"""
return _request("path", "GET")
@pytest.fixture
def authorization_policy(configuration: Configuration, user: User) -> AuthorizationPolicy:

19
tests/ahriman/web/middlewares/test_auth_handler.py
View File

@ -1,6 +1,7 @@
import pytest
from aiohttp import web
from pytest_mock import MockerFixture
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from ahriman.core.auth.auth import Auth
@ -29,11 +30,11 @@ async def test_permits(authorization_policy: AuthorizationPolicy, user: User) ->
authorization_policy.validator.verify_access.assert_called_with(user.username, user.access, "/endpoint")
async def test_auth_handler_api(aiohttp_request: Any, auth: Auth, mocker: MockerFixture) -> None:
async def test_auth_handler_api(auth: Auth, mocker: MockerFixture) -> None:
"""
must ask for status permission for api calls
"""
aiohttp_request = aiohttp_request._replace(path="/status-api")
aiohttp_request = pytest.helpers.request("", "/status-api", "GET")
request_handler = AsyncMock()
mocker.patch("ahriman.core.auth.auth.Auth.is_safe_request", return_value=False)
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
@ -43,11 +44,11 @@ async def test_auth_handler_api(aiohttp_request: Any, auth: Auth, mocker: Mocker
check_permission_mock.assert_called_with(aiohttp_request, UserAccess.Status, aiohttp_request.path)
async def test_auth_handler_api_post(aiohttp_request: Any, auth: Auth, mocker: MockerFixture) -> None:
async def test_auth_handler_api_post(auth: Auth, mocker: MockerFixture) -> None:
"""
must ask for status permission for api calls with POST
"""
aiohttp_request = aiohttp_request._replace(path="/status-api", method="POST")
aiohttp_request = pytest.helpers.request("", "/status-api", "POST")
request_handler = AsyncMock()
mocker.patch("ahriman.core.auth.auth.Auth.is_safe_request", return_value=False)
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
@ -57,12 +58,12 @@ async def test_auth_handler_api_post(aiohttp_request: Any, auth: Auth, mocker: M
check_permission_mock.assert_called_with(aiohttp_request, UserAccess.Status, aiohttp_request.path)
async def test_auth_handler_read(aiohttp_request: Any, auth: Auth, mocker: MockerFixture) -> None:
async def test_auth_handler_read(auth: Auth, mocker: MockerFixture) -> None:
"""
must ask for read permission for api calls with GET
"""
for method in ("GET", "HEAD", "OPTIONS"):
aiohttp_request = aiohttp_request._replace(method=method)
aiohttp_request = pytest.helpers.request("", "", method)
request_handler = AsyncMock()
mocker.patch("ahriman.core.auth.auth.Auth.is_safe_request", return_value=False)
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
@ -72,12 +73,12 @@ async def test_auth_handler_read(aiohttp_request: Any, auth: Auth, mocker: Mocke
check_permission_mock.assert_called_with(aiohttp_request, UserAccess.Read, aiohttp_request.path)
async def test_auth_handler_write(aiohttp_request: Any, auth: Auth, mocker: MockerFixture) -> None:
async def test_auth_handler_write(auth: Auth, mocker: MockerFixture) -> None:
"""
must ask for read permission for api calls with POST
"""
for method in ("CONNECT", "DELETE", "PATCH", "POST", "PUT", "TRACE"):
aiohttp_request = aiohttp_request._replace(method=method)
aiohttp_request = pytest.helpers.request("", "", method)
request_handler = AsyncMock()
mocker.patch("ahriman.core.auth.auth.Auth.is_safe_request", return_value=False)
check_permission_mock = mocker.patch("aiohttp_security.check_permission")

16
tests/ahriman/web/middlewares/test_exception_handler.py
View File

@ -3,45 +3,47 @@ import pytest
from aiohttp.web_exceptions import HTTPBadRequest
from pytest_mock import MockerFixture
from typing import Any
from unittest.mock import AsyncMock
from ahriman.web.middlewares.exception_handler import exception_handler
async def test_exception_handler(aiohttp_request: Any, mocker: MockerFixture) -> None:
async def test_exception_handler(mocker: MockerFixture) -> None:
"""
must pass success response
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock()
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
await handler(aiohttp_request, request_handler)
await handler(request, request_handler)
logging_mock.assert_not_called()
async def test_exception_handler_client_error(aiohttp_request: Any, mocker: MockerFixture) -> None:
async def test_exception_handler_client_error(mocker: MockerFixture) -> None:
"""
must pass client exception
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPBadRequest())
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
with pytest.raises(HTTPBadRequest):
await handler(aiohttp_request, request_handler)
await handler(request, request_handler)
logging_mock.assert_not_called()
async def test_exception_handler_server_error(aiohttp_request: Any, mocker: MockerFixture) -> None:
async def test_exception_handler_server_error(mocker: MockerFixture) -> None:
"""
must log server exception and re-raise it
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=Exception())
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
with pytest.raises(Exception):
await handler(aiohttp_request, request_handler)
await handler(request, request_handler)
logging_mock.assert_called_once()

12
tests/ahriman/web/views/conftest.py
View File

@ -6,6 +6,18 @@ from pytest_aiohttp import TestClient
from pytest_mock import MockerFixture
from typing import Any
from ahriman.web.views.base import BaseView
@pytest.fixture
def base(application: web.Application) -> BaseView:
"""
base view fixture
:param application: application fixture
:return: generated base view fixture
"""
return BaseView(pytest.helpers.request(application, "", ""))
@pytest.fixture
def client(application: web.Application, loop: BaseEventLoop,

78
tests/ahriman/web/views/test_view_base.py Normal file
View File

@ -0,0 +1,78 @@
import pytest
from multidict import MultiDict
from ahriman.web.views.base import BaseView
def test_service(base: BaseView) -> None:
"""
must return service
"""
assert base.service
def test_spawn(base: BaseView) -> None:
"""
must return spawn thread
"""
assert base.spawner
def test_validator(base: BaseView) -> None:
"""
must return service
"""
assert base.validator
async def test_extract_data_json(base: BaseView) -> None:
"""
must parse and return json
"""
json = {"key1": "value1", "key2": "value2"}
async def get_json():
return json
base._request = pytest.helpers.request(base.request.app, "", "", json=get_json)
assert await base.extract_data() == json
async def test_extract_data_post(base: BaseView) -> None:
"""
must parse and return form data
"""
json = {"key1": "value1", "key2": "value2"}
async def get_json():
raise ValueError()
async def get_data():
return json
base._request = pytest.helpers.request(base.request.app, "", "", json=get_json, data=get_data)
assert await base.extract_data() == json
async def test_data_as_json(base: BaseView) -> None:
"""
must parse multi value form payload
"""
json = {"key1": "value1", "key2": ["value2", "value3"], "key3": ["value4", "value5", "value6"]}
async def get_json():
raise ValueError()
async def get_data():
result = MultiDict()
for key, values in json.items():
if isinstance(values, list):
for value in values:
result.add(key, value)
else:
result.add(key, values)
return result
base._request = pytest.helpers.request(base.request.app, "", "", json=get_json, data=get_data)
assert await base.data_as_json() == json