mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-04-24 07:17:17 +00:00
Evgenii Alekseev
168b2f6880
* make auth method asyncs * oauth2 demo support * full coverage * update docs
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
from aiohttp.test_utils import TestClient
|
|
from pytest_mock import MockerFixture
|
|
from unittest.mock import MagicMock
|
|
|
|
from ahriman.core.auth.oauth import OAuth
|
|
from ahriman.models.user import User
|
|
|
|
|
|
async def test_get_default_validator(client_with_auth: TestClient) -> None:
|
|
"""
|
|
must return 405 in case if no OAuth enabled
|
|
"""
|
|
get_response = await client_with_auth.get("/user-api/v1/login")
|
|
assert get_response.status == 405
|
|
|
|
|
|
async def test_get_redirect_to_oauth(client_with_auth: TestClient) -> None:
|
|
"""
|
|
must redirect to OAuth service provider in case if no code is supplied
|
|
"""
|
|
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
|
oauth.get_oauth_url.return_value = "https://example.com"
|
|
|
|
get_response = await client_with_auth.get("/user-api/v1/login")
|
|
assert get_response.ok
|
|
oauth.get_oauth_url.assert_called_once()
|
|
|
|
|
|
async def test_get_redirect_to_oauth_empty_code(client_with_auth: TestClient) -> None:
|
|
"""
|
|
must redirect to OAuth service provider in case if empty code is supplied
|
|
"""
|
|
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
|
oauth.get_oauth_url.return_value = "https://example.com"
|
|
|
|
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": ""})
|
|
assert get_response.ok
|
|
oauth.get_oauth_url.assert_called_once()
|
|
|
|
|
|
async def test_get(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
|
"""
|
|
must login user correctly from OAuth
|
|
"""
|
|
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
|
oauth.get_oauth_username.return_value = "user"
|
|
oauth.known_username.return_value = True
|
|
oauth.enabled = False # lol
|
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
|
|
|
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": "code"})
|
|
|
|
assert get_response.ok
|
|
oauth.get_oauth_username.assert_called_with("code")
|
|
oauth.known_username.assert_called_with("user")
|
|
remember_mock.assert_called_once()
|
|
|
|
|
|
async def test_get_unauthorized(client_with_auth: TestClient, mocker: MockerFixture) -> None:
|
|
"""
|
|
must return unauthorized from OAuth
|
|
"""
|
|
oauth = client_with_auth.app["validator"] = MagicMock(spec=OAuth)
|
|
oauth.known_username.return_value = False
|
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
|
|
|
get_response = await client_with_auth.get("/user-api/v1/login", params={"code": "code"})
|
|
|
|
assert get_response.status == 401
|
|
remember_mock.assert_not_called()
|
|
|
|
|
|
async def test_post(client_with_auth: TestClient, user: User, mocker: MockerFixture) -> None:
|
|
"""
|
|
must login user correctly
|
|
"""
|
|
payload = {"username": user.username, "password": user.password}
|
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
|
|
|
post_response = await client_with_auth.post("/user-api/v1/login", json=payload)
|
|
assert post_response.ok
|
|
|
|
post_response = await client_with_auth.post("/user-api/v1/login", data=payload)
|
|
assert post_response.ok
|
|
|
|
remember_mock.assert_called()
|
|
|
|
|
|
async def test_post_skip(client: TestClient, user: User) -> None:
|
|
"""
|
|
must process if no auth configured
|
|
"""
|
|
payload = {"username": user.username, "password": user.password}
|
|
post_response = await client.post("/user-api/v1/login", json=payload)
|
|
assert post_response.ok
|
|
|
|
|
|
async def test_post_unauthorized(client_with_auth: TestClient, user: User, mocker: MockerFixture) -> None:
|
|
"""
|
|
must return unauthorized on invalid auth
|
|
"""
|
|
payload = {"username": user.username, "password": ""}
|
|
remember_mock = mocker.patch("aiohttp_security.remember")
|
|
|
|
post_response = await client_with_auth.post("/user-api/v1/login", json=payload)
|
|
assert post_response.status == 401
|
|
remember_mock.assert_not_called()
|