mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-07-23 10:49:55 +00:00
@ -1,12 +1,16 @@
|
||||
import pytest
|
||||
|
||||
from asyncio import BaseEventLoop
|
||||
from aiohttp import web
|
||||
from aiohttp.test_utils import TestClient
|
||||
from collections import namedtuple
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import ahriman.core.auth.helpers
|
||||
|
||||
from ahriman.core.auth import OAuth
|
||||
from ahriman.core.configuration import Configuration
|
||||
from ahriman.core.database import SQLite
|
||||
from ahriman.core.spawn import Spawn
|
||||
@ -105,3 +109,61 @@ def application_with_debug(configuration: Configuration, user: User, spawner: Sp
|
||||
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
|
||||
mocker.patch.object(ahriman.core.auth.helpers, "_has_aiohttp_security", False)
|
||||
return setup_service("x86_64", configuration, spawner)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture
|
||||
|
||||
Args:
|
||||
application(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_oauth_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
application_with_auth["validator"] = MagicMock(spec=OAuth)
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
@ -1,6 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp import web
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
@ -63,11 +64,43 @@ async def test_auth_handler_api(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Read, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_static(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow static calls
|
||||
"""
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
await client_with_auth.get("/static/favicon.ico")
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_unauthorized(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow pages with unauthorized access
|
||||
"""
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
await client_with_auth.get("/")
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_allow_read_only(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must allow pages with allow read only flag
|
||||
"""
|
||||
aiohttp_request = pytest.helpers.request("", "/status-api", "GET")
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler(allow_read_only=True)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_auth_handler_api_no_method(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must ask for write permission if handler does not have get_permission method
|
||||
@ -77,9 +110,9 @@ async def test_auth_handler_api_no_method(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission = None
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_api_post(mocker: MockerFixture) -> None:
|
||||
@ -88,12 +121,12 @@ async def test_auth_handler_api_post(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
aiohttp_request = pytest.helpers.request("", "/status-api", "POST")
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Write
|
||||
request_handler.get_permission.return_value = UserAccess.Full
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
async def test_auth_handler_read(mocker: MockerFixture) -> None:
|
||||
@ -106,7 +139,7 @@ async def test_auth_handler_read(mocker: MockerFixture) -> None:
|
||||
request_handler.get_permission.return_value = UserAccess.Read
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Read, aiohttp_request.path)
|
||||
|
||||
@ -118,12 +151,12 @@ async def test_auth_handler_write(mocker: MockerFixture) -> None:
|
||||
for method in ("CONNECT", "DELETE", "PATCH", "POST", "PUT", "TRACE"):
|
||||
aiohttp_request = pytest.helpers.request("", "", method)
|
||||
request_handler = AsyncMock()
|
||||
request_handler.get_permission.return_value = UserAccess.Write
|
||||
request_handler.get_permission.return_value = UserAccess.Full
|
||||
check_permission_mock = mocker.patch("aiohttp_security.check_permission")
|
||||
|
||||
handler = auth_handler()
|
||||
handler = auth_handler(allow_read_only=False)
|
||||
await handler(aiohttp_request, request_handler)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Write, aiohttp_request.path)
|
||||
check_permission_mock.assert_called_once_with(aiohttp_request, UserAccess.Full, aiohttp_request.path)
|
||||
|
||||
|
||||
def test_setup_auth(application_with_auth: web.Application, auth: Auth, mocker: MockerFixture) -> None:
|
||||
|
@ -1,13 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp import web
|
||||
from asyncio import BaseEventLoop
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from ahriman.core.auth import OAuth
|
||||
from ahriman.web.views.base import BaseView
|
||||
|
||||
|
||||
@ -23,61 +17,3 @@ def base(application: web.Application) -> BaseView:
|
||||
BaseView: generated base view fixture
|
||||
"""
|
||||
return BaseView(pytest.helpers.request(application, "", ""))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture
|
||||
|
||||
Args:
|
||||
application(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_oauth_auth(application_with_auth: web.Application, event_loop: BaseEventLoop,
|
||||
aiohttp_client: Any, mocker: MockerFixture) -> TestClient:
|
||||
"""
|
||||
web client fixture with full authorization functions
|
||||
|
||||
Args:
|
||||
application_with_auth(web.Application): application fixture
|
||||
event_loop(BaseEventLoop): context event loop
|
||||
aiohttp_client(Any): aiohttp client fixture
|
||||
mocker(MockerFixture): mocker object
|
||||
|
||||
Returns:
|
||||
TestClient: web client test instance
|
||||
"""
|
||||
mocker.patch("pathlib.Path.iterdir", return_value=[])
|
||||
application_with_auth["validator"] = MagicMock(spec=OAuth)
|
||||
return event_loop.run_until_complete(aiohttp_client(application_with_auth))
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AddView.get_permission(request) == UserAccess.Write
|
||||
assert await AddView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await RemoveView.get_permission(request) == UserAccess.Write
|
||||
assert await RemoveView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -13,7 +13,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await RequestView.get_permission(request) == UserAccess.Read
|
||||
assert await RequestView.get_permission(request) == UserAccess.Reporter
|
||||
|
||||
|
||||
async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await SearchView.get_permission(request) == UserAccess.Read
|
||||
assert await SearchView.get_permission(request) == UserAccess.Reporter
|
||||
|
||||
|
||||
async def test_get(client: TestClient, aur_package_ahriman: AURPackage, mocker: MockerFixture) -> None:
|
||||
|
@ -1,65 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||
from ahriman.models.user_access import UserAccess
|
||||
from ahriman.web.views.status.ahriman import AhrimanView
|
||||
|
||||
|
||||
async def test_get_permission() -> None:
|
||||
"""
|
||||
must return correct permission for the request
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AhrimanView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await AhrimanView.get_permission(request) == UserAccess.Write
|
||||
|
||||
|
||||
async def test_get(client: TestClient) -> None:
|
||||
"""
|
||||
must return valid service status
|
||||
"""
|
||||
response = await client.get("/status-api/v1/ahriman")
|
||||
status = BuildStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status == BuildStatusEnum.Unknown
|
||||
|
||||
|
||||
async def test_post(client: TestClient) -> None:
|
||||
"""
|
||||
must update service status correctly
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
post_response = await client.post("/status-api/v1/ahriman", json=payload)
|
||||
assert post_response.status == 204
|
||||
|
||||
response = await client.get("/status-api/v1/ahriman")
|
||||
status = BuildStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
async def test_post_exception(client: TestClient) -> None:
|
||||
"""
|
||||
must raise exception on invalid payload
|
||||
"""
|
||||
post_response = await client.post("/status-api/v1/ahriman", json={})
|
||||
assert post_response.status == 400
|
||||
|
||||
|
||||
async def test_post_exception_inside(client: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
exception handler must handle 500 errors
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||
|
||||
post_response = await client.post("/status-api/v1/ahriman", json=payload)
|
||||
assert post_response.status == 500
|
@ -17,7 +17,7 @@ async def test_get_permission() -> None:
|
||||
assert await PackageView.get_permission(request) == UserAccess.Read
|
||||
for method in ("DELETE", "POST"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await PackageView.get_permission(request) == UserAccess.Write
|
||||
assert await PackageView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
|
@ -18,7 +18,7 @@ async def test_get_permission() -> None:
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Write
|
||||
assert await PackagesView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package, package_python_schedule: Package) -> None:
|
||||
|
@ -1,10 +1,12 @@
|
||||
import pytest
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
import ahriman.version as version
|
||||
|
||||
from ahriman.models.build_status import BuildStatusEnum
|
||||
from ahriman.models.internal_status import InternalStatus
|
||||
from ahriman.models.package import Package
|
||||
from ahriman.models.user_access import UserAccess
|
||||
from ahriman.web.views.status.status import StatusView
|
||||
@ -17,6 +19,9 @@ async def test_get_permission() -> None:
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await StatusView.get_permission(request) == UserAccess.Read
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await StatusView.get_permission(request) == UserAccess.Full
|
||||
|
||||
|
||||
async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
||||
@ -33,3 +38,37 @@ async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
||||
assert json["version"] == version.__version__
|
||||
assert json["packages"]
|
||||
assert json["packages"]["total"] == 1
|
||||
|
||||
|
||||
async def test_post(client: TestClient) -> None:
|
||||
"""
|
||||
must update service status correctly
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
post_response = await client.post("/status-api/v1/status", json=payload)
|
||||
assert post_response.status == 204
|
||||
|
||||
response = await client.get("/status-api/v1/status")
|
||||
status = InternalStatus.from_json(await response.json())
|
||||
|
||||
assert response.ok
|
||||
assert status.status.status == BuildStatusEnum.Success
|
||||
|
||||
|
||||
async def test_post_exception(client: TestClient) -> None:
|
||||
"""
|
||||
must raise exception on invalid payload
|
||||
"""
|
||||
post_response = await client.post("/status-api/v1/status", json={})
|
||||
assert post_response.status == 400
|
||||
|
||||
|
||||
async def test_post_exception_inside(client: TestClient, mocker: MockerFixture) -> None:
|
||||
"""
|
||||
exception handler must handle 500 errors
|
||||
"""
|
||||
payload = {"status": BuildStatusEnum.Success.value}
|
||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||
|
||||
post_response = await client.post("/status-api/v1/status", json=payload)
|
||||
assert post_response.status == 500
|
||||
|
@ -12,7 +12,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "HEAD"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await IndexView.get_permission(request) == UserAccess.Safe
|
||||
assert await IndexView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_get(client_with_auth: TestClient) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("GET", "POST"):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await LoginView.get_permission(request) == UserAccess.Safe
|
||||
assert await LoginView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_get_default_validator(client_with_auth: TestClient) -> None:
|
||||
|
@ -14,7 +14,7 @@ async def test_get_permission() -> None:
|
||||
"""
|
||||
for method in ("POST",):
|
||||
request = pytest.helpers.request("", "", method)
|
||||
assert await LogoutView.get_permission(request) == UserAccess.Safe
|
||||
assert await LogoutView.get_permission(request) == UserAccess.Unauthorized
|
||||
|
||||
|
||||
async def test_post(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
||||
|
Reference in New Issue
Block a user