162 lines
5.2 KiB
Python
162 lines
5.2 KiB
Python
import os
|
|
import pytest
|
|
import importlib
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from fastapi import HTTPException, Depends, Request
|
|
from app.auth.dependencies import get_current_user, require_roles, oauth2_scheme
|
|
from app.models.auth import CognitoUser
|
|
|
|
# Mock user for testing
|
|
TEST_USER = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=["admin", "user"],
|
|
groups=["test_group"]
|
|
)
|
|
|
|
# Mock the underlying get_user_from_token function
|
|
def mock_get_user_from_token(token: str) -> CognitoUser:
|
|
if token == "valid_token":
|
|
return TEST_USER
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
# Mock endpoint for testing the require_roles decorator
|
|
@require_roles("admin")
|
|
async def mock_protected_endpoint(user: CognitoUser = Depends(get_current_user)):
|
|
return {"message": "Success", "user": user.username}
|
|
|
|
# Patch the get_user_from_token function for testing
|
|
@pytest.fixture(autouse=True)
|
|
def mock_auth(monkeypatch):
|
|
monkeypatch.setattr("app.auth.dependencies.get_user_from_token", mock_get_user_from_token)
|
|
|
|
# Test get_current_user dependency
|
|
def test_get_current_user_success():
|
|
user = get_current_user("valid_token")
|
|
assert user == TEST_USER
|
|
assert user.username == "testuser"
|
|
assert user.roles == ["admin", "user"]
|
|
|
|
def test_get_current_user_invalid_token():
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_current_user("invalid_token")
|
|
assert exc.value.status_code == 401
|
|
|
|
# Test require_roles decorator
|
|
@pytest.mark.asyncio
|
|
async def test_require_roles_success():
|
|
# Create test user with required role
|
|
user = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=["admin"],
|
|
groups=[]
|
|
)
|
|
|
|
result = await mock_protected_endpoint(user=user)
|
|
assert result == {"message": "Success", "user": "testuser"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_require_roles_missing_role():
|
|
# Create test user without required role
|
|
user = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=["user"],
|
|
groups=[]
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await mock_protected_endpoint(user=user)
|
|
assert exc.value.status_code == 403
|
|
assert exc.value.detail == "You do not have the required roles to access this endpoint."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_require_roles_no_roles():
|
|
# Create test user with no roles
|
|
user = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=[],
|
|
groups=[]
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await mock_protected_endpoint(user=user)
|
|
assert exc.value.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_require_roles_multiple_roles():
|
|
# Test requiring multiple roles
|
|
@require_roles("admin", "super_user")
|
|
async def mock_multi_role_endpoint(user: CognitoUser = Depends(get_current_user)):
|
|
return {"message": "Success"}
|
|
|
|
# User with all required roles
|
|
user_with_roles = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=["admin", "super_user", "user"],
|
|
groups=[]
|
|
)
|
|
result = await mock_multi_role_endpoint(user=user_with_roles)
|
|
assert result == {"message": "Success"}
|
|
|
|
# User missing one required role
|
|
user_missing_role = CognitoUser(
|
|
username="testuser",
|
|
email="test@example.com",
|
|
roles=["admin", "user"],
|
|
groups=[]
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await mock_multi_role_endpoint(user=user_missing_role)
|
|
assert exc.value.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oauth2_scheme_configuration():
|
|
# Verify that we have a properly configured OAuth2PasswordBearer instance
|
|
assert isinstance(oauth2_scheme, OAuth2PasswordBearer)
|
|
|
|
# Create a mock request with no Authorization header
|
|
mock_request = Request(scope={
|
|
'type': 'http',
|
|
'headers': [],
|
|
'method': 'GET',
|
|
'scheme': 'http',
|
|
'path': '/',
|
|
'query_string': b'',
|
|
'client': ('127.0.0.1', 8000)
|
|
})
|
|
|
|
# Test that the scheme raises 401 when no token is provided
|
|
with pytest.raises(HTTPException) as exc:
|
|
await oauth2_scheme(mock_request)
|
|
assert exc.value.status_code == 401
|
|
assert exc.value.detail == "Not authenticated"
|
|
|
|
def test_mock_auth_import(monkeypatch):
|
|
# Save original env var value
|
|
original_value = os.environ.get("MOCK_AUTH")
|
|
|
|
try:
|
|
# Set MOCK_AUTH to true
|
|
monkeypatch.setenv("MOCK_AUTH", "true")
|
|
|
|
# Reload the dependencies module to trigger the import condition
|
|
import app.auth.dependencies
|
|
importlib.reload(app.auth.dependencies)
|
|
|
|
# Verify that mock_get_user_from_token was imported
|
|
from app.auth.dependencies import get_user_from_token
|
|
assert get_user_from_token.__module__ == 'app.auth.mock_auth'
|
|
|
|
finally:
|
|
# Restore original env var
|
|
if original_value is None:
|
|
monkeypatch.delenv("MOCK_AUTH", raising=False)
|
|
else:
|
|
monkeypatch.setenv("MOCK_AUTH", original_value)
|
|
|
|
# Reload again to restore original state
|
|
importlib.reload(app.auth.dependencies) |