165 lines
7.4 KiB
Python
165 lines
7.4 KiB
Python
# #region Test.AppModule.WsAuth [C:3] [TYPE Module] [SEMANTICS test,app,ws,auth,jwt,apikey]
|
|
# @BRIEF Tests for app.py — _authenticate_websocket: JWT/API key auth for WebSocket connections.
|
|
# @RELATION BINDS_TO -> [App.AppModule]
|
|
# @TEST_EDGE: ws_missing_token -> returns False
|
|
# @TEST_EDGE: ws_jwt_auth -> accepts valid JWT
|
|
# @TEST_EDGE: ws_jwt_fallback_apikey -> accepts valid API key after JWT fail
|
|
# @TEST_EDGE: ws_both_fail -> returns False when both auth methods fail
|
|
# @TEST_INVARIANT: ws_auth_gate -> VERIFIED_BY: test_websocket_auth_missing_token, test_websocket_auth_jwt_valid, test_websocket_auth_api_key
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestAuthenticateWebsocket:
|
|
"""_authenticate_websocket — JWT/API key auth."""
|
|
|
|
# #region Test.AppModule.TestMissingToken [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_missing_token(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {}
|
|
assert await _authenticate_websocket(ws, "ws/logs") is False
|
|
# #endregion Test.AppModule.TestMissingToken
|
|
|
|
# #region Test.AppModule.TestJwtValid [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_jwt_valid(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "valid.jwt"}
|
|
with patch("src.core.auth.jwt.decode_token", return_value={"sub": "testuser"}) as md:
|
|
assert await _authenticate_websocket(ws, "ws/logs") is True
|
|
md.assert_called_once_with("valid.jwt")
|
|
# #endregion Test.AppModule.TestJwtValid
|
|
|
|
# #region Test.AppModule.TestJwtNoSub [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_jwt_no_sub(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "no-sub"}
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", return_value={"role": "admin"}),
|
|
patch("src.core.auth.api_key.hash_api_key", return_value="hk"),
|
|
patch("src.core.database.SessionLocal") as msl,
|
|
):
|
|
msl.return_value = MagicMock()
|
|
msl.return_value.query.return_value.filter.return_value.first.return_value = None
|
|
assert await _authenticate_websocket(ws, "ws/logs") is False
|
|
# #endregion Test.AppModule.TestJwtNoSub
|
|
|
|
# #region Test.AppModule.TestJwtFallbackApikey [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_jwt_fallback_apikey(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "valid-key"}
|
|
key = MagicMock(); key.active = True; key.name = "Test"
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", side_effect=Exception("expired")),
|
|
patch("src.core.auth.api_key.hash_api_key", return_value="hk"),
|
|
patch("src.core.database.SessionLocal") as msl,
|
|
):
|
|
msl.return_value = MagicMock()
|
|
msl.return_value.query.return_value.filter.return_value.first.return_value = key
|
|
assert await _authenticate_websocket(ws, "ws/logs") is True
|
|
# #endregion Test.AppModule.TestJwtFallbackApikey
|
|
|
|
# #region Test.AppModule.TestApikeyInactive [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_apikey_inactive(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "inactive"}
|
|
key = MagicMock(); key.active = False
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", side_effect=Exception("fail")),
|
|
patch("src.core.auth.api_key.hash_api_key", return_value="hk"),
|
|
patch("src.core.database.SessionLocal") as msl,
|
|
):
|
|
msl.return_value = MagicMock()
|
|
msl.return_value.query.return_value.filter.return_value.first.return_value = key
|
|
assert await _authenticate_websocket(ws, "ws/logs") is False
|
|
# #endregion Test.AppModule.TestApikeyInactive
|
|
|
|
# #region Test.AppModule.TestBothFail [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_both_fail(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "bad"}
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", side_effect=Exception("invalid")),
|
|
patch("src.core.auth.api_key.hash_api_key", return_value="hk"),
|
|
patch("src.core.database.SessionLocal") as msl,
|
|
):
|
|
msl.return_value = MagicMock()
|
|
msl.return_value.query.return_value.filter.return_value.first.return_value = None
|
|
assert await _authenticate_websocket(ws, "ws/logs") is False
|
|
# #endregion Test.AppModule.TestBothFail
|
|
|
|
|
|
class TestAuthenticateWebsocketApiKeyException:
|
|
"""_authenticate_websocket — API key DB exception is caught."""
|
|
|
|
# #region Test.AppModule.TestApikeyDbException [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_apikey_db_exception(self):
|
|
from src.app import _authenticate_websocket
|
|
ws = MagicMock(); ws.query_params = {"token": "crash"}
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", side_effect=Exception("JWT fail")),
|
|
patch("src.core.auth.api_key.hash_api_key", return_value="hk"),
|
|
patch("src.core.database.SessionLocal") as msl,
|
|
):
|
|
db = MagicMock(); db.query.side_effect = Exception("DB lost")
|
|
msl.return_value = db
|
|
assert await _authenticate_websocket(ws, "ws/logs") is False
|
|
# #endregion Test.AppModule.TestApikeyDbException
|
|
|
|
|
|
# #region Test.AppModule.AuthorizeWebsocket [C:3] [TYPE Class] [SEMANTICS test,app,ws,authorization,rbac]
|
|
# @BRIEF Verify WebSocket authorization grants access only through is_admin or explicit permissions.
|
|
# @RELATION BINDS_TO -> [App.AppModule.AuthorizeWebsocket]
|
|
class TestAuthorizeWebsocket:
|
|
def _authorize(self, user, resource="tasks", action="READ"):
|
|
from src.app import _authorize_websocket
|
|
|
|
ws = MagicMock()
|
|
ws.query_params = {"token": "valid.jwt"}
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.first.return_value = user
|
|
with (
|
|
patch("src.core.auth.jwt.decode_token", return_value={"sub": "testuser"}),
|
|
patch("src.core.database.SessionLocal", return_value=db),
|
|
):
|
|
result = _authorize_websocket(ws, resource, action)
|
|
db.close.assert_called_once()
|
|
return result
|
|
|
|
def test_is_admin_role_bypasses_permission_check(self):
|
|
user = SimpleNamespace(
|
|
is_active=True,
|
|
roles=[SimpleNamespace(name="Admin", is_admin=True, permissions=[])],
|
|
)
|
|
assert self._authorize(user) is True
|
|
|
|
def test_legacy_admin_name_without_flag_does_not_bypass_permissions(self):
|
|
user = SimpleNamespace(
|
|
is_active=True,
|
|
roles=[SimpleNamespace(name="Admin", is_admin=False, permissions=[])],
|
|
)
|
|
assert self._authorize(user) is False
|
|
|
|
def test_explicit_permission_allows_websocket_access(self):
|
|
permission = SimpleNamespace(resource="tasks", action="READ")
|
|
user = SimpleNamespace(
|
|
is_active=True,
|
|
roles=[SimpleNamespace(name="Operator", is_admin=False, permissions=[permission])],
|
|
)
|
|
assert self._authorize(user) is True
|
|
# #endregion Test.AppModule.AuthorizeWebsocket
|
|
# #endregion Test.AppModule.WsAuth
|