mirror of
https://github.com/arcan1s/ahriman.git
synced 2025-04-26 16:27:17 +00:00
Compare commits
2 Commits
983d84ffd9
...
571a83ef73
Author | SHA1 | Date | |
---|---|---|---|
571a83ef73 | |||
2188b38c85 |
@ -19,9 +19,8 @@
|
||||
#
|
||||
try:
|
||||
import aiohttp_security
|
||||
_has_aiohttp_security = True
|
||||
except ImportError:
|
||||
_has_aiohttp_security = False
|
||||
aiohttp_security = None # type: ignore[assignment]
|
||||
|
||||
from typing import Any
|
||||
|
||||
@ -40,7 +39,7 @@ async def authorized_userid(*args: Any, **kwargs: Any) -> Any:
|
||||
Returns:
|
||||
Any: ``None`` in case if no aiohttp_security module found and function call otherwise
|
||||
"""
|
||||
if _has_aiohttp_security:
|
||||
if aiohttp_security is not None:
|
||||
return await aiohttp_security.authorized_userid(*args, **kwargs) # pylint: disable=no-value-for-parameter
|
||||
return None
|
||||
|
||||
@ -56,7 +55,7 @@ async def check_authorized(*args: Any, **kwargs: Any) -> Any:
|
||||
Returns:
|
||||
Any: ``None`` in case if no aiohttp_security module found and function call otherwise
|
||||
"""
|
||||
if _has_aiohttp_security:
|
||||
if aiohttp_security is not None:
|
||||
return await aiohttp_security.check_authorized(*args, **kwargs) # pylint: disable=no-value-for-parameter
|
||||
return None
|
||||
|
||||
@ -72,7 +71,7 @@ async def forget(*args: Any, **kwargs: Any) -> Any:
|
||||
Returns:
|
||||
Any: ``None`` in case if no aiohttp_security module found and function call otherwise
|
||||
"""
|
||||
if _has_aiohttp_security:
|
||||
if aiohttp_security is not None:
|
||||
return await aiohttp_security.forget(*args, **kwargs) # pylint: disable=no-value-for-parameter
|
||||
return None
|
||||
|
||||
@ -88,6 +87,6 @@ async def remember(*args: Any, **kwargs: Any) -> Any:
|
||||
Returns:
|
||||
Any: ``None`` in case if no aiohttp_security module found and function call otherwise
|
||||
"""
|
||||
if _has_aiohttp_security:
|
||||
if aiohttp_security is not None:
|
||||
return await aiohttp_security.remember(*args, **kwargs) # pylint: disable=no-value-for-parameter
|
||||
return None
|
||||
|
@ -10,33 +10,21 @@ def test_import_aiohttp_security() -> None:
|
||||
"""
|
||||
must import aiohttp_security correctly
|
||||
"""
|
||||
assert helpers._has_aiohttp_security
|
||||
|
||||
|
||||
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 not helpers._has_aiohttp_security
|
||||
assert helpers.aiohttp_security
|
||||
|
||||
|
||||
async def test_authorized_userid_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call authorized_userid from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
authorized_userid_mock = mocker.patch("aiohttp_security.authorized_userid")
|
||||
mocker.patch.object(helpers, "aiohttp_security", None)
|
||||
await helpers.authorized_userid()
|
||||
authorized_userid_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_authorized_userid_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call authorized_userid from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
authorized_userid_mock = mocker.patch("aiohttp_security.authorized_userid")
|
||||
await helpers.authorized_userid()
|
||||
authorized_userid_mock.assert_called_once_with()
|
||||
@ -46,17 +34,14 @@ async def test_check_authorized_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call check_authorized from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
check_authorized_mock = mocker.patch("aiohttp_security.check_authorized")
|
||||
mocker.patch.object(helpers, "aiohttp_security", None)
|
||||
await helpers.check_authorized()
|
||||
check_authorized_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_check_authorized_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call check_authorized from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
check_authorized_mock = mocker.patch("aiohttp_security.check_authorized")
|
||||
await helpers.check_authorized()
|
||||
check_authorized_mock.assert_called_once_with()
|
||||
@ -66,17 +51,14 @@ async def test_forget_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call forget from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
mocker.patch.object(helpers, "aiohttp_security", None)
|
||||
await helpers.forget()
|
||||
forget_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_forget_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call forget from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
await helpers.forget()
|
||||
forget_mock.assert_called_once_with()
|
||||
@ -86,17 +68,23 @@ async def test_remember_dummy(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must not call remember from library if not enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
mocker.patch.object(helpers, "aiohttp_security", None)
|
||||
await helpers.remember()
|
||||
remember_mock.assert_not_called()
|
||||
|
||||
|
||||
async def test_remember_library(mocker: MockerFixture) -> None:
|
||||
"""
|
||||
must call remember from library if enabled
|
||||
"""
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
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
|
||||
|
@ -129,7 +129,7 @@ def application(configuration: Configuration, spawner: Spawn, database: SQLite,
|
||||
configuration.set_option("web", "port", "8080")
|
||||
mocker.patch("ahriman.core.database.SQLite.load", return_value=database)
|
||||
mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
mocker.patch.object(helpers, "aiohttp_security", None)
|
||||
_, repository_id = configuration.check_loaded()
|
||||
|
||||
return setup_server(configuration, spawner, [repository_id])
|
||||
@ -155,7 +155,6 @@ def application_with_auth(configuration: Configuration, user: User, spawner: Spa
|
||||
configuration.set_option("web", "port", "8080")
|
||||
mocker.patch("ahriman.core.database.SQLite.load", return_value=database)
|
||||
mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", True)
|
||||
_, repository_id = configuration.check_loaded()
|
||||
application = setup_server(configuration, spawner, [repository_id])
|
||||
|
||||
@ -165,31 +164,6 @@ def application_with_auth(configuration: Configuration, user: User, spawner: Spa
|
||||
return application
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def application_with_debug(configuration: Configuration, spawner: Spawn, database: SQLite,
|
||||
mocker: MockerFixture) -> Application:
|
||||
"""
|
||||
application fixture with debug enabled
|
||||
|
||||
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", "debug", "yes")
|
||||
configuration.set_option("web", "port", "8080")
|
||||
mocker.patch("ahriman.core.database.SQLite.load", return_value=database)
|
||||
mocker.patch("aiohttp_apispec.setup_aiohttp_apispec")
|
||||
mocker.patch.object(helpers, "_has_aiohttp_security", False)
|
||||
_, repository_id = configuration.check_loaded()
|
||||
|
||||
return setup_server(configuration, spawner, [repository_id])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(application: Application, event_loop: BaseEventLoop, aiohttp_client: Any,
|
||||
mocker: MockerFixture) -> TestClient:
|
||||
|
@ -81,7 +81,7 @@ async def test_get(client_with_oauth_auth: TestClient, mocker: MockerFixture) ->
|
||||
oauth.known_username.return_value = True
|
||||
oauth.enabled = False # lol
|
||||
oauth.max_age = 60
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
remember_mock = mocker.patch("ahriman.web.views.v1.user.login.remember")
|
||||
request_schema = pytest.helpers.schema_request(LoginView.get, location="querystring")
|
||||
|
||||
payload = {"code": "code"}
|
||||
@ -102,7 +102,7 @@ async def test_get_unauthorized(client_with_oauth_auth: TestClient, mocker: Mock
|
||||
oauth = client_with_oauth_auth.app[AuthKey]
|
||||
oauth.known_username.return_value = False
|
||||
oauth.max_age = 60
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
remember_mock = mocker.patch("ahriman.web.views.v1.user.login.remember")
|
||||
response_schema = pytest.helpers.schema_response(LoginView.post, code=401)
|
||||
|
||||
response = await client_with_oauth_auth.get(
|
||||
@ -118,7 +118,7 @@ async def test_post(client_with_auth: TestClient, user: User, mocker: MockerFixt
|
||||
must log in user correctly
|
||||
"""
|
||||
payload = {"username": user.username, "password": user.password}
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
remember_mock = mocker.patch("ahriman.web.views.v1.user.login.remember")
|
||||
request_schema = pytest.helpers.schema_request(LoginView.post)
|
||||
|
||||
assert not request_schema.validate(payload)
|
||||
@ -148,7 +148,7 @@ async def test_post_unauthorized(client_with_auth: TestClient, user: User, mocke
|
||||
response_schema = pytest.helpers.schema_response(LoginView.post, code=401)
|
||||
|
||||
payload = {"username": user.username, "password": ""}
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
remember_mock = mocker.patch("ahriman.web.views.v1.user.login.remember")
|
||||
|
||||
response = await client_with_auth.post("/api/v1/login", json=payload, headers={"accept": "application/json"})
|
||||
assert response.status == 401
|
||||
@ -161,7 +161,7 @@ async def test_post_invalid_json(client_with_auth: TestClient, mocker: MockerFix
|
||||
must return unauthorized on invalid payload
|
||||
"""
|
||||
response_schema = pytest.helpers.schema_response(LoginView.post, code=400)
|
||||
remember_mock = mocker.patch("aiohttp_security.remember")
|
||||
remember_mock = mocker.patch("ahriman.web.views.v1.user.login.remember")
|
||||
|
||||
response = await client_with_auth.post("/api/v1/login")
|
||||
assert response.status == 400
|
||||
|
@ -28,8 +28,8 @@ async def test_post(client_with_auth: TestClient, mocker: MockerFixture) -> None
|
||||
"""
|
||||
must log out user correctly
|
||||
"""
|
||||
mocker.patch("aiohttp_security.check_authorized")
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
mocker.patch("ahriman.web.views.v1.user.logout.check_authorized")
|
||||
forget_mock = mocker.patch("ahriman.web.views.v1.user.logout.forget")
|
||||
|
||||
response = await client_with_auth.post("/api/v1/logout")
|
||||
assert response.ok
|
||||
@ -40,8 +40,8 @@ async def test_post_unauthorized(client_with_auth: TestClient, mocker: MockerFix
|
||||
"""
|
||||
must raise exception if unauthorized
|
||||
"""
|
||||
mocker.patch("aiohttp_security.check_authorized", side_effect=HTTPUnauthorized())
|
||||
forget_mock = mocker.patch("aiohttp_security.forget")
|
||||
mocker.patch("ahriman.web.views.v1.user.logout.check_authorized", side_effect=HTTPUnauthorized())
|
||||
forget_mock = mocker.patch("ahriman.web.views.v1.user.logout.forget")
|
||||
response_schema = pytest.helpers.schema_response(LogoutView.post, code=401)
|
||||
|
||||
response = await client_with_auth.post("/api/v1/logout", headers={"accept": "application/json"})
|
||||
|
Loading…
Reference in New Issue
Block a user