mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-04-24 23:37:18 +00:00
extract schemas automatically from views
This commit is contained in:
parent
8f4a2547e8
commit
e08ab2db10
@ -213,9 +213,9 @@ class Configuration(configparser.RawConfigParser):
|
|||||||
if self.has_section(full_section):
|
if self.has_section(full_section):
|
||||||
return full_section, section
|
return full_section, section
|
||||||
# okay lets just use section as type
|
# okay lets just use section as type
|
||||||
if not self.has_section(section):
|
if self.has_section(section):
|
||||||
raise configparser.NoSectionError(section)
|
|
||||||
return section, section
|
return section, section
|
||||||
|
raise configparser.NoSectionError(section)
|
||||||
|
|
||||||
def load(self, path: Path) -> None:
|
def load(self, path: Path) -> None:
|
||||||
"""
|
"""
|
||||||
|
@ -173,7 +173,7 @@ class BaseView(View, CorsViewMixin):
|
|||||||
Raises:
|
Raises:
|
||||||
HTTPMethodNotAllowed: in case if there is no GET method implemented
|
HTTPMethodNotAllowed: in case if there is no GET method implemented
|
||||||
"""
|
"""
|
||||||
get_method: Optional[Callable[[], Awaitable[StreamResponse]]] = getattr(self, "get", None)
|
get_method: Optional[Callable[..., Awaitable[StreamResponse]]] = getattr(self, "get", None)
|
||||||
# using if/else in order to suppress mypy warning which doesn't know that
|
# using if/else in order to suppress mypy warning which doesn't know that
|
||||||
# ``_raise_allowed_methods`` raises exception
|
# ``_raise_allowed_methods`` raises exception
|
||||||
if get_method is not None:
|
if get_method is not None:
|
||||||
|
@ -3,8 +3,9 @@ import pytest
|
|||||||
from asyncio import BaseEventLoop
|
from asyncio import BaseEventLoop
|
||||||
from aiohttp.web import Application, Resource, UrlMappingMatchInfo
|
from aiohttp.web import Application, Resource, UrlMappingMatchInfo
|
||||||
from aiohttp.test_utils import TestClient
|
from aiohttp.test_utils import TestClient
|
||||||
|
from marshmallow import Schema
|
||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
from typing import Any, Dict, Optional
|
from typing import Any, Awaitable, Callable, Dict, Optional
|
||||||
from unittest.mock import MagicMock
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
import ahriman.core.auth.helpers
|
import ahriman.core.auth.helpers
|
||||||
@ -54,6 +55,41 @@ def request(application: Application, path: str, method: str, json: Any = None,
|
|||||||
return request_mock
|
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
|
||||||
|
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
|
||||||
|
schema = schemas[code]["schema"]
|
||||||
|
if callable(schema):
|
||||||
|
schema = schema()
|
||||||
|
return schema
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def application(configuration: Configuration, spawner: Spawn, database: SQLite, repository: Repository,
|
def application(configuration: Configuration, spawner: Spawn, database: SQLite, repository: Repository,
|
||||||
mocker: MockerFixture) -> Application:
|
mocker: MockerFixture) -> Application:
|
||||||
|
@ -4,8 +4,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.package_names_schema import PackageNamesSchema
|
|
||||||
from ahriman.web.views.service.add import AddView
|
from ahriman.web.views.service.add import AddView
|
||||||
|
|
||||||
|
|
||||||
@ -23,7 +21,7 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call post request correctly
|
must call post request correctly
|
||||||
"""
|
"""
|
||||||
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
||||||
request_schema = PackageNamesSchema()
|
request_schema = pytest.helpers.schema_request(AddView.post)
|
||||||
|
|
||||||
payload = {"packages": ["ahriman"]}
|
payload = {"packages": ["ahriman"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -37,7 +35,7 @@ async def test_post_empty(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call raise 400 on empty request
|
must call raise 400 on empty request
|
||||||
"""
|
"""
|
||||||
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(AddView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/service/add", json={"packages": [""]})
|
response = await client.post("/api/v1/service/add", json={"packages": [""]})
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -4,9 +4,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.pgp_key_id_schema import PGPKeyIdSchema
|
|
||||||
from ahriman.web.schemas.pgp_key_schema import PGPKeySchema
|
|
||||||
from ahriman.web.views.service.pgp import PGPView
|
from ahriman.web.views.service.pgp import PGPView
|
||||||
|
|
||||||
|
|
||||||
@ -27,8 +24,8 @@ async def test_get(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must retrieve key from the keyserver
|
must retrieve key from the keyserver
|
||||||
"""
|
"""
|
||||||
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", return_value="imported")
|
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", return_value="imported")
|
||||||
request_schema = PGPKeyIdSchema()
|
request_schema = pytest.helpers.schema_request(PGPView.get, location="querystring")
|
||||||
response_schema = PGPKeySchema()
|
response_schema = pytest.helpers.schema_response(PGPView.get)
|
||||||
|
|
||||||
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
|
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -44,7 +41,7 @@ async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must raise 400 on missing parameters
|
must raise 400 on missing parameters
|
||||||
"""
|
"""
|
||||||
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download")
|
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(PGPView.get, code=400)
|
||||||
|
|
||||||
response = await client.get("/api/v1/service/pgp")
|
response = await client.get("/api/v1/service/pgp")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
@ -57,7 +54,7 @@ async def test_get_process_exception(client: TestClient, mocker: MockerFixture)
|
|||||||
must raise 404 on invalid PGP server response
|
must raise 404 on invalid PGP server response
|
||||||
"""
|
"""
|
||||||
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", side_effect=Exception())
|
import_mock = mocker.patch("ahriman.core.sign.gpg.GPG.key_download", side_effect=Exception())
|
||||||
response_schema = ErrorSchema()
|
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"})
|
response = await client.get("/api/v1/service/pgp", params={"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"})
|
||||||
assert response.status == 404
|
assert response.status == 404
|
||||||
@ -70,7 +67,7 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call post request correctly
|
must call post request correctly
|
||||||
"""
|
"""
|
||||||
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
|
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
|
||||||
request_schema = PGPKeyIdSchema()
|
request_schema = pytest.helpers.schema_request(PGPView.post)
|
||||||
|
|
||||||
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
|
payload = {"key": "0xdeadbeaf", "server": "keyserver.ubuntu.com"}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -84,7 +81,7 @@ async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None
|
|||||||
must raise exception on missing key payload
|
must raise exception on missing key payload
|
||||||
"""
|
"""
|
||||||
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
|
import_mock = mocker.patch("ahriman.core.spawn.Spawn.key_import")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(PGPView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/service/pgp")
|
response = await client.post("/api/v1/service/pgp")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -4,8 +4,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.package_names_schema import PackageNamesSchema
|
|
||||||
from ahriman.web.views.service.rebuild import RebuildView
|
from ahriman.web.views.service.rebuild import RebuildView
|
||||||
|
|
||||||
|
|
||||||
@ -23,7 +21,7 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call post request correctly
|
must call post request correctly
|
||||||
"""
|
"""
|
||||||
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
|
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
|
||||||
request_schema = PackageNamesSchema()
|
request_schema = pytest.helpers.schema_request(RebuildView.post)
|
||||||
|
|
||||||
payload = {"packages": ["python", "ahriman"]}
|
payload = {"packages": ["python", "ahriman"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -37,7 +35,7 @@ async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None
|
|||||||
must raise exception on missing packages payload
|
must raise exception on missing packages payload
|
||||||
"""
|
"""
|
||||||
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
|
rebuild_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_rebuild")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(RebuildView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/service/rebuild")
|
response = await client.post("/api/v1/service/rebuild")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -4,8 +4,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.package_names_schema import PackageNamesSchema
|
|
||||||
from ahriman.web.views.service.remove import RemoveView
|
from ahriman.web.views.service.remove import RemoveView
|
||||||
|
|
||||||
|
|
||||||
@ -23,7 +21,7 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call post request correctly
|
must call post request correctly
|
||||||
"""
|
"""
|
||||||
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
|
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
|
||||||
request_schema = PackageNamesSchema()
|
request_schema = pytest.helpers.schema_request(RemoveView.post)
|
||||||
|
|
||||||
payload = {"packages": ["ahriman"]}
|
payload = {"packages": ["ahriman"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -37,7 +35,7 @@ async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None
|
|||||||
must raise exception on missing packages payload
|
must raise exception on missing packages payload
|
||||||
"""
|
"""
|
||||||
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
|
remove_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_remove")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(RemoveView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/service/remove")
|
response = await client.post("/api/v1/service/remove")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -4,8 +4,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.package_names_schema import PackageNamesSchema
|
|
||||||
from ahriman.web.views.service.request import RequestView
|
from ahriman.web.views.service.request import RequestView
|
||||||
|
|
||||||
|
|
||||||
@ -23,7 +21,7 @@ async def test_post(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must call post request correctly
|
must call post request correctly
|
||||||
"""
|
"""
|
||||||
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
||||||
request_schema = PackageNamesSchema()
|
request_schema = pytest.helpers.schema_request(RequestView.post)
|
||||||
|
|
||||||
payload = {"packages": ["ahriman"]}
|
payload = {"packages": ["ahriman"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -37,7 +35,7 @@ async def test_post_exception(client: TestClient, mocker: MockerFixture) -> None
|
|||||||
must raise exception on missing packages payload
|
must raise exception on missing packages payload
|
||||||
"""
|
"""
|
||||||
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
add_mock = mocker.patch("ahriman.core.spawn.Spawn.packages_add")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(RequestView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/service/request")
|
response = await client.post("/api/v1/service/request")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -5,9 +5,6 @@ from pytest_mock import MockerFixture
|
|||||||
|
|
||||||
from ahriman.models.aur_package import AURPackage
|
from ahriman.models.aur_package import AURPackage
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.aur_package_schema import AURPackageSchema
|
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.search_schema import SearchSchema
|
|
||||||
from ahriman.web.views.service.search import SearchView
|
from ahriman.web.views.service.search import SearchView
|
||||||
|
|
||||||
|
|
||||||
@ -25,8 +22,8 @@ async def test_get(client: TestClient, aur_package_ahriman: AURPackage, mocker:
|
|||||||
must call get request correctly
|
must call get request correctly
|
||||||
"""
|
"""
|
||||||
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[aur_package_ahriman])
|
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[aur_package_ahriman])
|
||||||
request_schema = SearchSchema()
|
request_schema = pytest.helpers.schema_request(SearchView.get, location="querystring")
|
||||||
response_schema = AURPackageSchema()
|
response_schema = pytest.helpers.schema_response(SearchView.get)
|
||||||
|
|
||||||
payload = {"for": ["ahriman"]}
|
payload = {"for": ["ahriman"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -42,7 +39,7 @@ async def test_get_exception(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must raise 400 on empty search string
|
must raise 400 on empty search string
|
||||||
"""
|
"""
|
||||||
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
|
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(SearchView.get, code=400)
|
||||||
|
|
||||||
response = await client.get("/api/v1/service/search")
|
response = await client.get("/api/v1/service/search")
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
@ -55,7 +52,7 @@ async def test_get_empty(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must raise 404 on empty search result
|
must raise 404 on empty search result
|
||||||
"""
|
"""
|
||||||
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[])
|
mocker.patch("ahriman.core.alpm.remote.AUR.multisearch", return_value=[])
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(SearchView.get, code=404)
|
||||||
|
|
||||||
response = await client.get("/api/v1/service/search", params={"for": ["ahriman"]})
|
response = await client.get("/api/v1/service/search", params={"for": ["ahriman"]})
|
||||||
assert response.status == 404
|
assert response.status == 404
|
||||||
@ -67,7 +64,7 @@ async def test_get_join(client: TestClient, mocker: MockerFixture) -> None:
|
|||||||
must join search args with space
|
must join search args with space
|
||||||
"""
|
"""
|
||||||
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
|
search_mock = mocker.patch("ahriman.core.alpm.remote.AUR.multisearch")
|
||||||
request_schema = SearchSchema()
|
request_schema = pytest.helpers.schema_request(SearchView.get, location="querystring")
|
||||||
|
|
||||||
payload = {"for": ["ahriman", "maybe"]}
|
payload = {"for": ["ahriman", "maybe"]}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
|
@ -5,9 +5,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from ahriman.models.build_status import BuildStatusEnum
|
from ahriman.models.build_status import BuildStatusEnum
|
||||||
from ahriman.models.package import Package
|
from ahriman.models.package import Package
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.log_schema import LogSchema
|
|
||||||
from ahriman.web.schemas.logs_schema import LogsSchema
|
|
||||||
from ahriman.web.views.status.logs import LogsView
|
from ahriman.web.views.status.logs import LogsView
|
||||||
|
|
||||||
|
|
||||||
@ -57,7 +54,7 @@ async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
|||||||
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
||||||
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
|
await client.post(f"/api/v1/packages/{package_ahriman.base}/logs",
|
||||||
json={"created": 42.0, "message": "message", "process_id": 42})
|
json={"created": 42.0, "message": "message", "process_id": 42})
|
||||||
response_schema = LogsSchema()
|
response_schema = pytest.helpers.schema_response(LogsView.get)
|
||||||
|
|
||||||
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
|
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
|
||||||
assert response.status == 200
|
assert response.status == 200
|
||||||
@ -71,7 +68,7 @@ async def test_get_not_found(client: TestClient, package_ahriman: Package) -> No
|
|||||||
"""
|
"""
|
||||||
must return not found for missing package
|
must return not found for missing package
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(LogsView.get, code=404)
|
||||||
|
|
||||||
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
|
response = await client.get(f"/api/v1/packages/{package_ahriman.base}/logs")
|
||||||
assert response.status == 404
|
assert response.status == 404
|
||||||
@ -84,7 +81,7 @@ async def test_post(client: TestClient, package_ahriman: Package) -> None:
|
|||||||
"""
|
"""
|
||||||
await client.post(f"/api/v1/packages/{package_ahriman.base}",
|
await client.post(f"/api/v1/packages/{package_ahriman.base}",
|
||||||
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
||||||
request_schema = LogSchema()
|
request_schema = pytest.helpers.schema_request(LogsView.post)
|
||||||
|
|
||||||
payload = {"created": 42.0, "message": "message", "process_id": 42}
|
payload = {"created": 42.0, "message": "message", "process_id": 42}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -100,7 +97,7 @@ async def test_post_exception(client: TestClient, package_ahriman: Package) -> N
|
|||||||
"""
|
"""
|
||||||
must raise exception on invalid payload
|
must raise exception on invalid payload
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(LogsView.post, code=400)
|
||||||
|
|
||||||
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/logs", json={})
|
response = await client.post(f"/api/v1/packages/{package_ahriman.base}/logs", json={})
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
|
@ -5,8 +5,6 @@ from aiohttp.test_utils import TestClient
|
|||||||
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
from ahriman.models.build_status import BuildStatus, BuildStatusEnum
|
||||||
from ahriman.models.package import Package
|
from ahriman.models.package import Package
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.package_status_schema import PackageStatusSchema, PackageStatusSimplifiedSchema
|
|
||||||
from ahriman.web.views.status.package import PackageView
|
from ahriman.web.views.status.package import PackageView
|
||||||
|
|
||||||
|
|
||||||
@ -66,7 +64,7 @@ async def test_get(client: TestClient, package_ahriman: Package, package_python_
|
|||||||
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
||||||
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
|
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
|
||||||
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
|
||||||
response_schema = PackageStatusSchema()
|
response_schema = pytest.helpers.schema_response(PackageView.get)
|
||||||
|
|
||||||
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
|
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
|
||||||
assert response.ok
|
assert response.ok
|
||||||
@ -82,7 +80,7 @@ async def test_get_not_found(client: TestClient, package_ahriman: Package) -> No
|
|||||||
"""
|
"""
|
||||||
must return Not Found for unknown package
|
must return Not Found for unknown package
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(PackageView.get, code=404)
|
||||||
|
|
||||||
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
|
response = await client.get(f"/api/v1/packages/{package_ahriman.base}")
|
||||||
assert response.status == 404
|
assert response.status == 404
|
||||||
@ -93,7 +91,7 @@ async def test_post(client: TestClient, package_ahriman: Package) -> None:
|
|||||||
"""
|
"""
|
||||||
must update package status
|
must update package status
|
||||||
"""
|
"""
|
||||||
request_schema = PackageStatusSimplifiedSchema()
|
request_schema = pytest.helpers.schema_request(PackageView.post)
|
||||||
|
|
||||||
payload = {"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()}
|
payload = {"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -108,7 +106,7 @@ async def test_post_exception(client: TestClient, package_ahriman: Package) -> N
|
|||||||
"""
|
"""
|
||||||
must raise exception on invalid payload
|
must raise exception on invalid payload
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(PackageView.post, code=400)
|
||||||
|
|
||||||
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json={})
|
response = await client.post(f"/api/v1/packages/{package_ahriman.base}", json={})
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
@ -119,7 +117,7 @@ async def test_post_light(client: TestClient, package_ahriman: Package) -> None:
|
|||||||
"""
|
"""
|
||||||
must update package status only
|
must update package status only
|
||||||
"""
|
"""
|
||||||
request_schema = PackageStatusSimplifiedSchema()
|
request_schema = pytest.helpers.schema_request(PackageView.post)
|
||||||
|
|
||||||
payload = {"status": BuildStatusEnum.Unknown.value, "package": package_ahriman.view()}
|
payload = {"status": BuildStatusEnum.Unknown.value, "package": package_ahriman.view()}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -144,8 +142,8 @@ async def test_post_not_found(client: TestClient, package_ahriman: Package) -> N
|
|||||||
"""
|
"""
|
||||||
must raise exception on status update for unknown package
|
must raise exception on status update for unknown package
|
||||||
"""
|
"""
|
||||||
request_schema = PackageStatusSimplifiedSchema()
|
request_schema = pytest.helpers.schema_request(PackageView.post)
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(PackageView.post, code=400)
|
||||||
|
|
||||||
payload = {"status": BuildStatusEnum.Success.value}
|
payload = {"status": BuildStatusEnum.Success.value}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
|
@ -6,7 +6,6 @@ from pytest_mock import MockerFixture
|
|||||||
from ahriman.models.build_status import BuildStatusEnum
|
from ahriman.models.build_status import BuildStatusEnum
|
||||||
from ahriman.models.package import Package
|
from ahriman.models.package import Package
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.package_status_schema import PackageStatusSchema
|
|
||||||
from ahriman.web.views.status.packages import PackagesView
|
from ahriman.web.views.status.packages import PackagesView
|
||||||
|
|
||||||
|
|
||||||
@ -30,7 +29,7 @@ async def test_get(client: TestClient, package_ahriman: Package, package_python_
|
|||||||
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
||||||
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
|
await client.post(f"/api/v1/packages/{package_python_schedule.base}",
|
||||||
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_python_schedule.view()})
|
||||||
response_schema = PackageStatusSchema()
|
response_schema = pytest.helpers.schema_response(PackagesView.get)
|
||||||
|
|
||||||
response = await client.get("/api/v1/packages")
|
response = await client.get("/api/v1/packages")
|
||||||
assert response.ok
|
assert response.ok
|
||||||
|
@ -9,9 +9,6 @@ from ahriman.models.build_status import BuildStatusEnum
|
|||||||
from ahriman.models.internal_status import InternalStatus
|
from ahriman.models.internal_status import InternalStatus
|
||||||
from ahriman.models.package import Package
|
from ahriman.models.package import Package
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.internal_status_schema import InternalStatusSchema
|
|
||||||
from ahriman.web.schemas.status_schema import StatusSchema
|
|
||||||
from ahriman.web.views.status.status import StatusView
|
from ahriman.web.views.status.status import StatusView
|
||||||
|
|
||||||
|
|
||||||
@ -33,7 +30,7 @@ async def test_get(client: TestClient, package_ahriman: Package) -> None:
|
|||||||
"""
|
"""
|
||||||
await client.post(f"/api/v1/packages/{package_ahriman.base}",
|
await client.post(f"/api/v1/packages/{package_ahriman.base}",
|
||||||
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
json={"status": BuildStatusEnum.Success.value, "package": package_ahriman.view()})
|
||||||
response_schema = InternalStatusSchema()
|
response_schema = pytest.helpers.schema_response(StatusView.get)
|
||||||
|
|
||||||
response = await client.get("/api/v1/status")
|
response = await client.get("/api/v1/status")
|
||||||
assert response.ok
|
assert response.ok
|
||||||
@ -49,7 +46,7 @@ async def test_post(client: TestClient) -> None:
|
|||||||
"""
|
"""
|
||||||
must update service status correctly
|
must update service status correctly
|
||||||
"""
|
"""
|
||||||
request_schema = StatusSchema()
|
request_schema = pytest.helpers.schema_request(StatusView.post)
|
||||||
|
|
||||||
payload = {"status": BuildStatusEnum.Success.value}
|
payload = {"status": BuildStatusEnum.Success.value}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -67,7 +64,7 @@ async def test_post_exception(client: TestClient) -> None:
|
|||||||
"""
|
"""
|
||||||
must raise exception on invalid payload
|
must raise exception on invalid payload
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(StatusView.post, code=400)
|
||||||
|
|
||||||
response = await client.post("/api/v1/status", json={})
|
response = await client.post("/api/v1/status", json={})
|
||||||
assert response.status == 400
|
assert response.status == 400
|
||||||
@ -80,7 +77,7 @@ async def test_post_exception_inside(client: TestClient, mocker: MockerFixture)
|
|||||||
"""
|
"""
|
||||||
payload = {"status": BuildStatusEnum.Success.value}
|
payload = {"status": BuildStatusEnum.Success.value}
|
||||||
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
mocker.patch("ahriman.core.status.watcher.Watcher.update_self", side_effect=Exception())
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(StatusView.post, code=500)
|
||||||
|
|
||||||
response = await client.post("/api/v1/status", json=payload)
|
response = await client.post("/api/v1/status", json=payload)
|
||||||
assert response.status == 500
|
assert response.status == 500
|
||||||
|
@ -5,9 +5,6 @@ from pytest_mock import MockerFixture
|
|||||||
|
|
||||||
from ahriman.models.user import User
|
from ahriman.models.user import User
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.schemas.login_schema import LoginSchema
|
|
||||||
from ahriman.web.schemas.oauth2_schema import OAuth2Schema
|
|
||||||
from ahriman.web.views.user.login import LoginView
|
from ahriman.web.views.user.login import LoginView
|
||||||
|
|
||||||
|
|
||||||
@ -34,7 +31,7 @@ async def test_get_redirect_to_oauth(client_with_oauth_auth: TestClient) -> None
|
|||||||
"""
|
"""
|
||||||
oauth = client_with_oauth_auth.app["validator"]
|
oauth = client_with_oauth_auth.app["validator"]
|
||||||
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
||||||
request_schema = OAuth2Schema()
|
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
|
||||||
|
|
||||||
payload = {}
|
payload = {}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -49,7 +46,7 @@ async def test_get_redirect_to_oauth_empty_code(client_with_oauth_auth: TestClie
|
|||||||
"""
|
"""
|
||||||
oauth = client_with_oauth_auth.app["validator"]
|
oauth = client_with_oauth_auth.app["validator"]
|
||||||
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
oauth.get_oauth_url.return_value = "https://httpbin.org"
|
||||||
request_schema = OAuth2Schema()
|
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
|
||||||
|
|
||||||
payload = {"code": ""}
|
payload = {"code": ""}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -68,7 +65,7 @@ async def test_get(client_with_oauth_auth: TestClient, mocker: MockerFixture) ->
|
|||||||
oauth.enabled = False # lol
|
oauth.enabled = False # lol
|
||||||
oauth.max_age = 60
|
oauth.max_age = 60
|
||||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||||
request_schema = OAuth2Schema()
|
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
|
||||||
|
|
||||||
payload = {"code": "code"}
|
payload = {"code": "code"}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -89,7 +86,7 @@ async def test_get_unauthorized(client_with_oauth_auth: TestClient, mocker: Mock
|
|||||||
oauth.known_username.return_value = False
|
oauth.known_username.return_value = False
|
||||||
oauth.max_age = 60
|
oauth.max_age = 60
|
||||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(LoginView.post, code=401)
|
||||||
|
|
||||||
response = await client_with_oauth_auth.get(
|
response = await client_with_oauth_auth.get(
|
||||||
"/api/v1/login", params={"code": "code"}, headers={"accept": "application/json"})
|
"/api/v1/login", params={"code": "code"}, headers={"accept": "application/json"})
|
||||||
@ -105,7 +102,7 @@ async def test_post(client_with_auth: TestClient, user: User, mocker: MockerFixt
|
|||||||
"""
|
"""
|
||||||
payload = {"username": user.username, "password": user.password}
|
payload = {"username": user.username, "password": user.password}
|
||||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||||
request_schema = LoginSchema()
|
request_schema = pytest.helpers.schema_request(LoginView.post)
|
||||||
|
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
|
|
||||||
@ -122,7 +119,7 @@ async def test_post_skip(client: TestClient, user: User) -> None:
|
|||||||
"""
|
"""
|
||||||
must process if no auth configured
|
must process if no auth configured
|
||||||
"""
|
"""
|
||||||
request_schema = LoginSchema()
|
request_schema = pytest.helpers.schema_request(LoginView.post)
|
||||||
|
|
||||||
payload = {"username": user.username, "password": user.password}
|
payload = {"username": user.username, "password": user.password}
|
||||||
assert not request_schema.validate(payload)
|
assert not request_schema.validate(payload)
|
||||||
@ -134,7 +131,7 @@ async def test_post_unauthorized(client_with_auth: TestClient, user: User, mocke
|
|||||||
"""
|
"""
|
||||||
must return unauthorized on invalid auth
|
must return unauthorized on invalid auth
|
||||||
"""
|
"""
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(LoginView.post, code=401)
|
||||||
|
|
||||||
payload = {"username": user.username, "password": ""}
|
payload = {"username": user.username, "password": ""}
|
||||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||||
|
@ -5,7 +5,6 @@ from aiohttp.web import HTTPUnauthorized
|
|||||||
from pytest_mock import MockerFixture
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
from ahriman.models.user_access import UserAccess
|
from ahriman.models.user_access import UserAccess
|
||||||
from ahriman.web.schemas.error_schema import ErrorSchema
|
|
||||||
from ahriman.web.views.user.logout import LogoutView
|
from ahriman.web.views.user.logout import LogoutView
|
||||||
|
|
||||||
|
|
||||||
@ -36,7 +35,7 @@ async def test_post_unauthorized(client_with_auth: TestClient, mocker: MockerFix
|
|||||||
"""
|
"""
|
||||||
mocker.patch("aiohttp_security.check_authorized", side_effect=HTTPUnauthorized())
|
mocker.patch("aiohttp_security.check_authorized", side_effect=HTTPUnauthorized())
|
||||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||||
response_schema = ErrorSchema()
|
response_schema = pytest.helpers.schema_response(LogoutView.post, code=401)
|
||||||
|
|
||||||
response = await client_with_auth.post("/api/v1/logout", headers={"accept": "application/json"})
|
response = await client_with_auth.post("/api/v1/logout", headers={"accept": "application/json"})
|
||||||
assert response.status == 401
|
assert response.status == 401
|
||||||
|
Loading…
Reference in New Issue
Block a user