mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-06-28 06:41:43 +00:00
@ -34,7 +34,7 @@ def test_run(args: argparse.Namespace, configuration: Configuration, package_ahr
|
||||
"""
|
||||
args = _default_args(args)
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
application_mock = mocker.patch("ahriman.core.status.client.Client.get_self")
|
||||
application_mock = mocker.patch("ahriman.core.status.client.Client.get_internal")
|
||||
packages_mock = mocker.patch("ahriman.core.status.client.Client.get",
|
||||
return_value=[(package_ahriman, BuildStatus(BuildStatusEnum.Success)),
|
||||
(package_python_schedule, BuildStatus(BuildStatusEnum.Failed))])
|
||||
@ -55,7 +55,7 @@ def test_run_empty_exception(args: argparse.Namespace, configuration: Configurat
|
||||
args = _default_args(args)
|
||||
args.exit_code = True
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch("ahriman.core.status.client.Client.get_self")
|
||||
mocker.patch("ahriman.core.status.client.Client.get_internal")
|
||||
mocker.patch("ahriman.core.status.client.Client.get", return_value=[])
|
||||
check_mock = mocker.patch("ahriman.application.handlers.Handler.check_if_empty")
|
||||
|
||||
|
@ -28,7 +28,7 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
|
||||
args.as_service = False
|
||||
args.exit_code = False
|
||||
args.password = "pa55w0rd"
|
||||
args.role = UserAccess.Read
|
||||
args.role = UserAccess.Reporter
|
||||
args.secure = False
|
||||
return args
|
||||
|
||||
|
@ -519,7 +519,7 @@ def test_subparsers_user_add_option_role(parser: argparse.ArgumentParser) -> Non
|
||||
"""
|
||||
args = parser.parse_args(["user-add", "username"])
|
||||
assert isinstance(args.role, UserAccess)
|
||||
args = parser.parse_args(["user-add", "username", "--role", "write"])
|
||||
args = parser.parse_args(["user-add", "username", "--role", "full"])
|
||||
assert isinstance(args.role, UserAccess)
|
||||
|
||||
|
||||
@ -549,7 +549,7 @@ def test_subparsers_user_list_option_role(parser: argparse.ArgumentParser) -> No
|
||||
"""
|
||||
user-list command must convert role option to useraccess instance
|
||||
"""
|
||||
args = parser.parse_args(["user-list", "--role", "write"])
|
||||
args = parser.parse_args(["user-list", "--role", "full"])
|
||||
assert isinstance(args.role, UserAccess)
|
||||
|
||||
|
||||
|
@ -8,7 +8,7 @@ from unittest import mock
|
||||
from ahriman import version
|
||||
from ahriman.application.lock import Lock
|
||||
from ahriman.core.exceptions import DuplicateRun, UnsafeRun
|
||||
from ahriman.models.build_status import BuildStatusEnum
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
|
||||
|
||||
@ -57,7 +57,7 @@ def test_check_version(lock: Lock, mocker: MockerFixture) -> None:
|
||||
must check version correctly
|
||||
"""
|
||||
mocker.patch("ahriman.core.status.client.Client.get_internal",
|
||||
return_value=InternalStatus(version=version.__version__))
|
||||
return_value=InternalStatus(status=BuildStatus(), version=version.__version__))
|
||||
logging_mock = mocker.patch("logging.Logger.warning")
|
||||
|
||||
lock.check_version()
|
||||
@ -69,7 +69,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",
|
||||
return_value=InternalStatus(version="version"))
|
||||
return_value=InternalStatus(status=BuildStatus(), version="version"))
|
||||
logging_mock = mocker.patch("logging.Logger.warning")
|
||||
|
||||
lock.check_version()
|
||||
|
@ -426,7 +426,7 @@ def user() -> User:
|
||||
Returns:
|
||||
User: user descriptor instance
|
||||
"""
|
||||
return User("user", "pa55w0rd", UserAccess.Read)
|
||||
return User("user", "pa55w0rd", UserAccess.Reporter)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
@ -71,4 +71,4 @@ async def test_verify_access(auth: Auth, user: User) -> None:
|
||||
must allow any access
|
||||
"""
|
||||
assert await auth.verify_access(user.username, user.access, None)
|
||||
assert await auth.verify_access(user.username, UserAccess.Write, None)
|
||||
assert await auth.verify_access(user.username, UserAccess.Full, None)
|
||||
|
@ -79,4 +79,4 @@ async def test_verify_access(mapping: Mapping, user: User, mocker: MockerFixture
|
||||
"""
|
||||
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
|
||||
assert await mapping.verify_access(user.username, user.access, None)
|
||||
assert not await mapping.verify_access(user.username, UserAccess.Write, None)
|
||||
assert not await mapping.verify_access(user.username, UserAccess.Full, None)
|
||||
|
@ -0,0 +1,8 @@
|
||||
from ahriman.core.database.migrations.m002_user_access import steps
|
||||
|
||||
|
||||
def test_migration_package_source() -> None:
|
||||
"""
|
||||
migration must not be empty
|
||||
"""
|
||||
assert steps
|
@ -29,7 +29,7 @@ def test_user_list_filter_by_username(database: SQLite) -> None:
|
||||
must return users filtered by its id
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
second = User("2", "", UserAccess.Full)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
@ -46,7 +46,7 @@ def test_user_list_filter_by_access(database: SQLite) -> None:
|
||||
must return users filtered by its access
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
second = User("2", "", UserAccess.Full)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
@ -64,7 +64,7 @@ def test_user_list_filter_by_username_access(database: SQLite) -> None:
|
||||
must return users filtered by its access and username
|
||||
"""
|
||||
first = User("1", "", UserAccess.Read)
|
||||
second = User("2", "", UserAccess.Write)
|
||||
second = User("2", "", UserAccess.Full)
|
||||
third = User("3", "", UserAccess.Read)
|
||||
|
||||
database.user_update(first)
|
||||
@ -72,7 +72,7 @@ def test_user_list_filter_by_username_access(database: SQLite) -> None:
|
||||
database.user_update(third)
|
||||
|
||||
assert database.user_list("1", UserAccess.Read) == [first]
|
||||
assert not database.user_list("1", UserAccess.Write)
|
||||
assert not database.user_list("1", UserAccess.Full)
|
||||
|
||||
|
||||
def test_user_remove_update(database: SQLite, user: User) -> None:
|
||||
@ -92,6 +92,6 @@ def test_user_update(database: SQLite, user: User) -> None:
|
||||
assert database.user_get(user.username) == user
|
||||
|
||||
new_user = user.hash_password("salt")
|
||||
new_user.access = UserAccess.Write
|
||||
new_user.access = UserAccess.Full
|
||||
database.user_update(new_user)
|
||||
assert database.user_get(new_user.username) == new_user
|
||||
|
@ -3,7 +3,7 @@ from pytest_mock import MockerFixture
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.status.client import Client
|
||||
from ahriman.core.status.web_client import WebClient
|
||||
from ahriman.models.build_status import BuildStatusEnum
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
from ahriman.models.package import Package
|
||||
|
||||
@ -51,14 +51,11 @@ def test_get_internal(client: Client) -> None:
|
||||
"""
|
||||
must return dummy status for web service
|
||||
"""
|
||||
assert client.get_internal() == InternalStatus()
|
||||
expected = InternalStatus(BuildStatus())
|
||||
actual = client.get_internal()
|
||||
actual.status.timestamp = expected.status.timestamp
|
||||
|
||||
|
||||
def test_get_self(client: Client) -> None:
|
||||
"""
|
||||
must return unknown status for service
|
||||
"""
|
||||
assert client.get_self().status == BuildStatusEnum.Unknown
|
||||
assert actual == expected
|
||||
|
||||
|
||||
def test_remove(client: Client, package_ahriman: Package) -> None:
|
||||
|
@ -13,14 +13,6 @@ from ahriman.models.package import Package
|
||||
from ahriman.models.user import User
|
||||
|
||||
|
||||
def test_ahriman_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate service status url correctly
|
||||
"""
|
||||
assert web_client._ahriman_url.startswith(web_client.address)
|
||||
assert web_client._ahriman_url.endswith("/status-api/v1/ahriman")
|
||||
|
||||
|
||||
def test_status_url(web_client: WebClient) -> None:
|
||||
"""
|
||||
must generate package status url correctly
|
||||
@ -173,7 +165,7 @@ def test_get_internal(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
must return web service status
|
||||
"""
|
||||
response_obj = Response()
|
||||
response_obj._content = json.dumps(InternalStatus(architecture="x86_64").view()).encode("utf8")
|
||||
response_obj._content = json.dumps(InternalStatus(BuildStatus(), architecture="x86_64").view()).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
@ -188,7 +180,7 @@ def test_get_internal_failed(web_client: WebClient, mocker: MockerFixture) -> No
|
||||
must suppress any exception happened during web service status getting
|
||||
"""
|
||||
mocker.patch("requests.Session.get", side_effect=Exception())
|
||||
assert web_client.get_internal() == InternalStatus()
|
||||
assert web_client.get_internal().architecture is None
|
||||
|
||||
|
||||
def test_get_internal_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
@ -196,38 +188,7 @@ def test_get_internal_failed_http_error(web_client: WebClient, mocker: MockerFix
|
||||
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() == InternalStatus()
|
||||
|
||||
|
||||
def test_get_self(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return service status
|
||||
"""
|
||||
response_obj = Response()
|
||||
response_obj._content = json.dumps(BuildStatus().view()).encode("utf8")
|
||||
response_obj.status_code = 200
|
||||
|
||||
requests_mock = mocker.patch("requests.Session.get", return_value=response_obj)
|
||||
|
||||
result = web_client.get_self()
|
||||
requests_mock.assert_called_once_with(web_client._ahriman_url)
|
||||
assert result.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_get_self_failed(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress any exception happened during service status getting
|
||||
"""
|
||||
mocker.patch("requests.Session.get", side_effect=Exception())
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
def test_get_self_failed_http_error(web_client: WebClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must suppress HTTP exception happened during service status getting
|
||||
"""
|
||||
mocker.patch("requests.Session.get", side_effect=requests.exceptions.HTTPError())
|
||||
assert web_client.get_self().status == BuildStatusEnum.Unknown
|
||||
assert web_client.get_internal().architecture is None
|
||||
|
||||
|
||||
def test_remove(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
|
||||
|
@ -320,9 +320,11 @@ def test_walk(resource_path_root: Path) -> None:
|
||||
resource_path_root / "models" / "package_gcc10_srcinfo",
|
||||
resource_path_root / "models" / "package_tpacpi-bat-git_srcinfo",
|
||||
resource_path_root / "models" / "package_yay_srcinfo",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "failed-modal.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "login-modal.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "package-actions-modals.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "package-actions-script.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "package-add-modal.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "success-modal.jinja2",
|
||||
resource_path_root / "web" / "templates" / "build-status" / "table.jinja2",
|
||||
resource_path_root / "web" / "templates" / "static" / "favicon.ico",
|
||||
resource_path_root / "web" / "templates" / "utils" / "bootstrap-scripts.jinja2",
|
||||
resource_path_root / "web" / "templates" / "utils" / "style.jinja2",
|
||||
|
@ -54,7 +54,8 @@ def internal_status(counters: Counters) -> InternalStatus:
|
||||
Returns:
|
||||
InternalStatus: internal status test instance
|
||||
"""
|
||||
return InternalStatus(architecture="x86_64",
|
||||
return InternalStatus(status=BuildStatus(),
|
||||
architecture="x86_64",
|
||||
packages=counters,
|
||||
version=version.__version__,
|
||||
repository="aur-clone")
|
||||
|
@ -4,31 +4,6 @@ import time
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
|
||||
|
||||
def test_build_status_enum_badges_color() -> None:
|
||||
"""
|
||||
status color must be one of shields.io supported
|
||||
"""
|
||||
SUPPORTED_COLORS = [
|
||||
"brightgreen", "green", "yellowgreen", "yellow", "orange", "red", "blue", "lightgrey",
|
||||
"success", "important", "critical", "informational", "inactive", "blueviolet"
|
||||
]
|
||||
|
||||
for status in BuildStatusEnum:
|
||||
assert status.badges_color() in SUPPORTED_COLORS
|
||||
|
||||
|
||||
def test_build_status_enum_bootstrap_color() -> None:
|
||||
"""
|
||||
status color must be one of bootstrap supported
|
||||
"""
|
||||
SUPPORTED_COLORS = [
|
||||
"primary", "secondary", "success", "danger", "warning", "info", "light", "dark"
|
||||
]
|
||||
|
||||
for status in BuildStatusEnum:
|
||||
assert status.bootstrap_color() in SUPPORTED_COLORS
|
||||
|
||||
|
||||
def test_build_status_init_1() -> None:
|
||||
"""
|
||||
must construct status object from None
|
||||
|
@ -6,9 +6,10 @@ def test_from_option(user: User) -> None:
|
||||
"""
|
||||
must generate user from options
|
||||
"""
|
||||
user.access = UserAccess.Read
|
||||
assert User.from_option(user.username, user.password) == user
|
||||
# default is read access
|
||||
user.access = UserAccess.Write
|
||||
user.access = UserAccess.Full
|
||||
assert User.from_option(user.username, user.password) != user
|
||||
assert User.from_option(user.username, user.password, user.access) == user
|
||||
|
||||
@ -72,16 +73,16 @@ def test_verify_access_read(user: User) -> None:
|
||||
"""
|
||||
user.access = UserAccess.Read
|
||||
assert user.verify_access(UserAccess.Read)
|
||||
assert not user.verify_access(UserAccess.Write)
|
||||
assert not user.verify_access(UserAccess.Full)
|
||||
|
||||
|
||||
def test_verify_access_write(user: User) -> None:
|
||||
"""
|
||||
user with write access must be able to do anything
|
||||
"""
|
||||
user.access = UserAccess.Write
|
||||
user.access = UserAccess.Full
|
||||
assert user.verify_access(UserAccess.Read)
|
||||
assert user.verify_access(UserAccess.Write)
|
||||
assert user.verify_access(UserAccess.Full)
|
||||
|
||||
|
||||
def test_repr(user: User) -> None:
|
||||
|
@ -0,0 +1,51 @@
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.models.user_access import UserAccess
|
||||
|
||||
|
||||
def test_permits_full() -> None:
|
||||
"""
|
||||
full access must allow everything
|
||||
"""
|
||||
assert UserAccess.Full.permits(UserAccess.Full)
|
||||
assert UserAccess.Full.permits(UserAccess.Reporter)
|
||||
assert UserAccess.Full.permits(UserAccess.Read)
|
||||
assert UserAccess.Full.permits(UserAccess.Unauthorized)
|
||||
|
||||
|
||||
def test_permits_reporter() -> None:
|
||||
"""
|
||||
reporter access must allow everything except full
|
||||
"""
|
||||
assert not UserAccess.Reporter.permits(UserAccess.Full)
|
||||
assert UserAccess.Reporter.permits(UserAccess.Reporter)
|
||||
assert UserAccess.Reporter.permits(UserAccess.Read)
|
||||
assert UserAccess.Reporter.permits(UserAccess.Unauthorized)
|
||||
|
||||
|
||||
def test_permits_read() -> None:
|
||||
"""
|
||||
read access must allow read only and unauthorized
|
||||
"""
|
||||
assert not UserAccess.Read.permits(UserAccess.Full)
|
||||
assert not UserAccess.Read.permits(UserAccess.Reporter)
|
||||
assert UserAccess.Read.permits(UserAccess.Read)
|
||||
assert UserAccess.Read.permits(UserAccess.Unauthorized)
|
||||
|
||||
|
||||
def test_permits_unauthorized() -> None:
|
||||
"""
|
||||
unauthorized access must only allow unauthorized
|
||||
"""
|
||||
assert not UserAccess.Unauthorized.permits(UserAccess.Full)
|
||||
assert not UserAccess.Unauthorized.permits(UserAccess.Reporter)
|
||||
assert not UserAccess.Unauthorized.permits(UserAccess.Read)
|
||||
assert UserAccess.Unauthorized.permits(UserAccess.Unauthorized)
|
||||
|
||||
|
||||
def test_permits_unknown(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must return False in case if input does not match
|
||||
"""
|
||||
mocker.patch.object(UserAccess, "_member_names_", ["Read"])
|
||||
assert not UserAccess.Full.permits(UserAccess.Unauthorized)
|
||||
|
@ -1,12 +1,16 @@
|
||||
import pytest
|
||||
|
||||
from asyncio import BaseEventLoop
|
||||
from aiohttp import web
|
||||
from aiohttp.test_utils import TestClient
|
||||
from collections import namedtuple
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import ahriman.core.auth.helpers
|
||||
|
||||
from ahriman.core.auth import OAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database import SQLite
|
||||
from ahriman.core.spawn import Spawn
|
||||
@ -105,3 +109,61 @@ def application_with_debug(configuration: Configuration, user: User, spawner: Sp
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", False)
|
||||
return setup_service("x86_64", configuration, spawner)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture
|
||||
|
||||
Args:
|
||||
application(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_oauth_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
application_with_auth["validator"] = MagicMock(spec=OAuth)
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
@ -1,6 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp import web
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
@ -63,11 +64,43 @@ async def test_auth_handler_api(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Read, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_static(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow static calls
|
||||
"""
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
await client_with_auth.get("/static/favicon.ico")
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_unauthorized(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow pages with unauthorized access
|
||||
"""
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
await client_with_auth.get("/")
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_allow_read_only(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow pages with allow read only flag
|
||||
"""
|
||||
aiohttp_request = pytest.helpers.request("", "/status-api", "GET")
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler(allow_read_only=True)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_api_no_method(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must ask for write permission if handler does not have get_permission method
|
||||
@ -77,9 +110,9 @@ async def test_auth_handler_api_no_method(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission = None
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_api_post(mocker: MockerFixture) -> None:
|
||||
@ -88,12 +121,12 @@ async def test_auth_handler_api_post(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
aiohttp_request = pytest.helpers.request("", "/status-api", "POST")
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Write
|
||||
request_handler.get_permission.return_value = UserAccess.Full
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_read(mocker: MockerFixture) -> None:
|
||||
@ -106,7 +139,7 @@ async def test_auth_handler_read(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Read, aiohttp_request.path)
|
||||
|
||||
@ -118,12 +151,12 @@ async def test_auth_handler_write(mocker: MockerFixture) -> None:
|
||||
for method in ("CONNECT", "DELETE", "PATCH", "POST", "PUT", "TRACE"):
|
||||
aiohttp_request = pytest.helpers.request("", "", method)
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Write
|
||||
request_handler.get_permission.return_value = UserAccess.Full
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
def test_setup_auth(application_with_auth: web.Application, auth: Auth, mocker: MockerFixture) -> None:
|
||||
|
@ -1,13 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp import web
|
||||
from asyncio import BaseEventLoop
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.auth import OAuth
|
||||
from ahriman.web.views.base import BaseView
|
||||
|
||||
|
||||
@ -23,61 +17,3 @@ def base(application: web.Application) -> BaseView:
|
||||
BaseView: generated base view fixture
|
||||
"""
|
||||
return BaseView(pytest.helpers.request(application, "", ""))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture
|
||||
|
||||
Args:
|
||||
application(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_oauth_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
application_with_auth["validator"] = MagicMock(spec=OAuth)
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AddView.get_permission(request) == UserAccess.Write
|
||||
assert await AddView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await RemoveView.get_permission(request) == UserAccess.Write
|
||||
assert await RemoveView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await RequestView.get_permission(request) == UserAccess.Read
|
||||
assert await RequestView.get_permission(request) == UserAccess.Reporter
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await SearchView.get_permission(request) == UserAccess.Read
|
||||
assert await SearchView.get_permission(request) == UserAccess.Reporter
|
||||
|
||||
|
||||
async def test_get(client: TestClient, aur_package_ahriman: AURPackage, mocker: MockerFixture) -> None:
|
||||
|
@ -1,65 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.user_access import UserAccess
|
||||
from ahriman.web.views.status.ahriman import AhrimanView
|
||||
|
||||
|
||||
async def test_get_permission() -> None:
|
||||
"""
|
||||
must return correct permission for the request
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AhrimanView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AhrimanView.get_permission(request) == UserAccess.Write
|
||||
|
||||
|
||||
async def test_get(client: TestClient) -> None:
|
||||
"""
|
||||
must return valid service status
|
||||
"""
|
||||
response = await client.get("/status-api/v1/ahriman")
|
||||
status = BuildStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
async def test_post(client: TestClient) -> None:
|
||||
"""
|
||||
must update service status correctly
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
post_response = await client.post("/status-api/v1/ahriman", json=payload)
|
||||
assert post_response.status == 204
|
||||
|
||||
response = await client.get("/status-api/v1/ahriman")
|
||||
status = BuildStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
async def test_post_exception(client: TestClient) -> None:
|
||||
"""
|
||||
must raise exception on invalid payload
|
||||
"""
|
||||
post_response = await client.post("/status-api/v1/ahriman", json={})
|
||||
assert post_response.status == 400
|
||||
|
||||
|
||||
async def test_post_exception_inside(client: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
exception handler must handle 500 errors
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||
|
||||
post_response = await client.post("/status-api/v1/ahriman", json=payload)
|
||||
assert post_response.status == 500
|
@ -17,7 +17,7 @@ async def test_get_permission() -> None:
|
||||
assert await PackageView.get_permission(request) == UserAccess.Read
|
||||
for method in ("DELETE", "POST"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await PackageView.get_permission(request) == UserAccess.Write
|
||||
assert await PackageView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
|
@ -18,7 +18,7 @@ async def test_get_permission() -> None:
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Write
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
|
@ -1,10 +1,12 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
import ahriman.version as version
|
||||
|
||||
from ahriman.models.build_status import BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.user_access import UserAccess
|
||||
from ahriman.web.views.status.status import StatusView
|
||||
@ -17,6 +19,9 @@ async def test_get_permission() -> None:
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await StatusView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await StatusView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
||||
@ -33,3 +38,37 @@ async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
||||
assert json["version"] == version.__version__
|
||||
assert json["packages"]
|
||||
assert json["packages"]["total"] == 1
|
||||
|
||||
|
||||
async def test_post(client: TestClient) -> None:
|
||||
"""
|
||||
must update service status correctly
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
post_response = await client.post("/status-api/v1/status", json=payload)
|
||||
assert post_response.status == 204
|
||||
|
||||
response = await client.get("/status-api/v1/status")
|
||||
status = InternalStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
async def test_post_exception(client: TestClient) -> None:
|
||||
"""
|
||||
must raise exception on invalid payload
|
||||
"""
|
||||
post_response = await client.post("/status-api/v1/status", json={})
|
||||
assert post_response.status == 400
|
||||
|
||||
|
||||
async def test_post_exception_inside(client: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
exception handler must handle 500 errors
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||
|
||||
post_response = await client.post("/status-api/v1/status", json=payload)
|
||||
assert post_response.status == 500
|
||||
|
@ -12,7 +12,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await IndexView.get_permission(request) == UserAccess.Safe
|
||||
assert await IndexView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_get(client_with_auth: TestClient) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "POST"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await LoginView.get_permission(request) == UserAccess.Safe
|
||||
assert await LoginView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_get_default_validator(client_with_auth: TestClient) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await LogoutView.get_permission(request) == UserAccess.Safe
|
||||
assert await LogoutView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_post(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -14,7 +14,7 @@ client_secret = client_secret
|
||||
oauth_provider = GoogleClient
|
||||
oauth_scopes = https://www.googleapis.com/auth/userinfo.email
|
||||
salt = salt
|
||||
safe_build_status = no
|
||||
allow_read_only = no
|
||||
|
||||
[build]
|
||||
archbuild_flags =
|
||||
|
Reference in New Issue
Block a user