reorder tests

This commit is contained in:
2026-07-13 13:00:13 +03:00
parent 92ac036f78
commit 7db38ff9e4
800 changed files with 307 additions and 996 deletions
@@ -0,0 +1,122 @@
import argparse
import pytest
from pytest_mock import MockerFixture
from ahriman.application.handlers.web.web import Web
from ahriman.core.configuration import Configuration
from ahriman.core.repository import Repository
from ahriman.models.log_handler import LogHandler
def _default_args(args: argparse.Namespace) -> argparse.Namespace:
"""
default arguments for these test cases
Args:
args(argparse.Namespace): command line arguments fixture
Returns:
argparse.Namespace: generated arguments for these test cases
"""
args.parser = lambda: True
args.configuration = None # doesn't matter actually
args.force = False
args.log_handler = None
args.report = True
args.quiet = False
args.unsafe = False
return args
def test_run(args: argparse.Namespace, configuration: Configuration, repository: Repository,
mocker: MockerFixture) -> None:
"""
must run command
"""
args = _default_args(args)
mocker.patch("ahriman.core.repository.Repository.load", return_value=repository)
setup_mock = mocker.patch("ahriman.application.handlers.web.web.setup_server")
run_mock = mocker.patch("ahriman.application.handlers.web.web.run_server")
start_mock = mocker.patch("ahriman.core.spawn.Spawn.start")
trigger_mock = mocker.patch("ahriman.core.triggers.TriggerLoader.load")
stop_mock = mocker.patch("ahriman.core.spawn.Spawn.stop")
join_mock = mocker.patch("ahriman.core.spawn.Spawn.join")
_, repository_id = configuration.check_loaded()
mocker.patch("ahriman.application.handlers.handler.Handler.repositories_extract", return_value=[repository_id])
Web.run(args, repository_id, configuration, report=False)
setup_mock.assert_called_once_with(configuration, pytest.helpers.anyvar(int), [repository_id])
run_mock.assert_called_once_with(pytest.helpers.anyvar(int))
start_mock.assert_called_once_with()
trigger_mock.assert_called_once_with(repository_id, configuration)
trigger_mock().on_start.assert_called_once_with()
stop_mock.assert_called_once_with()
join_mock.assert_called_once_with()
def test_extract_arguments(args: argparse.Namespace, configuration: Configuration):
"""
must extract correct args
"""
expected = [
"--configuration", str(configuration.path),
]
probe = _default_args(args)
assert list(Web.extract_arguments(probe, configuration)) == expected
probe.force = True
expected.extend(["--force"])
assert list(Web.extract_arguments(probe, configuration)) == expected
probe.log_handler = LogHandler.Console
expected.extend(["--log-handler", probe.log_handler.value])
assert list(Web.extract_arguments(probe, configuration)) == expected
probe.quiet = True
expected.extend(["--quiet"])
assert list(Web.extract_arguments(probe, configuration)) == expected
probe.unsafe = True
expected.extend(["--unsafe"])
assert list(Web.extract_arguments(probe, configuration)) == expected
configuration.set_option("web", "wait_timeout", "60")
expected.extend(["--wait-timeout", "60"])
assert list(Web.extract_arguments(probe, configuration)) == expected
def test_extract_arguments_full(parser: argparse.ArgumentParser, configuration: Configuration):
"""
must extract all available args except for blacklisted
"""
# append all options from parser
args = argparse.Namespace()
for action in parser._actions:
if action.default == argparse.SUPPRESS:
continue
# extract option from the following list
value = action.const or \
next(iter(action.choices or []), None) or \
(not action.default if isinstance(action.default, bool) else None) or \
(42 if action.type == int else None) or \
"random string"
if action.type is not None:
value = action.type(value)
setattr(args, action.dest, value)
assert list(Web.extract_arguments(args, configuration)) == [
"--configuration", str(configuration.path),
"--force",
"--log-handler", "console",
"--quiet",
"--unsafe",
]
def test_disallow_multi_architecture_run() -> None:
"""
must not allow multi architecture run
"""
assert not Web.ALLOW_MULTI_ARCHITECTURE_RUN
@@ -0,0 +1,54 @@
import pytest
from ahriman.core.auth.mapping import Mapping
from ahriman.core.auth.oauth import OAuth
from ahriman.core.auth.pam import PAM
from ahriman.core.configuration import Configuration
from ahriman.core.database import SQLite
@pytest.fixture
def mapping(configuration: Configuration, database: SQLite) -> Mapping:
"""
auth provider fixture
Args:
configuration(Configuration): configuration fixture
database(SQLite): database fixture
Returns:
Mapping: auth service instance
"""
return Mapping(configuration, database)
@pytest.fixture
def oauth(configuration: Configuration, database: SQLite) -> OAuth:
"""
OAuth provider fixture
Args:
configuration(Configuration): configuration fixture
database(SQLite): database fixture
Returns:
OAuth: OAuth2 service instance
"""
configuration.set("web", "address", "https://example.com")
return OAuth(configuration, database)
@pytest.fixture
def pam(configuration: Configuration, database: SQLite) -> PAM:
"""
PAM provider fixture
Args:
configuration(Configuration): configuration fixture
database(SQLite): database fixture
Returns:
PAM: PAM service instance
"""
configuration.set_option("auth", "full_access_group", "wheel")
return PAM(configuration, database)
@@ -0,0 +1,92 @@
from ahriman.core.auth import Auth
from ahriman.core.auth.mapping import Mapping
from ahriman.core.auth.oauth import OAuth
from ahriman.core.auth.pam import PAM
from ahriman.core.configuration import Configuration
from ahriman.core.database import SQLite
from ahriman.models.user import User
from ahriman.models.user_access import UserAccess
def test_auth_control(auth: Auth) -> None:
"""
must return a control for authorization
"""
assert auth.auth_control
assert "button" in auth.auth_control # I think it should be a button
def test_is_external(auth: Auth) -> None:
"""
must not be external provider
"""
assert not auth.is_external
def test_load_dummy(configuration: Configuration, database: SQLite) -> None:
"""
must load dummy validator if authorization is not enabled
"""
configuration.set_option("auth", "target", "disabled")
auth = Auth.load(configuration, database)
assert isinstance(auth, Auth)
def test_load_dummy_empty(configuration: Configuration, database: SQLite) -> None:
"""
must load dummy validator if no option set
"""
auth = Auth.load(configuration, database)
assert isinstance(auth, Auth)
def test_load_mapping(configuration: Configuration, database: SQLite) -> None:
"""
must load mapping validator if option set
"""
configuration.set_option("auth", "target", "configuration")
auth = Auth.load(configuration, database)
assert isinstance(auth, Mapping)
def test_load_oauth(configuration: Configuration, database: SQLite) -> None:
"""
must load OAuth2 validator if option set
"""
configuration.set_option("auth", "target", "oauth")
configuration.set_option("web", "address", "https://example.com")
auth = Auth.load(configuration, database)
assert isinstance(auth, OAuth)
def test_load_pam(configuration: Configuration, database: SQLite) -> None:
"""
must load pam validator if option set
"""
configuration.set_option("auth", "target", "pam")
configuration.set_option("auth", "full_access_group", "wheel")
auth = Auth.load(configuration, database)
assert isinstance(auth, PAM)
async def test_check_credentials(auth: Auth, user: User) -> None:
"""
must pass any credentials
"""
assert await auth.check_credentials(user.username, user.password)
assert await auth.check_credentials("", None)
async def test_known_username(auth: Auth, user: User) -> None:
"""
must allow any username
"""
assert await auth.known_username(user.username)
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.Full, None)
@@ -0,0 +1,123 @@
import importlib
import sys
from pytest_mock import MockerFixture
from ahriman.core.auth import helpers
def test_import_aiohttp_security() -> None:
"""
must import aiohttp_security correctly
"""
assert helpers.aiohttp_security
def test_import_aiohttp_session() -> None:
"""
must import aiohttp_session correctly
"""
assert helpers.aiohttp_session
async def test_authorized_userid_dummy(mocker: MockerFixture) -> None:
"""
must not call authorized_userid from library if not enabled
"""
mocker.patch.object(helpers, "aiohttp_security", None)
await helpers.authorized_userid()
async def test_authorized_userid_library(mocker: MockerFixture) -> None:
"""
must call authorized_userid from library if enabled
"""
authorized_userid_mock = mocker.patch("aiohttp_security.authorized_userid")
await helpers.authorized_userid()
authorized_userid_mock.assert_called_once_with()
async def test_check_authorized_dummy(mocker: MockerFixture) -> None:
"""
must not call check_authorized from library if not enabled
"""
mocker.patch.object(helpers, "aiohttp_security", None)
await helpers.check_authorized()
async def test_check_authorized_library(mocker: MockerFixture) -> None:
"""
must call check_authorized from library if enabled
"""
check_authorized_mock = mocker.patch("aiohttp_security.check_authorized")
await helpers.check_authorized()
check_authorized_mock.assert_called_once_with()
async def test_forget_dummy(mocker: MockerFixture) -> None:
"""
must not call forget from library if not enabled
"""
mocker.patch.object(helpers, "aiohttp_security", None)
await helpers.forget()
async def test_get_session_dummy(mocker: MockerFixture) -> None:
"""
must return empty dict if no aiohttp_session module found
"""
mocker.patch.object(helpers, "aiohttp_session", None)
assert await helpers.get_session() == {}
async def test_get_session_library(mocker: MockerFixture) -> None:
"""
must call get_session from library if enabled
"""
get_session_mock = mocker.patch("aiohttp_session.get_session")
await helpers.get_session()
get_session_mock.assert_called_once_with()
async def test_forget_library(mocker: MockerFixture) -> None:
"""
must call forget from library if enabled
"""
forget_mock = mocker.patch("aiohttp_security.forget")
await helpers.forget()
forget_mock.assert_called_once_with()
async def test_remember_dummy(mocker: MockerFixture) -> None:
"""
must not call remember from library if not enabled
"""
mocker.patch.object(helpers, "aiohttp_security", None)
await helpers.remember()
async def test_remember_library(mocker: MockerFixture) -> None:
"""
must call remember from library if enabled
"""
remember_mock = mocker.patch("aiohttp_security.remember")
await helpers.remember()
remember_mock.assert_called_once_with()
def test_import_aiohttp_security_missing(mocker: MockerFixture) -> None:
"""
must set missing flag if no aiohttp_security module found
"""
mocker.patch.dict(sys.modules, {"aiohttp_security": None})
importlib.reload(helpers)
assert helpers.aiohttp_security is None
def test_import_aiohttp_session_missing(mocker: MockerFixture) -> None:
"""
must set missing flag if no aiohttp_session module found
"""
mocker.patch.dict(sys.modules, {"aiohttp_session": None})
importlib.reload(helpers)
assert helpers.aiohttp_session is None
@@ -0,0 +1,79 @@
from pytest_mock import MockerFixture
from ahriman.core.auth.mapping import Mapping
from ahriman.models.user import User
from ahriman.models.user_access import UserAccess
async def test_check_credentials(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must return true for valid credentials
"""
current_password = user.password
user = user.hash_password(mapping.salt)
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
assert await mapping.check_credentials(user.username, current_password)
# here password is hashed so it is invalid
assert not await mapping.check_credentials(user.username, user.password)
async def test_check_credentials_empty(mapping: Mapping) -> None:
"""
must reject on empty credentials
"""
assert not await mapping.check_credentials("", None)
async def test_check_credentials_unknown(mapping: Mapping, user: User) -> None:
"""
must reject on unknown user
"""
assert not await mapping.check_credentials(user.username, user.password)
async def test_get_user(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must return user from storage by username
"""
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
assert await mapping.get_user(user.username) == user
async def test_get_user_normalized(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must return user from storage by username case-insensitive
"""
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
assert await mapping.get_user(user.username.upper()) == user
async def test_get_user_unknown(mapping: Mapping, user: User) -> None:
"""
must return None in case if no user found
"""
assert await mapping.get_user(user.username) is None
async def test_known_username(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must allow only known users
"""
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
assert await mapping.known_username(user.username)
async def test_known_username_unknown(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must not allow unknown users
"""
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=None)
assert not await mapping.known_username(user.password)
async def test_verify_access(mapping: Mapping, user: User, mocker: MockerFixture) -> None:
"""
must verify user access
"""
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.Full, None)
@@ -0,0 +1,127 @@
import aioauth_client
import pytest
from pytest_mock import MockerFixture
from ahriman.core.auth.oauth import OAuth
from ahriman.core.exceptions import OptionError
def test_auth_control(oauth: OAuth) -> None:
"""
must return a control for authorization
"""
assert oauth.auth_control
assert "<a" in oauth.auth_control # I think it should be a link
def test_is_external(oauth: OAuth) -> None:
"""
must be external provider
"""
assert oauth.is_external
def test_get_provider() -> None:
"""
must return valid provider type
"""
assert OAuth.get_provider("OAuth2Client") == aioauth_client.OAuth2Client
assert OAuth.get_provider("GoogleClient") == aioauth_client.GoogleClient
assert OAuth.get_provider("GithubClient") == aioauth_client.GithubClient
def test_get_provider_not_a_type() -> None:
"""
must raise an exception if attribute is not a type
"""
with pytest.raises(OptionError):
OAuth.get_provider("RANDOM")
def test_get_provider_invalid_type() -> None:
"""
must raise an exception if attribute is not an OAuth2 client
"""
with pytest.raises(OptionError):
OAuth.get_provider("User")
with pytest.raises(OptionError):
OAuth.get_provider("OAuth1Client")
def test_get_client(oauth: OAuth) -> None:
"""
must return valid OAuth2 client
"""
client = oauth.get_client()
assert isinstance(client, aioauth_client.GoogleClient)
assert client.client_id == oauth.client_id
assert client.client_secret == oauth.client_secret
def test_get_oauth_url(oauth: OAuth, mocker: MockerFixture) -> None:
"""
must generate valid OAuth authorization URL
"""
authorize_url_mock = mocker.patch("aioauth_client.GoogleClient.get_authorize_url")
oauth.get_oauth_url(state="state")
authorize_url_mock.assert_called_once_with(scope=oauth.scopes, redirect_uri=oauth.redirect_uri, state="state")
async def test_get_oauth_username(oauth: OAuth, mocker: MockerFixture) -> None:
"""
must return authorized user ID
"""
access_token_mock = mocker.patch("aioauth_client.GoogleClient.get_access_token", return_value=("token", ""))
user_info_mock = mocker.patch("aioauth_client.GoogleClient.user_info",
return_value=(aioauth_client.User(email="email"), ""))
assert await oauth.get_oauth_username("code", state="state", session={"state": "state"}) == "email"
access_token_mock.assert_called_once_with("code", redirect_uri=oauth.redirect_uri)
user_info_mock.assert_called_once_with()
async def test_get_oauth_username_empty_email(oauth: OAuth, mocker: MockerFixture) -> None:
"""
must read username if email is not available
"""
mocker.patch("aioauth_client.GoogleClient.get_access_token", return_value=("token", ""))
mocker.patch("aioauth_client.GoogleClient.user_info", return_value=(aioauth_client.User(username="username"), ""))
assert await oauth.get_oauth_username("code", state="state", session={"state": "state"}) == "username"
async def test_get_oauth_username_exception_1(oauth: OAuth, mocker: MockerFixture) -> None:
"""
must return None in case of OAuth request error (get_access_token)
"""
mocker.patch("aioauth_client.GoogleClient.get_access_token", side_effect=Exception)
user_info_mock = mocker.patch("aioauth_client.GoogleClient.user_info")
assert await oauth.get_oauth_username("code", state="state", session={"state": "state"}) is None
user_info_mock.assert_not_called()
async def test_get_oauth_username_exception_2(oauth: OAuth, mocker: MockerFixture) -> None:
"""
must return None in case of OAuth request error (user_info)
"""
mocker.patch("aioauth_client.GoogleClient.get_access_token", return_value=("token", ""))
mocker.patch("aioauth_client.GoogleClient.user_info", side_effect=Exception)
username = await oauth.get_oauth_username("code", state="state", session={"state": "state"})
assert username is None
async def test_get_oauth_username_csrf_missing(oauth: OAuth) -> None:
"""
must return None if CSRF state is missing
"""
assert await oauth.get_oauth_username("code", state=None, session={"state": "state"}) is None
async def test_get_oauth_username_csrf_mismatch(oauth: OAuth) -> None:
"""
must return None if CSRF state does not match session
"""
assert await oauth.get_oauth_username("code", state="wrong", session={"state": "state"}) is None
@@ -0,0 +1,118 @@
from pytest_mock import MockerFixture
from ahriman.core.auth.pam import PAM
from ahriman.core.exceptions import CalledProcessError
from ahriman.models.user import User
from ahriman.models.user_access import UserAccess
def test_group_members() -> None:
"""
must return current group members
"""
assert "root" in PAM.group_members("root")
def test_group_members_unknown() -> None:
"""
must return empty list for unknown group
"""
assert not PAM.group_members("somerandomgroupname")
async def test_check_credentials(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must correctly check user credentials via PAM
"""
authenticate_mock = mocker.patch("ahriman.core.auth.pam.check_output")
mapping_mock = mocker.patch("ahriman.core.auth.mapping.Mapping.check_credentials")
assert await pam.check_credentials(user.username, user.password)
authenticate_mock.assert_called_once_with("su", "--command", "true", "-", user.username,
input_data=user.password)
mapping_mock.assert_not_called()
async def test_check_credentials_empty(pam: PAM) -> None:
"""
must reject on empty credentials
"""
assert not await pam.check_credentials("", None)
async def test_check_credentials_root(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must reject on root logon attempt
"""
mocker.patch("ahriman.core.auth.pam.check_output")
assert not await pam.check_credentials("root", user.password)
pam.permit_root_login = True
assert await pam.check_credentials("root", user.password)
async def test_check_credentials_mapping(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must correctly check user credentials via database if PAM rejected
"""
mocker.patch("ahriman.core.auth.pam.check_output",
side_effect=CalledProcessError(1, ["command"], "error"))
mapping_mock = mocker.patch("ahriman.core.auth.mapping.Mapping.check_credentials")
await pam.check_credentials(user.username, user.password)
mapping_mock.assert_called_once_with(pam, user.username, user.password)
async def test_known_username(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must check if user exists in system
"""
getpwnam_mock = mocker.patch("ahriman.core.auth.pam.getpwnam")
mapping_mock = mocker.patch("ahriman.core.auth.mapping.Mapping.known_username")
assert await pam.known_username(user.username)
getpwnam_mock.assert_called_once_with(user.username)
mapping_mock.assert_not_called()
async def test_known_username_mapping(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must fall back to username checking to database if no user found in system
"""
mocker.patch("ahriman.core.auth.pam.getpwnam", side_effect=KeyError)
mapping_mock = mocker.patch("ahriman.core.auth.mapping.Mapping.known_username")
await pam.known_username(user.username)
mapping_mock.assert_called_once_with(pam, user.username)
async def test_verify_access(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must verify user access via PAM groups
"""
mocker.patch("ahriman.core.auth.pam.PAM.get_user", return_value=None)
mocker.patch("ahriman.core.auth.pam.PAM.group_members", return_value=[user.username])
assert await pam.verify_access(user.username, UserAccess.Full, None)
async def test_verify_access_readonly(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must set user access to read only if it doesn't belong to the admin group
"""
mocker.patch("ahriman.core.auth.pam.PAM.get_user", return_value=None)
mocker.patch("ahriman.core.auth.pam.PAM.group_members", return_value=[])
assert not await pam.verify_access(user.username, UserAccess.Full, None)
assert not await pam.verify_access(user.username, UserAccess.Reporter, None)
assert await pam.verify_access(user.username, UserAccess.Read, None)
async def test_verify_access_override(pam: PAM, user: User, mocker: MockerFixture) -> None:
"""
must verify user access via database if there is override
"""
mocker.patch("ahriman.core.auth.pam.PAM.get_user", return_value=user)
group_mock = mocker.patch("ahriman.core.auth.pam.PAM.group_members")
assert await pam.verify_access(user.username, user.access, None)
group_mock.assert_not_called()
@@ -0,0 +1,24 @@
import importlib
import sys
from pytest_mock import MockerFixture
from ahriman.web import apispec
def test_import_apispec() -> None:
"""
must correctly import apispec
"""
assert apispec.aiohttp_apispec
def test_import_apispec_missing(mocker: MockerFixture) -> None:
"""
must correctly process missing module
"""
mocker.patch.dict(sys.modules, {"aiohttp_apispec": None})
importlib.reload(apispec)
assert apispec.aiohttp_apispec is None
assert apispec.Schema
assert apispec.fields("arg", kwargs=42)
@@ -0,0 +1,161 @@
from aiohttp.web import HTTPFound
from pytest_mock import MockerFixture
from unittest.mock import MagicMock
from ahriman.models.user_access import UserAccess
from ahriman.web.apispec import decorators
from ahriman.web.apispec.decorators import _response_schema, apidocs
from ahriman.web.schemas import LoginSchema
def test_response_schema() -> None:
"""
must generate response schema
"""
schema = _response_schema(None)
assert schema.pop(204)
assert schema.pop(401)
assert schema.pop(403)
assert schema.pop(500)
def test_response_schema_no_403() -> None:
"""
must generate response schema without 403 error
"""
schema = _response_schema(None, error_403_enabled=False)
assert 403 not in schema
def test_response_schema_400() -> None:
"""
must generate response schema with 400 error
"""
schema = _response_schema(None, error_400_enabled=True)
assert schema.pop(400)
def test_response_schema_404() -> None:
"""
must generate response schema with 404 error
"""
schema = _response_schema(None, error_404_description="description")
assert schema.pop(404)
def test_response_schema_200() -> None:
"""
must generate response schema with 200 response
"""
schema = _response_schema(LoginSchema)
response = schema.pop(200)
assert response["schema"] == LoginSchema
assert 204 not in schema
def test_response_schema_code() -> None:
"""
must override status code
"""
schema = _response_schema(None, response_code=HTTPFound)
assert schema.pop(302)
assert 204 not in schema
def test_apidocs() -> None:
"""
must return decorated function
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
)(MagicMock())
assert annotated.__apispec__
def test_apidocs_authorization() -> None:
"""
must return decorated function with authorization details
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Full,
)(MagicMock())
assert any(schema["put_into"] == "cookies" for schema in annotated.__schemas__)
def test_apidocs_match() -> None:
"""
must return decorated function with match details
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
match_schema=LoginSchema,
)(MagicMock())
assert any(schema["put_into"] == "match_info" for schema in annotated.__schemas__)
def test_apidocs_querystring() -> None:
"""
must return decorated function with query string details
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
query_schema=LoginSchema,
)(MagicMock())
assert any(schema["put_into"] == "querystring" for schema in annotated.__schemas__)
def test_apidocs_json() -> None:
"""
must return decorated function with json details
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
body_schema=LoginSchema,
)(MagicMock())
assert any(schema["put_into"] == "json" for schema in annotated.__schemas__)
def test_apidocs_form() -> None:
"""
must return decorated function with generic body details
"""
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
body_schema=LoginSchema,
body_location="form",
)(MagicMock())
assert any(schema["put_into"] == "form" for schema in annotated.__schemas__)
def test_apidocs_import_error(mocker: MockerFixture) -> None:
"""
must return same function if no apispec module available
"""
mocker.patch.object(decorators, "aiohttp_apispec", None)
mock = MagicMock()
annotated = apidocs(
tags=["tags"],
summary="summary",
description="description",
permission=UserAccess.Unauthorized,
)(mock)
assert annotated == mock
@@ -0,0 +1,67 @@
import pytest
from aiohttp.web import Application
from pytest_mock import MockerFixture
from ahriman import __version__
from ahriman.web.apispec import info
from ahriman.web.apispec.info import _info, _security, _servers, setup_apispec
from ahriman.web.keys import ConfigurationKey
def test_info() -> None:
"""
must generate info object for swagger
"""
info = _info()
assert info["title"] == "ahriman"
assert info["version"] == __version__
def test_security() -> None:
"""
must generate security definitions for swagger
"""
token = next(iter(_security()))["token"]
assert token == {"type": "apiKey", "name": "AHRIMAN", "in": "cookie"}
def test_servers(application: Application) -> None:
"""
must generate servers definitions
"""
servers = _servers(application)
assert servers == [{"url": "http://127.0.0.1:8080"}]
def test_servers_address(application: Application) -> None:
"""
must generate servers definitions with address
"""
application[ConfigurationKey].set_option("web", "address", "https://example.com")
servers = _servers(application)
assert servers == [{"url": "https://example.com"}]
def test_setup_apispec(application: Application, mocker: MockerFixture) -> None:
"""
must set api specification
"""
apispec_mock = mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
assert setup_apispec(application)
apispec_mock.assert_called_once_with(
application,
url="/api-docs/swagger.json",
openapi_version="3.0.2",
info=pytest.helpers.anyvar(int),
servers=pytest.helpers.anyvar(int),
security=pytest.helpers.anyvar(int),
)
def test_setup_apispec_import_error(application: Application, mocker: MockerFixture) -> None:
"""
must return none if apispec is not available
"""
mocker.patch.object(info, "aiohttp_apispec", None)
assert setup_apispec(application) is None
+220
View File
@@ -0,0 +1,220 @@
import pytest
import pytest_asyncio
from aiohttp.test_utils import TestClient
from aiohttp.web import Application, Resource, UrlMappingMatchInfo
from collections.abc import Awaitable, Callable
from marshmallow import Schema
from pytest_mock import MockerFixture
from typing import Any
from unittest.mock import MagicMock, Mock
from ahriman.core.auth import helpers
from ahriman.core.auth.oauth import OAuth
from ahriman.core.configuration import Configuration
from ahriman.core.database import SQLite
from ahriman.core.spawn import Spawn
from ahriman.models.user import User
from ahriman.web.keys import AuthKey
from ahriman.web.web import setup_server
@pytest.helpers.register
def patch_view(application: Application, attribute: str, mock: Mock) -> Mock:
"""
patch given attribute in views. This method is required because of dynamic load
Args:
application(Application): application fixture
attribute(str): attribute name to patch
mock(Mock): mock object
Returns:
Mock: mock set to object
"""
for route in application.router.routes():
if hasattr(route.handler, attribute):
setattr(route.handler, attribute, mock)
return mock
@pytest.helpers.register
def request(application: Application, path: str, method: str, params: Any = None, json: Any = None, data: Any = None,
extra: dict[str, Any] | None = None, resource: Resource | None = None) -> MagicMock:
"""
request generator helper
Args:
application(Application): application fixture
path(str): path for the request
method(str): method for the request
params(Any, optional): query parameters (Default value = None)
json(Any, optional): json payload of the request (Default value = None)
data(Any, optional): form data payload of the request (Default value = None)
extra(dict[str, Any] | None, optional): extra info which will be injected for ``get_extra_info`` command
resource(Resource | None, optional): optional web resource for the request (Default value = None)
Returns:
MagicMock: dummy request mock
"""
request_mock = MagicMock()
request_mock.app = application
request_mock.path = path
request_mock.method = method
request_mock.query = params
request_mock.json = json
request_mock.post = data
if resource is not None:
route_mock = MagicMock()
route_mock.resource = resource
request_mock.match_info = UrlMappingMatchInfo({}, route_mock)
extra = extra or {}
request_mock.get_extra_info.side_effect = extra.get
return request_mock
@pytest.helpers.register
def schema_request(handler: Callable[..., Awaitable[Any]], *, location: str = "json") -> Schema:
"""
extract request schema from docs
Args:
handler(Callable[[], Awaitable[Any]]): request handler
location(str, optional): location of the request (Default value = "json")
Returns:
Schema: request schema as set by the decorators
"""
schemas: list[dict[str, Any]] = handler.__schemas__ # type: ignore[attr-defined]
return next(schema["schema"] for schema in schemas if schema["put_into"] == location)
@pytest.helpers.register
def schema_response(handler: Callable[..., Awaitable[Any]], *, code: int = 200) -> Schema:
"""
extract response schema from docs
Args:
handler(Callable[[], Awaitable[Any]]): request handler
code(int, optional): return code of the request (Default value = 200)
Returns:
Schema: response schema as set by the decorators
"""
schemas: dict[int, Any] = handler.__apispec__["responses"] # type: ignore[attr-defined]
schema = schemas[code]["schema"]
if callable(schema):
schema = schema()
return schema
@pytest.fixture
def application(configuration: Configuration, spawner: Spawn, database: SQLite, mocker: MockerFixture) -> Application:
"""
application fixture
Args:
configuration(Configuration): configuration fixture
spawner(Spawn): spawner fixture
database(SQLite): database fixture
mocker(MockerFixture): mocker object
Returns:
Application: application test instance
"""
configuration.set_option("web", "port", "8080")
mocker.patch("ahriman.core.configuration.Configuration.from_path", return_value=configuration)
mocker.patch("ahriman.core.database.SQLite.load", return_value=database)
mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
mocker.patch.object(helpers, "aiohttp_security", None)
_, repository_id = configuration.check_loaded()
return setup_server(configuration, spawner, [repository_id])
@pytest.fixture
def application_with_auth(configuration: Configuration, user: User, spawner: Spawn, database: SQLite,
mocker: MockerFixture) -> Application:
"""
application fixture with auth enabled
Args:
configuration(Configuration): configuration fixture
user(User): user descriptor fixture
spawner(Spawn): spawner fixture
database(SQLite): database fixture
mocker(MockerFixture): mocker object
Returns:
Application: application test instance
"""
configuration.set_option("auth", "target", "configuration")
configuration.set_option("web", "port", "8080")
mocker.patch("ahriman.core.configuration.Configuration.from_path", return_value=configuration)
mocker.patch("ahriman.core.database.SQLite.load", return_value=database)
mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
_, repository_id = configuration.check_loaded()
application = setup_server(configuration, spawner, [repository_id])
generated = user.hash_password(application[AuthKey].salt)
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=generated)
return application
@pytest_asyncio.fixture
async def client(application: Application, aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
"""
web client fixture
Args:
application(Application): application fixture
aiohttp_client(Any): aiohttp client fixture
mocker(MockerFixture): mocker object
Returns:
TestClient: web client test instance
"""
mocker.patch("pathlib.Path.iterdir", return_value=[])
return await aiohttp_client(application)
@pytest_asyncio.fixture
async def client_with_auth(application_with_auth: Application, aiohttp_client: Any,
mocker: MockerFixture) -> TestClient:
"""
web client fixture with full authorization functions
Args:
application_with_auth(Application): application fixture
aiohttp_client(Any): aiohttp client fixture
mocker(MockerFixture): mocker object
Returns:
TestClient: web client test instance
"""
mocker.patch("pathlib.Path.iterdir", return_value=[])
return await aiohttp_client(application_with_auth)
@pytest_asyncio.fixture
async def client_with_oauth_auth(application_with_auth: Application, aiohttp_client: Any,
mocker: MockerFixture) -> TestClient:
"""
web client fixture with full authorization functions
Args:
application_with_auth(Application): application fixture
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[AuthKey] = MagicMock(spec=OAuth)
return await aiohttp_client(application_with_auth)
@@ -0,0 +1,24 @@
import pytest
from ahriman.core.auth import Auth
from ahriman.core.configuration import Configuration
from ahriman.core.database import SQLite
from ahriman.web.middlewares.auth_handler import _AuthorizationPolicy
@pytest.fixture
def authorization_policy(configuration: Configuration, database: SQLite) -> _AuthorizationPolicy:
"""
fixture for authorization policy
Args:
configuration(Configuration): configuration fixture
database(SQLite): database fixture
Returns:
AuthorizationPolicy: authorization policy fixture
"""
configuration.set_option("auth", "target", "configuration")
validator = Auth.load(configuration, database)
policy = _AuthorizationPolicy(validator)
return policy
@@ -0,0 +1,197 @@
import pytest
import socket
from aiohttp.test_utils import TestClient
from aiohttp.web import Application
from cryptography import fernet
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock, call as MockCall
from ahriman.core.auth import Auth
from ahriman.core.configuration import Configuration
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.user import User
from ahriman.models.user_access import UserAccess
from ahriman.web.keys import AuthKey
from ahriman.web.middlewares.auth_handler import _AuthorizationPolicy, _auth_handler, _cookie_secret_key, setup_auth
async def test_authorized_userid(authorization_policy: _AuthorizationPolicy, user: User, mocker: MockerFixture) -> None:
"""
must return authorized user id
"""
mocker.patch("ahriman.core.database.SQLite.user_get", return_value=user)
assert await authorization_policy.authorized_userid(user.username) == user.username
async def test_authorized_userid_unknown(authorization_policy: _AuthorizationPolicy) -> None:
"""
must not allow unknown user id for authorization
"""
assert await authorization_policy.authorized_userid("somerandomname") is None
assert await authorization_policy.authorized_userid("somerandomname") is None
async def test_permits(authorization_policy: _AuthorizationPolicy, user: User) -> None:
"""
must call validator check
"""
authorization_policy.validator = AsyncMock()
authorization_policy.validator.verify_access.side_effect = lambda username, *args: username == user.username
assert await authorization_policy.permits(user.username, user.access, "/endpoint")
assert not await authorization_policy.permits("somerandomname", user.access, "/endpoint")
assert not await authorization_policy.permits(None, user.access, "/endpoint")
assert not await authorization_policy.permits(user.username, "random", "/endpoint")
assert not await authorization_policy.permits(None, BuildStatusEnum.Building, "/endpoint")
authorization_policy.validator.verify_access.assert_has_calls([
MockCall(user.username, user.access, "/endpoint"),
MockCall("somerandomname", user.access, "/endpoint"),
])
async def test_auth_handler_unix_socket(mocker: MockerFixture) -> None:
"""
must allow calls via unix sockets
"""
aiohttp_request = pytest.helpers.request(
"", "/api/v1/status", "GET", extra={"socket": socket.socket(socket.AF_UNIX)})
request_handler = AsyncMock()
request_handler.get_permission.return_value = UserAccess.Full
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
handler = _auth_handler(allow_read_only=False)
await handler(aiohttp_request, request_handler)
check_permission_mock.assert_not_called()
async def test_auth_handler_api(mocker: MockerFixture) -> None:
"""
must ask for status permission for api calls
"""
aiohttp_request = pytest.helpers.request("", "/api/v1/status", "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=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("", "/api/v1/status", "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
"""
aiohttp_request = pytest.helpers.request("", "/api/v1/status", "GET")
request_handler = AsyncMock()
request_handler.get_permission = None
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
handler = _auth_handler(allow_read_only=False)
await handler(aiohttp_request, request_handler)
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
async def test_auth_handler_api_post(mocker: MockerFixture) -> None:
"""
must ask for status permission for api calls with POST
"""
aiohttp_request = pytest.helpers.request("", "/api/v1/status", "POST")
request_handler = AsyncMock()
request_handler.get_permission.return_value = UserAccess.Full
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
handler = _auth_handler(allow_read_only=False)
await handler(aiohttp_request, request_handler)
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
async def test_auth_handler_read(mocker: MockerFixture) -> None:
"""
must ask for read permission for api calls with GET
"""
for method in ("GET", "HEAD", "OPTIONS"):
aiohttp_request = pytest.helpers.request("", "", method)
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=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_write(mocker: MockerFixture) -> None:
"""
must ask for read permission for api calls with POST
"""
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.Full
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
handler = _auth_handler(allow_read_only=False)
await handler(aiohttp_request, request_handler)
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
def test_cookie_secret_key(configuration: Configuration) -> None:
"""
must generate fernet key
"""
secret_key = _cookie_secret_key(configuration)
assert isinstance(secret_key, fernet.Fernet)
def test_cookie_secret_key_cached(configuration: Configuration) -> None:
"""
must use cookie key as set by configuration
"""
configuration.set_option("auth", "cookie_secret_key", fernet.Fernet.generate_key().decode("utf8"))
assert _cookie_secret_key(configuration) is not None
def test_setup_auth(application_with_auth: Application, configuration: Configuration, auth: Auth,
mocker: MockerFixture) -> None:
"""
must set up authorization
"""
setup_mock = mocker.patch("aiohttp_security.setup")
application = setup_auth(application_with_auth, configuration, auth)
assert application.get(AuthKey) is not None
setup_mock.assert_called_once_with(application_with_auth, pytest.helpers.anyvar(int), pytest.helpers.anyvar(int))
@@ -0,0 +1,85 @@
import hashlib
import pytest
from aiohttp import ETag
from aiohttp.web import HTTPNotModified, Response, StreamResponse
from unittest.mock import AsyncMock
from ahriman.web.middlewares.etag_handler import etag_handler
async def test_etag_handler() -> None:
"""
must set ETag header on GET responses
"""
request = pytest.helpers.request("", "", "GET")
request.if_none_match = None
request_handler = AsyncMock(return_value=Response(body=b"hello"))
handler = etag_handler()
result = await handler(request, request_handler)
assert result.etag is not None
async def test_etag_handler_not_modified() -> None:
"""
must raise NotModified when ETag matches If-None-Match
"""
body = b"hello"
request = pytest.helpers.request("", "", "GET")
request.if_none_match = (ETag(value=hashlib.md5(body, usedforsecurity=False).hexdigest()),)
request_handler = AsyncMock(return_value=Response(body=body))
handler = etag_handler()
with pytest.raises(HTTPNotModified):
await handler(request, request_handler)
async def test_etag_handler_no_match() -> None:
"""
must return full response when ETag does not match If-None-Match
"""
request = pytest.helpers.request("", "", "GET")
request.if_none_match = (ETag(value="outdated"),)
request_handler = AsyncMock(return_value=Response(body=b"hello"))
handler = etag_handler()
result = await handler(request, request_handler)
assert result.status == 200
assert result.etag is not None
async def test_etag_handler_skip_post() -> None:
"""
must skip ETag for non-GET/HEAD methods
"""
request = pytest.helpers.request("", "", "POST")
request_handler = AsyncMock(return_value=Response(body=b"hello"))
handler = etag_handler()
result = await handler(request, request_handler)
assert result.etag is None
async def test_etag_handler_skip_no_body() -> None:
"""
must skip ETag for responses without body
"""
request = pytest.helpers.request("", "", "GET")
request_handler = AsyncMock(return_value=Response())
handler = etag_handler()
result = await handler(request, request_handler)
assert result.etag is None
async def test_etag_handler_skip_stream() -> None:
"""
must skip ETag for streaming responses
"""
request = pytest.helpers.request("", "", "GET")
request_handler = AsyncMock(return_value=StreamResponse())
handler = etag_handler()
result = await handler(request, request_handler)
assert "ETag" not in result.headers
@@ -0,0 +1,191 @@
import json
import logging
import pytest
from aiohttp.web import HTTPBadRequest, HTTPInternalServerError, HTTPMethodNotAllowed, HTTPNoContent, HTTPUnauthorized
from pytest_mock import MockerFixture
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from ahriman.web.middlewares.exception_handler import _is_templated_unauthorized, exception_handler
def _extract_body(response: Any) -> Any:
"""
extract json body from given object
Args:
response(Any): response (any actually) object
Returns:
Any: body key from the object converted to json
"""
return json.loads(getattr(response, "body"))
def test_is_templated_unauthorized() -> None:
"""
must correct check if response should be rendered as template
"""
response_mock = MagicMock()
response_mock.path = "/api/v1/login"
response_mock.headers.getall.return_value = ["*/*"]
assert _is_templated_unauthorized(response_mock)
response_mock.path = "/api/v1/login"
response_mock.headers.getall.return_value = ["application/json"]
assert not _is_templated_unauthorized(response_mock)
response_mock.path = "/api/v1/logout"
response_mock.headers.getall.return_value = ["*/*"]
assert _is_templated_unauthorized(response_mock)
response_mock.path = "/api/v1/logout"
response_mock.headers.getall.return_value = ["application/json"]
assert not _is_templated_unauthorized(response_mock)
response_mock.path = "/api/v1/status"
response_mock.headers.getall.return_value = ["*/*"]
assert not _is_templated_unauthorized(response_mock)
response_mock.path = "/api/v1/status"
response_mock.headers.getall.return_value = ["application/json"]
assert not _is_templated_unauthorized(response_mock)
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(request, request_handler)
logging_mock.assert_not_called()
async def test_exception_handler_success(mocker: MockerFixture) -> None:
"""
must pass 2xx and 3xx codes
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPNoContent)
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
with pytest.raises(HTTPNoContent):
await handler(request, request_handler)
logging_mock.assert_not_called()
async def test_exception_handler_unauthorized(mocker: MockerFixture) -> None:
"""
must handle unauthorized exception as json response
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPUnauthorized)
mocker.patch("ahriman.web.middlewares.exception_handler._is_templated_unauthorized", return_value=False)
render_mock = mocker.patch("aiohttp_jinja2.render_template")
handler = exception_handler(logging.getLogger())
response = await handler(request, request_handler)
assert _extract_body(response) == {"error": HTTPUnauthorized().reason}
render_mock.assert_not_called()
async def test_exception_handler_unauthorized_templated(mocker: MockerFixture) -> None:
"""
must handle unauthorized exception as json response in html context
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPUnauthorized)
mocker.patch("ahriman.web.middlewares.exception_handler._is_templated_unauthorized", return_value=True)
render_mock = mocker.patch("aiohttp_jinja2.render_template")
handler = exception_handler(logging.getLogger())
await handler(request, request_handler)
context = {"code": 401, "reason": "Unauthorized"}
render_mock.assert_called_once_with("error.jinja2", request, context, status=HTTPUnauthorized.status_code)
async def test_exception_handler_options() -> None:
"""
must handle OPTIONS request
"""
request = pytest.helpers.request("", "", "OPTIONS")
request_handler = AsyncMock(side_effect=HTTPMethodNotAllowed("OPTIONS", ["GET"]))
handler = exception_handler(logging.getLogger())
with pytest.raises(HTTPNoContent) as response:
await handler(request, request_handler)
assert response.headers["Allow"] == "GET"
async def test_exception_handler_head() -> None:
"""
must handle missing HEAD requests
"""
request = pytest.helpers.request("", "", "HEAD")
request_handler = AsyncMock(side_effect=HTTPMethodNotAllowed("HEAD", ["HEAD,GET"]))
handler = exception_handler(logging.getLogger())
with pytest.raises(HTTPMethodNotAllowed) as response:
await handler(request, request_handler)
assert response.headers["Allow"] == "GET"
async def test_exception_handler_method_not_allowed() -> None:
"""
must handle not allowed methods
"""
request = pytest.helpers.request("", "", "POST")
request_handler = AsyncMock(side_effect=HTTPMethodNotAllowed("POST", ["GET"]))
handler = exception_handler(logging.getLogger())
with pytest.raises(HTTPMethodNotAllowed):
await handler(request, request_handler)
async def test_exception_handler_client_error(mocker: MockerFixture) -> None:
"""
must handle client exception
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPBadRequest)
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
response = await handler(request, request_handler)
assert _extract_body(response) == {"error": HTTPBadRequest().reason}
logging_mock.assert_not_called()
async def test_exception_handler_server_error(mocker: MockerFixture) -> None:
"""
must handle server exception
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=HTTPInternalServerError)
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
response = await handler(request, request_handler)
assert _extract_body(response) == {"error": HTTPInternalServerError().reason}
logging_mock.assert_called_once() # we do not check logging arguments
async def test_exception_handler_unknown_error(mocker: MockerFixture) -> None:
"""
must log server exception and re-raise it
"""
request = pytest.helpers.request("", "", "")
request_handler = AsyncMock(side_effect=Exception("An error"))
logging_mock = mocker.patch("logging.Logger.exception")
handler = exception_handler(logging.getLogger())
response = await handler(request, request_handler)
assert _extract_body(response) == {"error": "An error"}
logging_mock.assert_called_once() # we do not check logging arguments
@@ -0,0 +1,59 @@
import importlib
import pytest
import sys
from aiohttp.web import HTTPNotFound
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
import ahriman.web.middlewares.metrics_handler as metrics_handler
async def test_metrics(mocker: MockerFixture) -> None:
"""
must return metrics methods if library is available
"""
metrics_mock = AsyncMock()
mocker.patch.object(metrics_handler, "aiohttp_openmetrics", metrics_mock)
await metrics_handler.metrics(42)
metrics_mock.metrics.assert_called_once_with(42)
async def test_metrics_dummy(mocker: MockerFixture) -> None:
"""
must raise HTTPNotFound if no module found
"""
mocker.patch.object(metrics_handler, "aiohttp_openmetrics", None)
with pytest.raises(HTTPNotFound):
await metrics_handler.metrics(None)
async def test_metrics_handler() -> None:
"""
must return metrics handler if library is available
"""
assert metrics_handler.metrics_handler() == metrics_handler.aiohttp_openmetrics.metrics_middleware
async def test_metrics_handler_dummy(mocker: MockerFixture) -> None:
"""
must return dummy handler if no module found
"""
mocker.patch.object(metrics_handler, "aiohttp_openmetrics", None)
handler = metrics_handler.metrics_handler()
async def handle(result: int) -> int:
return result
assert await handler(42, handle) == 42
def test_import_openmetrics_missing(mocker: MockerFixture) -> None:
"""
must correctly process missing module
"""
mocker.patch.dict(sys.modules, {"aiohttp_openmetrics": None})
importlib.reload(metrics_handler)
assert metrics_handler.aiohttp_openmetrics is None
@@ -0,0 +1,43 @@
import logging
import pytest
from unittest.mock import AsyncMock, MagicMock
from typing import Any
from ahriman.web.middlewares.request_id_handler import request_id_handler
async def test_request_id_handler() -> None:
"""
must use request id from request if available
"""
request = pytest.helpers.request("", "", "")
request.headers = MagicMock()
request.headers.getone.return_value = "request_id"
response = MagicMock()
response.headers = {}
async def check_handler(_: Any) -> MagicMock:
record = logging.makeLogRecord({})
assert record.request_id == "request_id"
return response
handler = request_id_handler()
await handler(request, check_handler)
assert response.headers["X-Request-ID"] == "request_id"
async def test_request_id_handler_generate() -> None:
"""
must generate request id and set it in response header
"""
request = pytest.helpers.request("", "", "")
response = MagicMock()
response.headers = {}
request_handler = AsyncMock(return_value=response)
handler = request_id_handler()
await handler(request, request_handler)
assert "X-Request-ID" in response.headers
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1,9 @@
from ahriman.web.schemas import AuthSchema
def test_schema() -> None:
"""
must return valid schema
"""
schema = AuthSchema()
assert not schema.validate({"AHRIMAN": "key"})
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1,10 @@
from ahriman.models.package import Package
from ahriman.web.schemas import PackageNameSchema
def test_schema(package_ahriman: Package) -> None:
"""
must return valid schema
"""
schema = PackageNameSchema()
assert not schema.validate({"package": package_ahriman.base})
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1 @@
# schema testing goes in view class tests
@@ -0,0 +1,55 @@
import aiohttp_cors
import pytest
from aiohttp.web import Application
from pytest_mock import MockerFixture
from ahriman.web.cors import setup_cors
from ahriman.web.keys import ConfigurationKey
def test_setup_cors(application: Application) -> None:
"""
must setup CORS
"""
cors = application[aiohttp_cors.APP_CONFIG_KEY]
# let's test here that it is enabled for all requests
for route in application.router.routes():
# we don't want to deal with match info here though
try:
url = route.url_for()
except (KeyError, TypeError):
continue
request = pytest.helpers.request(application, url, route.method, resource=route.resource)
assert cors._cors_impl._router_adapter.is_cors_enabled_on_request(request)
def test_setup_cors_custom_origins(application: Application, mocker: MockerFixture) -> None:
"""
must setup CORS with custom origins
"""
configuration = application[ConfigurationKey]
configuration.set_option("web", "cors_allow_origins", "https://example.com https://httpbin.com")
setup_mock = mocker.patch("ahriman.web.cors.aiohttp_cors.setup", return_value=mocker.MagicMock())
setup_cors(application, configuration)
defaults = setup_mock.call_args.kwargs["defaults"]
assert "https://example.com" in defaults
assert "https://httpbin.com" in defaults
assert "*" not in defaults
def test_setup_cors_custom_methods(application: Application, mocker: MockerFixture) -> None:
"""
must setup CORS with custom methods
"""
configuration = application[ConfigurationKey]
configuration.set_option("web", "cors_allow_methods", "GET POST")
setup_mock = mocker.patch("ahriman.web.cors.aiohttp_cors.setup", return_value=mocker.MagicMock())
setup_cors(application, configuration)
defaults = setup_mock.call_args.kwargs["defaults"]
resource_options = next(iter(defaults.values()))
assert resource_options.allow_methods == {"GET", "POST"}
@@ -0,0 +1,40 @@
from aiohttp.web import Application
from pathlib import Path
from ahriman.core.configuration import Configuration
from ahriman.core.utils import walk
from ahriman.web.routes import _dynamic_routes, _identifier, setup_routes
def test_dynamic_routes(resource_path_root: Path, configuration: Configuration) -> None:
"""
must return all available routes
"""
views_root = resource_path_root.parent.parent / "ahriman-web" / "src" / "ahriman" / "web" / "views"
expected_views = [
file
for file in walk(views_root)
if file.suffix == ".py" and file.name not in ("__init__.py", "base.py", "status_view_guard.py")
]
routes = dict(_dynamic_routes(configuration))
assert all(isinstance(view, type) for view in routes.values())
assert len(set(routes.values())) == len(expected_views)
def test_identifier() -> None:
"""
must correctly extract route identifiers
"""
assert _identifier("/") == "_"
assert _identifier("/api/v1/status") == "_api_v1_status"
assert _identifier("/api/v1/packages/{package}") == "_api_v1_packages_:package"
def test_setup_routes(application: Application, configuration: Configuration) -> None:
"""
must generate non-empty list of routes
"""
application.router._named_resources = {}
setup_routes(application, configuration)
assert application.router.routes()
@@ -0,0 +1,27 @@
import pytest
from aiohttp.web import Application
from ahriman import __version__
from ahriman.models.repository_id import RepositoryId
from ahriman.web.server_info import server_info
from ahriman.web.views.index import IndexView
async def test_server_info(application: Application, repository_id: RepositoryId) -> None:
"""
must generate server info
"""
request = pytest.helpers.request(application, "", "GET")
view = IndexView(request)
result = await server_info(view)
assert result["repositories"] == [{"id": repository_id.id, **repository_id.view()}]
assert not result["auth"]["enabled"]
assert not result["auth"]["external"]
assert result["auth"]["username"] is None
assert result["auth"]["control"]
assert result["version"] == __version__
assert result["autorefresh_intervals"] == []
assert result["docs_enabled"]
assert result["index_url"] is None
+162
View File
@@ -0,0 +1,162 @@
import pytest
import socket
from aiohttp.web import Application
from pytest_mock import MockerFixture
from unittest.mock import call as MockCall
from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import InitializeError
from ahriman.core.spawn import Spawn
from ahriman.core.status.watcher import Watcher
from ahriman.web.keys import ConfigurationKey
from ahriman.web.web import _create_socket, _create_watcher, _on_shutdown, _on_startup, run_server, setup_server
async def test_create_socket(application: Application, mocker: MockerFixture) -> None:
"""
must create socket
"""
path = "/run/ahriman.sock"
application[ConfigurationKey].set_option("web", "unix_socket", str(path))
current_on_shutdown = len(application.on_shutdown)
bind_mock = mocker.patch("socket.socket.bind")
chmod_mock = mocker.patch("pathlib.Path.chmod")
unlink_mock = mocker.patch("pathlib.Path.unlink")
sock = _create_socket(application[ConfigurationKey], application)
assert sock.family == socket.AF_UNIX
assert sock.type == socket.SOCK_STREAM
bind_mock.assert_called_once_with(str(path))
chmod_mock.assert_called_once_with(0o666)
assert len(application.on_shutdown) == current_on_shutdown + 1
# provoke socket removal
await application.on_shutdown[-1](application)
unlink_mock.assert_has_calls([MockCall(missing_ok=True), MockCall(missing_ok=True)])
def test_create_socket_empty(application: Application) -> None:
"""
must skip socket creation if not set by configuration
"""
assert _create_socket(application[ConfigurationKey], application) is None
def test_create_socket_safe(application: Application, mocker: MockerFixture) -> None:
"""
must create socket with default permission set
"""
path = "/run/ahriman.sock"
application[ConfigurationKey].set_option("web", "unix_socket", str(path))
application[ConfigurationKey].set_option("web", "unix_socket_unsafe", "no")
mocker.patch("socket.socket.bind")
mocker.patch("pathlib.Path.unlink")
chmod_mock = mocker.patch("pathlib.Path.chmod")
sock = _create_socket(application[ConfigurationKey], application)
assert sock is not None
chmod_mock.assert_not_called()
def test_create_watcher(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must create watcher for repository
"""
mocker.patch("ahriman.core.configuration.Configuration.from_path", return_value=configuration)
database_mock = mocker.patch("ahriman.core.database.SQLite.load")
client_mock = mocker.patch("ahriman.core.status.Client.load")
configuration_path, repository_id = configuration.check_loaded()
result = _create_watcher(configuration_path, repository_id)
assert isinstance(result, Watcher)
database_mock.assert_called_once()
client_mock.assert_called_once()
async def test_on_shutdown(application: Application, mocker: MockerFixture) -> None:
"""
must write information to log
"""
logging_mock = mocker.patch("logging.Logger.warning")
await _on_shutdown(application)
logging_mock.assert_called_once_with(pytest.helpers.anyvar(str, True))
async def test_on_startup(application: Application, watcher: Watcher, mocker: MockerFixture) -> None:
"""
must call load method
"""
mocker.patch("aiohttp.web.Application.__getitem__", return_value={"": watcher})
load_mock = mocker.patch("ahriman.core.status.watcher.Watcher.load")
await _on_startup(application)
load_mock.assert_called_once_with()
async def test_on_startup_exception(application: Application, watcher: Watcher, mocker: MockerFixture) -> None:
"""
must throw exception on load error
"""
mocker.patch("aiohttp.web.Application.__getitem__", return_value={"": watcher})
mocker.patch("ahriman.core.status.watcher.Watcher.load", side_effect=Exception)
with pytest.raises(InitializeError):
await _on_startup(application)
def test_run(application: Application, mocker: MockerFixture) -> None:
"""
must run application
"""
port = 8080
application[ConfigurationKey].set_option("web", "port", str(port))
run_application_mock = mocker.patch("ahriman.web.web.run_app")
run_server(application)
run_application_mock.assert_called_once_with(
application, host="127.0.0.1", port=port, sock=None, handle_signals=True,
access_log=pytest.helpers.anyvar(int),
)
def test_run_with_auth(application_with_auth: Application, mocker: MockerFixture) -> None:
"""
must run application with enabled authorization
"""
port = 8080
application_with_auth[ConfigurationKey].set_option("web", "port", str(port))
run_application_mock = mocker.patch("ahriman.web.web.run_app")
run_server(application_with_auth)
run_application_mock.assert_called_once_with(
application_with_auth, host="127.0.0.1", port=port, sock=None, handle_signals=True,
access_log=pytest.helpers.anyvar(int),
)
def test_run_with_socket(application: Application, mocker: MockerFixture) -> None:
"""
must run application with socket
"""
port = 8080
application[ConfigurationKey].set_option("web", "port", str(port))
socket_mock = mocker.patch("ahriman.web.web._create_socket", return_value=42)
run_application_mock = mocker.patch("ahriman.web.web.run_app")
run_server(application)
socket_mock.assert_called_once_with(application[ConfigurationKey], application)
run_application_mock.assert_called_once_with(
application, host="127.0.0.1", port=port, sock=42, handle_signals=True,
access_log=pytest.helpers.anyvar(int),
)
def test_setup_no_repositories(configuration: Configuration, spawner: Spawn) -> None:
"""
must raise InitializeError if no repositories set
"""
with pytest.raises(InitializeError):
setup_server(configuration, spawner, [])
@@ -0,0 +1,48 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.core.configuration import Configuration
from ahriman.models.user_access import UserAccess
from ahriman.web.views.api.docs import DocsView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await DocsView.get_permission(request) == UserAccess.Unauthorized
def test_routes() -> None:
"""
must return correct routes
"""
assert DocsView.ROUTES == ["/api-docs"]
def test_routes_dynamic(configuration: Configuration) -> None:
"""
must correctly return docs route
"""
assert DocsView.ROUTES == DocsView.routes(configuration)
def test_routes_dynamic_not_found(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must disable docs route if no apispec package found
"""
mocker.patch("ahriman.web.views.api.docs.aiohttp_apispec", None)
assert DocsView.routes(configuration) == []
async def test_get(client: TestClient) -> None:
"""
must generate api-docs correctly
"""
response = await client.get("/api-docs")
assert response.ok
assert await response.text()
@@ -0,0 +1,125 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from typing import Any
from ahriman.core.configuration import Configuration
from ahriman.models.user_access import UserAccess
from ahriman.web.views.api.swagger import SwaggerView
def _client(client: TestClient, mocker: MockerFixture) -> TestClient:
"""
generate test client with docs. Thanks to deprecation, we can't change application state since it was run
Args:
client(TestClient): test client fixture
mocker(MockerFixture): mocker object
Returns:
TestClient: test client fixture with additional properties
"""
swagger_dict = {
"paths": {
"/api/v1/logout": {
"get": {
"parameters": [
{
"in": "cookie",
"name": "AHRIMAN",
"schema": {
"type": "string",
},
},
],
},
"head": {},
"post": {
"parameters": [
{
"in": "cookie",
"name": "AHRIMAN",
"schema": {
"type": "string",
},
},
{
"in": "body",
"name": "schema",
"schema": {
"type": "string",
},
},
],
},
},
},
"components": {},
"security": [
{
"token": {
"type": "apiKey",
"name": "AHRIMAN",
"in": "cookie",
},
},
],
}
source = client.app.__getitem__
def getitem(name: str) -> Any:
if name == "swagger_dict":
return swagger_dict
return source(name)
mocker.patch("aiohttp.web.Application.__getitem__", side_effect=getitem)
return client
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await SwaggerView.get_permission(request) == UserAccess.Unauthorized
def test_routes() -> None:
"""
must return correct routes
"""
assert SwaggerView.ROUTES == ["/api-docs/swagger.json"]
def test_routes_dynamic(configuration: Configuration) -> None:
"""
must correctly return openapi url
"""
assert SwaggerView.ROUTES == SwaggerView.routes(configuration)
def test_routes_dynamic_not_found(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must disable openapi route if no apispec package found
"""
mocker.patch("ahriman.web.views.api.swagger.aiohttp_apispec", None)
assert SwaggerView.routes(configuration) == []
async def test_get(client: TestClient, mocker: MockerFixture) -> None:
"""
must generate api-docs correctly
"""
client = _client(client, mocker)
response = await client.get("/api-docs/swagger.json")
assert response.ok
json = await response.json()
assert "securitySchemes" in json["components"]
assert not any(parameter["in"] == "body" for parameter in json["paths"]["/api/v1/logout"]["post"]["parameters"])
assert "requestBody" in json["paths"]["/api/v1/logout"]["post"]
assert "requestBody" not in json["paths"]["/api/v1/logout"]["get"]
assert "requestBody" not in json["paths"]["/api/v1/logout"]["head"]
@@ -0,0 +1,19 @@
import pytest
from aiohttp.web import Application
from ahriman.web.views.base import BaseView
@pytest.fixture
def base(application: Application) -> BaseView:
"""
base view fixture
Args:
application(Application): application fixture
Returns:
BaseView: generated base view fixture
"""
return BaseView(pytest.helpers.request(application, "", ""))
@@ -0,0 +1,19 @@
from ahriman.core.configuration import Configuration
from ahriman.web.views.status_view_guard import StatusViewGuard
def test_routes(configuration: Configuration) -> None:
"""
must correctly return routes list
"""
StatusViewGuard.ROUTES = routes = ["route1", "route2"]
assert StatusViewGuard.routes(configuration) == routes
def test_routes_empty(configuration: Configuration) -> None:
"""
must return empty routes list if option is set
"""
StatusViewGuard.ROUTES = ["route1", "route2"]
configuration.set_option("web", "service_only", "yes")
assert StatusViewGuard.routes(configuration) == []
@@ -0,0 +1,257 @@
import pytest
from aiohttp.test_utils import TestClient
from aiohttp.web import HTTPBadRequest, HTTPNotFound
from multidict import MultiDict
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.core.configuration import Configuration
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.keys import WatcherKey
from ahriman.web.views.base import BaseView
def test_routes() -> None:
"""
must return correct routes
"""
assert BaseView.ROUTES == []
def test_configuration(base: BaseView) -> None:
"""
must return configuration
"""
assert base.configuration
def test_services(base: BaseView) -> None:
"""
must return services
"""
assert base.services
def test_sign(base: BaseView) -> None:
"""
must return GPP wrapper instance
"""
assert base.sign
def test_spawn(base: BaseView) -> None:
"""
must return spawn thread
"""
assert base.spawner
def test_validator(base: BaseView) -> None:
"""
must return validator service
"""
assert base.validator
def test_workers(base: BaseView) -> None:
"""
must return worker service
"""
assert base.workers
async def test_get_permission(base: BaseView) -> None:
"""
must search for permission attribute in class
"""
for method in ("DELETE", "GET", "POST"):
setattr(BaseView, f"{method.upper()}_PERMISSION", "permission")
for method in ("DELETE", "GET", "HEAD", "POST"):
request = pytest.helpers.request(base.request.app, "", method)
assert await base.get_permission(request) == "permission"
for method in ("OPTIONS",):
request = pytest.helpers.request(base.request.app, "", method)
assert await base.get_permission(request) == UserAccess.Unauthorized
def test_get_routes(configuration: Configuration, mocker: MockerFixture) -> None:
"""
must return list of available routes
"""
routes = ["route1", "route2"]
mocker.patch.object(BaseView, "ROUTES", routes)
assert BaseView.routes(configuration) == routes
def test_get_non_empty() -> None:
"""
must correctly extract non-empty values
"""
assert BaseView.get_non_empty(lambda k: k, "key")
with pytest.raises(KeyError):
BaseView.get_non_empty(lambda k: None, "key")
with pytest.raises(KeyError):
BaseView.get_non_empty(lambda k: "", "key")
assert BaseView.get_non_empty(lambda k: [k], "key")
with pytest.raises(KeyError):
BaseView.get_non_empty(lambda k: [], "key")
async def test_head(client: TestClient) -> None:
"""
must implement head as get method
"""
response = await client.head("/")
assert response.ok
assert await response.text() == ""
async def test_head_not_allowed(client: TestClient) -> None:
"""
must raise MethodNotAllowed in case if no get method was implemented
"""
response = await client.head("/api/v1/service/add")
assert response.status == 405
def test_page(base: BaseView) -> None:
"""
must extract page from query parameters
"""
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(limit=2, offset=3))
assert base.page() == (2, 3)
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(offset=3))
assert base.page() == (-1, 3)
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(limit=2))
assert base.page() == (2, 0)
def test_page_bad_request(base: BaseView) -> None:
"""
must raise HTTPBadRequest in case if parameters are invalid
"""
with pytest.raises(HTTPBadRequest):
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(limit="string"))
base.page()
with pytest.raises(HTTPBadRequest):
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(offset="string"))
base.page()
with pytest.raises(HTTPBadRequest):
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(limit=-2))
base.page()
with pytest.raises(HTTPBadRequest):
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(offset=-1))
base.page()
def test_repository_id(base: BaseView, repository_id: RepositoryId) -> None:
"""
must repository identifier from parameters
"""
base._request = pytest.helpers.request(base.request.app, "", "",
params=MultiDict(architecture="i686", repository="repo"))
assert base.repository_id() == RepositoryId("i686", "repo")
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(architecture="i686"))
assert base.repository_id() == repository_id
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict(repository="repo"))
assert base.repository_id() == repository_id
base._request = pytest.helpers.request(base.request.app, "", "", params=MultiDict())
assert base.repository_id() == repository_id
def test_service(base: BaseView) -> None:
"""
must return service for repository
"""
repository_id = RepositoryId("i686", "repo")
base.request.app[WatcherKey] = {
repository_id: watcher
for watcher in base.request.app[WatcherKey].values()
}
assert base.service(repository_id) == base.services[repository_id]
def test_service_auto(base: BaseView, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must return service for repository if no parameters set
"""
mocker.patch("ahriman.web.views.base.BaseView.repository_id", return_value=repository_id)
assert base.service() == base.services[repository_id]
def test_service_not_found(base: BaseView) -> None:
"""
must raise HTTPNotFound if no repository found
"""
with pytest.raises(HTTPNotFound):
base.service(RepositoryId("repo", "arch"))
def test_service_package(base: BaseView, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must validate that package exists
"""
mocker.patch("ahriman.web.views.base.BaseView.repository_id", return_value=repository_id)
with pytest.raises(HTTPNotFound):
base.service(package_base="base")
async def test_username(base: BaseView, mocker: MockerFixture) -> None:
"""
must return identity of logged-in user
"""
policy = AsyncMock()
policy.identify.return_value = "identity"
mocker.patch("aiohttp.web.Application.get", return_value=policy)
json = AsyncMock()
json.return_value = {}
base._request = pytest.helpers.request(base.request.app, "", "", json=json)
assert await base.username() == "identity"
policy.identify.assert_called_once_with(base.request)
async def test_username_no_auth(base: BaseView) -> None:
"""
must return None in case if auth is disabled
"""
json = AsyncMock()
json.return_value = {}
base._request = pytest.helpers.request(base.request.app, "", "", json=json)
assert await base.username() is None
async def test_username_request(base: BaseView) -> None:
"""
must read packager from request
"""
json = AsyncMock()
json.return_value = {"packager": "identity"}
base._request = pytest.helpers.request(base.request.app, "", "", json=json)
assert await base.username() == "identity"
async def test_username_request_exception(base: BaseView) -> None:
"""
must not fail in case if it cannot read request
"""
assert await base.username() is None
@@ -0,0 +1,65 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.user_access import UserAccess
from ahriman.web.views.index import IndexView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await IndexView.get_permission(request) == UserAccess.Unauthorized
def test_routes() -> None:
"""
must return correct routes
"""
assert IndexView.ROUTES == ["/", "/index.html"]
async def test_get(client_with_auth: TestClient) -> None:
"""
must generate status page correctly (/)
"""
response = await client_with_auth.get("/")
assert response.ok
assert await response.text()
async def test_get_index(client_with_auth: TestClient) -> None:
"""
must generate status page correctly (/index.html)
"""
response = await client_with_auth.get("/index.html")
assert response.ok
assert await response.text()
async def test_get_without_auth(client: TestClient) -> None:
"""
must use dummy authorized_userid function in case if no security library installed
"""
response = await client.get("/")
assert response.ok
assert await response.text()
async def test_get_static(client: TestClient) -> None:
"""
must return static files
"""
response = await client.get("/static/favicon.ico")
assert response.ok
async def test_get_static_with_auth(client_with_auth: TestClient) -> None:
"""
must return static files with authorization enabled
"""
response = await client_with_auth.get("/static/favicon.ico")
assert response.ok
@@ -0,0 +1,42 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.user_access import UserAccess
from ahriman.web.views.static import StaticView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await StaticView.get_permission(request) == UserAccess.Unauthorized
def test_routes() -> None:
"""
must return correct routes
"""
assert StaticView.ROUTES == ["/favicon.ico"]
async def test_get(client_with_auth: TestClient) -> None:
"""
must redirect favicon to static files
"""
response = await client_with_auth.get("/favicon.ico", allow_redirects=False)
assert response.status == 302
assert response.headers["Location"] == "/static/favicon.ico"
async def test_get_not_found(client_with_auth: TestClient, mocker: MockerFixture) -> None:
"""
must raise not found if path is invalid
"""
static_route = next(route for route in client_with_auth.app.router.routes() if route.handler == StaticView)
mocker.patch.object(static_route.handler, "ROUTES", [])
response = await client_with_auth.get("/favicon.ico", allow_redirects=False)
assert response.status == 404
@@ -0,0 +1,254 @@
import asyncio
import pytest
from aiohttp.test_utils import TestClient
from aiohttp.web import HTTPBadRequest
from asyncio import Queue
from multidict import MultiDict
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.core.status.watcher import Watcher
from ahriman.models.event import EventType
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.keys import WatcherKey
from ahriman.web.views.base import BaseView
from ahriman.web.views.v1.auditlog.event_bus import EventBusView
async def _producer(watcher: Watcher, package_ahriman: Package) -> None:
"""
create producer
Args:
watcher(Watcher): watcher test instance
package_ahriman(Package): package test instance
"""
await asyncio.sleep(0.1)
await watcher.event_bus.broadcast(EventType.PackageRemoved, package_ahriman.base)
await watcher.event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base, status="success")
await asyncio.sleep(0.1)
await watcher.event_bus.shutdown()
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await EventBusView.get_permission(request) == UserAccess.Full
async def test_get_permission_build_log() -> None:
"""
must return full permission for build log stream
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict(event=EventType.BuildLog))
assert await EventBusView.get_permission(request) == UserAccess.Full
async def test_get_permission_build_log_with_read_events() -> None:
"""
must return full permission for mixed build log and read event stream
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict([
("event", EventType.BuildLog),
("event", EventType.PackageUpdated),
]))
assert await EventBusView.get_permission(request) == UserAccess.Full
async def test_get_permission_invalid_event() -> None:
"""
must return full permission for invalid event type
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict(event="invalid"))
assert await EventBusView.get_permission(request) == UserAccess.Full
async def test_get_permission_post() -> None:
"""
must use default permission for non-get requests
"""
request = pytest.helpers.request("", "", "POST", params=MultiDict(event=EventType.PackageUpdated))
assert await EventBusView.get_permission(request) == await BaseView.get_permission(request)
async def test_get_permission_read_events() -> None:
"""
must return read permission for package and status streams
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict(
("event", event_type) for event_type in EventBusView.READ_EVENTS
))
assert await EventBusView.get_permission(request) == UserAccess.Read
def test_routes() -> None:
"""
must return correct routes
"""
assert EventBusView.ROUTES == ["/api/v1/events/stream"]
async def test_run_timeout() -> None:
"""
must handle timeout and continue loop
"""
queue = Queue()
async def _shutdown() -> None:
await asyncio.sleep(0.05)
queue.shutdown()
response = AsyncMock()
response.is_connected = lambda: True
response.ping_interval = 0.01
asyncio.create_task(_shutdown())
await EventBusView._run(response, queue)
def test_topics() -> None:
"""
must parse event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict([
("event", EventType.PackageUpdated),
("event", EventType.PackageRemoved),
]))
assert EventBusView(request)._topics() == [EventType.PackageUpdated, EventType.PackageRemoved]
def test_topics_empty() -> None:
"""
must return None for missing event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict())
assert EventBusView(request)._topics() is None
def test_topics_invalid() -> None:
"""
must raise bad request for invalid event filters
"""
request = pytest.helpers.request("", "", "GET", params=MultiDict(event="invalid"))
with pytest.raises(HTTPBadRequest):
EventBusView(request)._topics()
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must stream events via SSE
"""
watcher = next(iter(client.app[WatcherKey].values()))
asyncio.create_task(_producer(watcher, package_ahriman))
request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
# no content validation here because it is a streaming response
assert not request_schema.validate({})
response = await client.get("/api/v1/events/stream")
assert response.status == 200
body = await response.text()
assert EventType.PackageUpdated in body
assert "ahriman" in body
async def test_get_with_topic_filter(client: TestClient, package_ahriman: Package) -> None:
"""
must filter events by topic
"""
watcher = next(iter(client.app[WatcherKey].values()))
asyncio.create_task(_producer(watcher, package_ahriman))
request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
payload = {"event": [EventType.PackageUpdated]}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/events/stream", params=payload)
assert response.status == 200
body = await response.text()
assert EventType.PackageUpdated in body
assert EventType.PackageRemoved not in body
async def test_get_with_object_id_filter(client: TestClient, package_ahriman: Package) -> None:
"""
must filter events by object_id
"""
watcher = next(iter(client.app[WatcherKey].values()))
asyncio.create_task(_producer(watcher, package_ahriman))
request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
payload = {"object_id": "non-existent-package"}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/events/stream", params=payload)
assert response.status == 200
body = await response.text()
assert "ahriman" not in body
async def test_get_bad_request(client: TestClient) -> None:
"""
must return bad request for invalid event type
"""
response_schema = pytest.helpers.schema_response(EventBusView.get, code=400)
response = await client.get("/api/v1/events/stream", params={"event": "invalid"})
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_get_not_found(client: TestClient) -> None:
"""
must return not found for unknown repository
"""
response_schema = pytest.helpers.schema_response(EventBusView.get, code=404)
response = await client.get("/api/v1/events/stream", params={"architecture": "unknown", "repository": "unknown"})
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_get_connection_reset(client: TestClient, mocker: MockerFixture) -> None:
"""
must handle connection reset
"""
mocker.patch.object(EventBusView, "_run", side_effect=ConnectionResetError)
response = await client.get("/api/v1/events/stream")
assert response.status == 200
async def test_head(client: TestClient) -> None:
"""
must check stream availability without opening SSE stream
"""
response = await client.head("/api/v1/events/stream", params={"event": EventType.PackageUpdated})
assert response.status == 200
assert response.headers["Content-Type"] == "text/event-stream"
assert not await response.text()
async def test_head_bad_request(client: TestClient) -> None:
"""
must return bad request for invalid event type
"""
response = await client.head("/api/v1/events/stream", params={"event": "invalid"})
assert response.status == 400
assert not await response.text()
async def test_head_not_found(client: TestClient) -> None:
"""
must return not found for unknown repository
"""
response = await client.head("/api/v1/events/stream", params={"architecture": "unknown", "repository": "unknown"})
assert response.status == 404
assert not await response.text()
@@ -0,0 +1,128 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.event import Event
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.auditlog.events import EventsView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET", "POST"):
request = pytest.helpers.request("", "", method)
assert await EventsView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert EventsView.ROUTES == ["/api/v1/events"]
async def test_get(client: TestClient) -> None:
"""
must return all events
"""
event1 = Event("event1", "object1", "message", key="value")
event2 = Event("event2", "object2")
await client.post("/api/v1/events", json=event1.view())
await client.post("/api/v1/events", json=event2.view())
response_schema = pytest.helpers.schema_response(EventsView.get)
response = await client.get("/api/v1/events")
assert response.ok
json = await response.json()
assert not response_schema.validate(json, many=True)
events = [Event.from_json(event) for event in json]
assert events == [event2, event1]
async def test_get_with_pagination(client: TestClient) -> None:
"""
must get events with pagination
"""
event1 = Event("event1", "object1", "message", key="value")
event2 = Event("event2", "object2")
await client.post("/api/v1/events", json=event1.view())
await client.post("/api/v1/events", json=event2.view())
request_schema = pytest.helpers.schema_request(EventsView.get, location="querystring")
payload = {"limit": 1, "offset": 1}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/events", params=payload)
assert response.status == 200
json = await response.json()
assert [Event.from_json(event) for event in json] == [event1]
async def test_get_with_filter(client: TestClient) -> None:
"""
must get events with filter by creation date
"""
event1 = Event("event1", "object1", "message", key="value", created=1)
event2 = Event("event2", "object2", created=2)
await client.post("/api/v1/events", json=event1.view())
await client.post("/api/v1/events", json=event2.view())
request_schema = pytest.helpers.schema_request(EventsView.get, location="querystring")
payload = {"from_date": 1, "to_date": 2}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/events", params=payload)
assert response.status == 200
json = await response.json()
assert [Event.from_json(event) for event in json] == [event1]
async def test_get_bad_request(client: TestClient) -> None:
"""
must return bad request for invalid query parameters
"""
response_schema = pytest.helpers.schema_response(EventsView.get, code=400)
response = await client.get("/api/v1/events", params={"limit": "limit"})
assert response.status == 400
assert not response_schema.validate(await response.json())
response = await client.get("/api/v1/events", params={"offset": "offset"})
assert response.status == 400
assert not response_schema.validate(await response.json())
response = await client.get("/api/v1/events", params={"from_date": "from_date"})
assert response.status == 400
assert not response_schema.validate(await response.json())
response = await client.get("/api/v1/events", params={"to_date": "to_date"})
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_post(client: TestClient) -> None:
"""
must create event
"""
event = Event("event1", "object1", "message", key="value")
request_schema = pytest.helpers.schema_request(EventsView.post)
payload = event.view()
assert not request_schema.validate(payload)
response = await client.post("/api/v1/events", json=payload)
assert response.status == 204
async def test_post_exception(client: TestClient) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(EventsView.post, code=400)
response = await client.post("/api/v1/events", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,80 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.user_access import UserAccess
from ahriman.models.worker import Worker
from ahriman.web.views.v1.distributed.workers import WorkersView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("DELETE", "GET", "POST"):
request = pytest.helpers.request("", "", method)
assert await WorkersView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert WorkersView.ROUTES == ["/api/v1/distributed"]
async def test_delete(client: TestClient) -> None:
"""
must delete all workers
"""
await client.post("/api/v1/distributed", json={"address": "address1", "identifier": "1"})
await client.post("/api/v1/distributed", json={"address": "address2", "identifier": "2"})
response = await client.delete("/api/v1/distributed")
assert response.status == 204
response = await client.get("/api/v1/distributed")
json = await response.json()
assert not json
async def test_get(client: TestClient) -> None:
"""
must return all workers
"""
await client.post("/api/v1/distributed", json={"address": "address1", "identifier": "1"})
await client.post("/api/v1/distributed", json={"address": "address2", "identifier": "2"})
response_schema = pytest.helpers.schema_response(WorkersView.get)
response = await client.get("/api/v1/distributed")
assert response.ok
json = await response.json()
assert not response_schema.validate(json, many=True)
workers = [Worker(item["address"], identifier=item["identifier"]) for item in json]
assert workers == [Worker("address1", identifier="1"), Worker("address2", identifier="2")]
async def test_post(client: TestClient) -> None:
"""
must update worker
"""
worker = Worker("address1", identifier="1")
request_schema = pytest.helpers.schema_request(WorkersView.post)
payload = worker.view()
assert not request_schema.validate(payload)
response = await client.post("/api/v1/distributed", json=payload)
assert response.status == 204
async def test_post_exception(client: TestClient) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(WorkersView.post, code=400)
response = await client.post("/api/v1/distributed", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,52 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.archives import Archives
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await Archives.get_permission(request) == UserAccess.Reporter
def test_routes() -> None:
"""
must return correct routes
"""
assert Archives.ROUTES == ["/api/v1/packages/{package}/archives"]
async def test_get(client: TestClient, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must get archives for package
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
mocker.patch("ahriman.core.status.watcher.Watcher.package_archives", return_value=[package_ahriman])
response_schema = pytest.helpers.schema_response(Archives.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/archives")
assert response.status == 200
archives = await response.json()
assert not response_schema.validate(archives)
async def test_get_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return not found for missing package
"""
response_schema = pytest.helpers.schema_response(Archives.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/archives")
assert response.status == 404
assert not response_schema.validate(await response.json())
@@ -0,0 +1,84 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.changes import Changes
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.changes import ChangesView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await ChangesView.get_permission(request) == UserAccess.Reporter
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await ChangesView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert ChangesView.ROUTES == ["/api/v1/packages/{package}/changes"]
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must get changes for package
"""
changes = Changes("sha", "change")
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/changes", json=changes.view())
response_schema = pytest.helpers.schema_response(ChangesView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/changes")
assert response.status == 200
assert await response.json() == changes.view()
assert not response_schema.validate(changes.view())
async def test_get_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return not found for missing package
"""
response_schema = pytest.helpers.schema_response(ChangesView.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/changes")
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post(client: TestClient, package_ahriman: Package) -> None:
"""
must update package changes
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(ChangesView.post)
changes = Changes("sha", "change")
assert not request_schema.validate(changes.view())
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/changes", json=changes.view())
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/changes")
assert await response.json() == changes.view()
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(ChangesView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/changes", json=[])
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,97 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.dependencies import Dependencies
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.dependencies import DependenciesView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await DependenciesView.get_permission(request) == UserAccess.Reporter
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await DependenciesView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert DependenciesView.ROUTES == ["/api/v1/packages/{package}/dependencies"]
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must get dependencies for package
"""
dependency = Dependencies({"path": ["package"]})
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/dependencies", json=dependency.view())
response_schema = pytest.helpers.schema_response(DependenciesView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/dependencies")
assert response.status == 200
dependencies = await response.json()
assert not response_schema.validate(dependencies)
assert dependencies == dependency.view()
async def test_get_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return not found for missing package
"""
response_schema = pytest.helpers.schema_response(DependenciesView.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/dependencies")
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post(client: TestClient, package_ahriman: Package) -> None:
"""
must create dependencies
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(DependenciesView.post)
payload = {"paths": {"path": ["package"]}}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/dependencies", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/dependencies")
dependencies = await response.json()
assert dependencies == payload
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(DependenciesView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/dependencies", json=[])
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_post_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on unknown package
"""
response_schema = pytest.helpers.schema_response(DependenciesView.post, code=404)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/dependencies", json={})
assert response.status == 404
assert not response_schema.validate(await response.json())
@@ -0,0 +1,60 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.hold import HoldView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await HoldView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert HoldView.ROUTES == ["/api/v1/packages/{package}/hold"]
async def test_post(client: TestClient, package_ahriman: Package) -> None:
"""
must update package hold status
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(HoldView.post)
payload = {"is_held": True}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/hold", json=payload)
assert response.status == 204
async def test_post_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return Not Found for unknown package
"""
response_schema = pytest.helpers.schema_response(HoldView.post, code=404)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/hold", json={"is_held": False})
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(HoldView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/hold", json=[])
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,123 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.logs import LogsView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await LogsView.get_permission(request) == UserAccess.Reporter
for method in ("DELETE", "POST"):
request = pytest.helpers.request("", "", method)
assert await LogsView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert LogsView.ROUTES == ["/api/v1/packages/{package}/logs"]
async def test_delete(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must delete logs for package
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message 1", "version": "42"})
await client.post(f"/api/v1/packages/{package_python_schedule.base}/logs",
json={"created": 42.0, "message": "message 2", "version": "42"})
request_schema = pytest.helpers.schema_request(LogsView.delete, location="querystring")
payload = {}
assert not request_schema.validate(payload)
payload = {"version": "42"}
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}/logs", params=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
logs = await response.json()
assert logs["logs"]
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}/logs")
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
logs = await response.json()
assert not logs["logs"]
response = await client.get(f"/api/v1/packages/{package_python_schedule.base}/logs")
logs = await response.json()
assert logs["logs"]
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must get logs for package
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message", "version": "42"})
response_schema = pytest.helpers.schema_response(LogsView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
assert response.status == 200
logs = await response.json()
assert not response_schema.validate(logs)
assert logs["logs"] == "[1970-01-01 00:00:42] message"
async def test_get_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return not found for missing package
"""
response_schema = pytest.helpers.schema_response(LogsView.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post(client: TestClient, package_ahriman: Package) -> None:
"""
must create logs record
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(LogsView.post)
payload = {"created": 42.0, "message": "message", "version": "42"}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/logs", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
logs = await response.json()
assert logs["logs"] == "[1970-01-01 00:00:42] message"
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(LogsView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/logs", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,174 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.package import PackageView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
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.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert PackageView.ROUTES == ["/api/v1/packages/{package}"]
async def test_delete(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must delete single base
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}")
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.status == 404
response = await client.get(f"/api/v1/packages/{package_python_schedule.base}")
assert response.ok
async def test_delete_unknown(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must suppress errors on unknown package deletion
"""
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}")
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.status == 404
response = await client.get(f"/api/v1/packages/{package_python_schedule.base}")
assert response.ok
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must return status for specific package
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
response_schema = pytest.helpers.schema_response(PackageView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.ok
json = await response.json()
assert not response_schema.validate(json, many=True)
packages = [Package.from_json(item["package"]) for item in json]
assert packages
assert {package.base for package in packages} == {package_ahriman.base}
async def test_get_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return Not Found for unknown package
"""
response_schema = pytest.helpers.schema_response(PackageView.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_post(client: TestClient, package_ahriman: Package, repository_id: RepositoryId) -> None:
"""
must update package status
"""
request_schema = pytest.helpers.schema_request(PackageView.post)
payload = {
"status": BuildStatusEnum.Success.value,
"package": package_ahriman.view(),
"repository": repository_id.view(),
}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.ok
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(PackageView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
async def test_post_light(client: TestClient, package_ahriman: Package, repository_id: RepositoryId) -> None:
"""
must update package status only
"""
request_schema = pytest.helpers.schema_request(PackageView.post)
payload = {
"status": BuildStatusEnum.Success.value,
"package": package_ahriman.view(),
"repository": repository_id.view(),
}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json=payload)
assert response.status == 204
payload = {
"status": BuildStatusEnum.Success.value,
"repository": repository_id.view(),
}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
assert response.ok
statuses = {
Package.from_json(item["package"]).base: BuildStatus.from_json(item["status"])
for item in await response.json()
}
assert statuses[package_ahriman.base].status == BuildStatusEnum.Success
async def test_post_not_found(client: TestClient, package_ahriman: Package, repository_id: RepositoryId) -> None:
"""
must raise exception on status update for unknown package
"""
request_schema = pytest.helpers.schema_request(PackageView.post)
response_schema = pytest.helpers.schema_response(PackageView.post, code=400)
payload = {
"status": BuildStatusEnum.Success.value,
"repository": repository_id.view(),
}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json=payload)
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,83 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.packages import (PackagesView)
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await PackagesView.get_permission(request) == UserAccess.Read
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await PackagesView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert PackagesView.ROUTES == ["/api/v1/packages"]
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must return status for all packages
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
response_schema = pytest.helpers.schema_response(PackagesView.get)
response = await client.get("/api/v1/packages")
assert response.status == 200
json = await response.json()
assert not response_schema.validate(json, many=True)
packages = [Package.from_json(item["package"]) for item in json]
assert packages
assert {package.base for package in packages} == {package_ahriman.base, package_python_schedule.base}
async def test_get_with_pagination(client: TestClient, package_ahriman: Package,
package_python_schedule: Package) -> None:
"""
must return paginated status for packages
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
request_schema = pytest.helpers.schema_request(PackagesView.get, location="querystring")
response_schema = pytest.helpers.schema_response(PackagesView.get)
payload = {"limit": 1, "offset": 1}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/packages", params=payload)
assert response.status == 200
json = await response.json()
assert not response_schema.validate(json, many=True)
packages = [Package.from_json(item["package"]) for item in json]
assert packages
assert {package.base for package in packages} == {package_python_schedule.base}
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must be able to reload packages
"""
load_mock = mocker.patch("ahriman.core.status.watcher.Watcher.load")
response = await client.post("/api/v1/packages")
assert response.status == 204
load_mock.assert_called_once_with()
@@ -0,0 +1,83 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.pkgbuild_patch import PkgbuildPatch
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.patch import PatchView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await PatchView.get_permission(request) == UserAccess.Reporter
for method in ("DELETE",):
request = pytest.helpers.request("", "", method)
assert await PatchView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert PatchView.ROUTES == ["/api/v1/packages/{package}/patches/{patch}"]
async def test_delete(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
"""
must delete patch for package
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
patch_key = "k"
await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json={"key": patch_key, "value": "v"})
await client.post(f"/api/v1/packages/{package_python_schedule.base}/patches",
json={"key": patch_key, "value": "v2"})
response = await client.delete(f"/api/v1/packages/{package_ahriman.base}/patches/{patch_key}")
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches")
assert not await response.json()
response = await client.get(f"/api/v1/packages/{package_python_schedule.base}/patches")
assert await response.json()
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must get patch for package
"""
patch = PkgbuildPatch("k", "v")
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json=patch.view())
response_schema = pytest.helpers.schema_response(PatchView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches/{patch.key}")
assert response.status == 200
patches = await response.json()
assert not response_schema.validate(patches)
assert patches == patch.view()
async def test_get_patch_not_found(client: TestClient, package_ahriman: Package) -> None:
"""
must return not found for missing patch
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
response_schema = pytest.helpers.schema_response(PatchView.get, code=404)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches/random")
assert response.status == 404
assert not response_schema.validate(await response.json())
@@ -0,0 +1,112 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.pkgbuild_patch import PkgbuildPatch
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.packages.patches import PatchesView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await PatchesView.get_permission(request) == UserAccess.Reporter
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await PatchesView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert PatchesView.ROUTES == ["/api/v1/packages/{package}/patches"]
async def test_get(client: TestClient, package_ahriman: Package) -> None:
"""
must get patch for package
"""
patch = PkgbuildPatch("k", "v")
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json=patch.view())
response_schema = pytest.helpers.schema_response(PatchesView.get)
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches")
assert response.status == 200
patches = await response.json()
assert not response_schema.validate(patches)
assert patches == [patch.view()]
async def test_post(client: TestClient, package_ahriman: Package) -> None:
"""
must create patch
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(PatchesView.post)
payload = {"key": "k", "value": "v"}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches")
patches = await response.json()
assert patches == [payload]
async def test_post_full_diff(client: TestClient, package_ahriman: Package) -> None:
"""
must create patch from full diff
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(PatchesView.post)
payload = {"value": "v"}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches")
patches = await response.json()
assert patches == [payload]
async def test_post_array(client: TestClient, package_ahriman: Package) -> None:
"""
must create patch from list variable
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
request_schema = pytest.helpers.schema_request(PatchesView.post)
payload = {"key": "k", "value": "(array value)"}
assert not request_schema.validate(payload)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json=payload)
assert response.status == 204
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/patches")
patches = await response.json()
parsed = [PkgbuildPatch(patch["key"], patch["value"]) for patch in patches]
assert parsed == [PkgbuildPatch("k", ["array", "value"])]
async def test_post_exception(client: TestClient, package_ahriman: Package) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(PatchesView.post, code=400)
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/patches", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,102 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.models.pkgbuild_patch import PkgbuildPatch
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.add import AddView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await AddView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert AddView.ROUTES == ["/api/v1/service/add"]
async def test_post(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(AddView.post)
response_schema = pytest.helpers.schema_response(AddView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/add", json=payload)
assert response.ok
add_mock.assert_called_once_with(repository_id, ["ahriman"], "username",
patches=[], now=True, increment=True, refresh=False)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_patches(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request with patches correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(AddView.post)
payload = {
"packages": ["ahriman"],
"patches": [
{
"key": "k",
"value": "v",
},
{
"key": "k2",
},
]
}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/add", json=payload)
assert response.ok
add_mock.assert_called_once_with(repository_id, ["ahriman"], "username",
patches=[PkgbuildPatch("k", "v"), PkgbuildPatch("k2", "")],
now=True, increment=True, refresh=False)
async def test_post_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must call raise 400 on empty request
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
response_schema = pytest.helpers.schema_response(AddView.post, code=400)
response = await client.post("/api/v1/service/add", json={"packages": [""]})
assert response.status == 400
assert not response_schema.validate(await response.json())
add_mock.assert_not_called()
response = await client.post("/api/v1/service/add", json={"packages": []})
assert response.status == 400
assert not response_schema.validate(await response.json())
add_mock.assert_not_called()
response = await client.post("/api/v1/service/add", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
add_mock.assert_not_called()
@@ -0,0 +1,54 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.core.formatters.configuration_printer import ConfigurationPrinter
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.config import ConfigView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await ConfigView.get_permission(request) == UserAccess.Full
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await ConfigView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert ConfigView.ROUTES == ["/api/v1/service/config"]
async def test_get(client: TestClient) -> None:
"""
must get web configuration
"""
response_schema = pytest.helpers.schema_response(ConfigView.get)
response = await client.get("/api/v1/service/config")
assert response.status == 200
json = await response.json()
assert json # check that it is not empty
assert not response_schema.validate(json)
# check that there are no keys which have to be hidden
assert not any(value["key"] in ConfigurationPrinter.HIDE_KEYS for value in json)
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must update package changes
"""
reload_mock = mocker.patch("ahriman.core.configuration.Configuration.reload")
response = await client.post("/api/v1/service/config")
assert response.status == 204
reload_mock.assert_called_once_with()
@@ -0,0 +1,77 @@
import pytest
from aiohttp.test_utils import TestClient
from ahriman.models.build_status import BuildStatusEnum
from ahriman.models.package import Package
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.logs import LogsView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("DELETE",):
request = pytest.helpers.request("", "", method)
assert await LogsView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert LogsView.ROUTES == ["/api/v1/service/logs"]
async def test_delete(client: TestClient, package_ahriman: Package) -> None:
"""
must delete all logs
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message 1", "version": "42"})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 43.0, "message": "message 2", "version": "43"})
response = await client.delete("/api/v1/service/logs")
assert response.status == 204
response = await client.get(f"/api/v2/packages/{package_ahriman.base}/logs")
json = await response.json()
assert not json
async def test_delete_partially(client: TestClient, package_ahriman: Package) -> None:
"""
must delete logs based on count input
"""
await client.post(f"/api/v1/packages/{package_ahriman.base}",
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 42.0, "message": "message 1", "version": "42"})
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
json={"created": 43.0, "message": "message 2", "version": "43"})
request_schema = pytest.helpers.schema_request(LogsView.delete, location="querystring")
payload = {"keep_last_records": 1}
assert not request_schema.validate(payload)
response = await client.delete("/api/v1/service/logs", params=payload)
assert response.status == 204
response = await client.get(f"/api/v2/packages/{package_ahriman.base}/logs")
json = await response.json()
assert json
async def test_delete_exception(client: TestClient) -> None:
"""
must raise exception on invalid payload
"""
response_schema = pytest.helpers.schema_response(LogsView.delete, code=400)
response = await client.delete("/api/v1/service/logs", params={"keep_last_records": "string"})
assert response.status == 400
assert not response_schema.validate(await response.json())
@@ -0,0 +1,101 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.pgp import PGPView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await PGPView.get_permission(request) == UserAccess.Reporter
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await PGPView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert PGPView.ROUTES == ["/api/v1/service/pgp"]
async def test_get(client: TestClient, mocker: MockerFixture) -> None:
"""
must retrieve key from the keyserver
"""
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", return_value="imported")
request_schema = pytest.helpers.schema_request(PGPView.get, location="querystring")
response_schema = pytest.helpers.schema_response(PGPView.get)
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/service/pgp", params=payload)
assert response.ok
import_mock.assert_called_once_with("keyserver.ubuntu.com", "0xdeadbeaf")
assert not response_schema.validate(await response.json())
assert await response.json() == {"key": "imported"}
async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise 400 on missing parameters
"""
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download")
response_schema = pytest.helpers.schema_response(PGPView.get, code=400)
response = await client.get("/api/v1/service/pgp")
assert response.status == 400
assert not response_schema.validate(await response.json())
import_mock.assert_not_called()
async def test_get_process_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise 404 on invalid PGP server response
"""
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", side_effect=Exception)
response_schema = pytest.helpers.schema_response(PGPView.get, code=400)
response = await client.get("/api/v1/service/pgp", params={"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"})
assert response.status == 404
assert not response_schema.validate(await response.json())
import_mock.assert_called_once_with("keyserver.ubuntu.com", "0xdeadbeaf")
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import", return_value="abc")
request_schema = pytest.helpers.schema_request(PGPView.post)
response_schema = pytest.helpers.schema_response(PGPView.post)
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/pgp", json=payload)
assert response.ok
import_mock.assert_called_once_with("0xdeadbeaf", "keyserver.ubuntu.com")
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise exception on missing key payload
"""
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
response_schema = pytest.helpers.schema_response(PGPView.post, code=400)
response = await client.post("/api/v1/service/pgp")
assert response.status == 400
assert not response_schema.validate(await response.json())
import_mock.assert_not_called()
@@ -0,0 +1,53 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.process import ProcessView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await ProcessView.get_permission(request) == UserAccess.Reporter
def test_routes() -> None:
"""
must return correct routes
"""
assert ProcessView.ROUTES == ["/api/v1/service/process/{process_id}"]
async def test_get(client: TestClient, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
process = "abc"
process_mock = mocker.patch("ahriman.core.spawn.Spawn.has_process", return_value=True)
response_schema = pytest.helpers.schema_response(ProcessView.get)
response = await client.get(f"/api/v1/service/process/{process}")
assert response.ok
process_mock.assert_called_once_with(process)
json = await response.json()
assert json["is_alive"]
assert not response_schema.validate(json)
async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must call raise 404 on unknown process
"""
process = "abc"
mocker.patch("ahriman.core.spawn.Spawn.has_process", return_value=False)
response_schema = pytest.helpers.schema_response(ProcessView.get, code=404)
response = await client.get(f"/api/v1/service/process/{process}")
assert response.status == 404
assert not response_schema.validate(await response.json())
@@ -0,0 +1,60 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.rebuild import RebuildView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await RebuildView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert RebuildView.ROUTES == ["/api/v1/service/rebuild"]
async def test_post(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RebuildView.post)
response_schema = pytest.helpers.schema_response(RebuildView.post)
payload = {"packages": ["python", "ahriman"]}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/rebuild", json=payload)
assert response.ok
rebuild_mock.assert_called_once_with(repository_id, "python", "username", increment=True)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise exception on missing packages payload
"""
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
response_schema = pytest.helpers.schema_response(RebuildView.post, code=400)
response = await client.post("/api/v1/service/rebuild")
assert response.status == 400
assert not response_schema.validate(await response.json())
rebuild_mock.assert_not_called()
@@ -0,0 +1,56 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.remove import RemoveView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await RemoveView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert RemoveView.ROUTES == ["/api/v1/service/remove"]
async def test_post(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove", return_value="abc")
request_schema = pytest.helpers.schema_request(RemoveView.post)
response_schema = pytest.helpers.schema_response(RemoveView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/remove", json=payload)
assert response.ok
remove_mock.assert_called_once_with(repository_id, ["ahriman"])
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise exception on missing packages payload
"""
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
response_schema = pytest.helpers.schema_response(RemoveView.post, code=400)
response = await client.post("/api/v1/service/remove")
assert response.status == 400
assert not response_schema.validate(await response.json())
remove_mock.assert_not_called()
@@ -0,0 +1,92 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.models.pkgbuild_patch import PkgbuildPatch
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.request import RequestView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await RequestView.get_permission(request) == UserAccess.Reporter
def test_routes() -> None:
"""
must return correct routes
"""
assert RequestView.ROUTES == ["/api/v1/service/request"]
async def test_post(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RequestView.post)
response_schema = pytest.helpers.schema_response(RequestView.post)
payload = {"packages": ["ahriman"]}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/request", json=payload)
assert response.ok
add_mock.assert_called_once_with(repository_id, ["ahriman"], "username",
patches=[], now=False, increment=False, refresh=False)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_patches(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request with patches correctly
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RequestView.post)
payload = {
"packages": ["ahriman"],
"patches": [
{
"key": "k",
"value": "v",
},
{
"key": "k2",
},
]
}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/request", json=payload)
assert response.ok
add_mock.assert_called_once_with(repository_id, ["ahriman"], "username",
patches=[PkgbuildPatch("k", "v"), PkgbuildPatch("k2", "")],
now=False, increment=False, refresh=False)
async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise exception on missing packages payload
"""
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
response_schema = pytest.helpers.schema_response(RequestView.post, code=400)
response = await client.post("/api/v1/service/request")
assert response.status == 400
assert not response_schema.validate(await response.json())
add_mock.assert_not_called()
@@ -0,0 +1,70 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from unittest.mock import AsyncMock
from ahriman.models.repository_id import RepositoryId
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.rollback import RollbackView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("POST",):
request = pytest.helpers.request("", "", method)
assert await RollbackView.get_permission(request) == UserAccess.Full
def test_routes() -> None:
"""
must return correct routes
"""
assert RollbackView.ROUTES == ["/api/v1/service/rollback"]
async def test_post(client: TestClient, repository_id: RepositoryId, mocker: MockerFixture) -> None:
"""
must call post request correctly
"""
rollback_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rollback", return_value="abc")
user_mock = AsyncMock()
user_mock.return_value = "username"
mocker.patch("ahriman.web.views.base.BaseView.username", side_effect=user_mock)
request_schema = pytest.helpers.schema_request(RollbackView.post)
response_schema = pytest.helpers.schema_response(RollbackView.post)
payload = {"package": "ahriman", "version": "version"}
assert not request_schema.validate(payload)
response = await client.post("/api/v1/service/rollback", json=payload)
assert response.ok
rollback_mock.assert_called_once_with(repository_id, "ahriman", "version", "username", hold=True)
json = await response.json()
assert json["process_id"] == "abc"
assert not response_schema.validate(json)
async def test_post_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must call raise 400 on empty request
"""
rollback_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rollback")
response_schema = pytest.helpers.schema_response(RollbackView.post, code=400)
response = await client.post("/api/v1/service/rollback", json={"package": "", "version": "version"})
assert response.status == 400
assert not response_schema.validate(await response.json())
rollback_mock.assert_not_called()
response = await client.post("/api/v1/service/rollback", json={"package": "ahriman", "version": ""})
assert response.status == 400
assert not response_schema.validate(await response.json())
rollback_mock.assert_not_called()
response = await client.post("/api/v1/service/rollback", json={})
assert response.status == 400
assert not response_schema.validate(await response.json())
rollback_mock.assert_not_called()
@@ -0,0 +1,80 @@
import pytest
from aiohttp.test_utils import TestClient
from pytest_mock import MockerFixture
from ahriman.models.aur_package import AURPackage
from ahriman.models.user_access import UserAccess
from ahriman.web.views.v1.service.search import SearchView
async def test_get_permission() -> None:
"""
must return correct permission for the request
"""
for method in ("GET",):
request = pytest.helpers.request("", "", method)
assert await SearchView.get_permission(request) == UserAccess.Reporter
def test_routes() -> None:
"""
must return correct routes
"""
assert SearchView.ROUTES == ["/api/v1/service/search"]
async def test_get(client: TestClient, aur_package_ahriman: AURPackage, mocker: MockerFixture) -> None:
"""
must call get request correctly
"""
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[aur_package_ahriman])
request_schema = pytest.helpers.schema_request(SearchView.get, location="querystring")
response_schema = pytest.helpers.schema_response(SearchView.get)
payload = {"for": ["ahriman"]}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/service/search", params=payload)
assert response.ok
assert await response.json() == [{"package": aur_package_ahriman.package_base,
"description": aur_package_ahriman.description}]
assert not response_schema.validate(await response.json(), many=True)
async def test_get_exception(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise 400 on empty search string
"""
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
response_schema = pytest.helpers.schema_response(SearchView.get, code=400)
response = await client.get("/api/v1/service/search")
assert response.status == 400
assert not response_schema.validate(await response.json())
search_mock.assert_not_called()
async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
"""
must raise 404 on empty search result
"""
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[])
response_schema = pytest.helpers.schema_response(SearchView.get, code=404)
response = await client.get("/api/v1/service/search", params={"for": ["ahriman"]})
assert response.status == 404
assert not response_schema.validate(await response.json())
async def test_get_join(client: TestClient, mocker: MockerFixture) -> None:
"""
must join search args with space
"""
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
request_schema = pytest.helpers.schema_request(SearchView.get, location="querystring")
payload = {"for": ["ahriman", "maybe"]}
assert not request_schema.validate(payload)
response = await client.get("/api/v1/service/search", params=payload)
assert response.ok
search_mock.assert_called_once_with("ahriman", "maybe")

Some files were not shown because too many files have changed in this diff Show More