test(auth): add tests for GET /auth/me endpoint
Add comprehensive tests for NPM forward auth endpoint: - Forward auth header parsing - User lookup methods (get_user_by_email, get_user_by_authentik_id) - Admin gate authorization logic - OpenAPI spec validation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.5
parent
884996fd83
commit
69045a78a7
@@ -812,3 +812,205 @@ class TestPhase4Schemas:
|
|||||||
response = ApiKeysListResponse(items=[key], total=1)
|
response = ApiKeysListResponse(items=[key], total=1)
|
||||||
assert len(response.items) == 1
|
assert len(response.items) == 1
|
||||||
assert response.total == 1
|
assert response.total == 1
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# GET /auth/me Endpoint Tests (NPM Forward Auth)
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
class TestAuthMeEndpoint:
|
||||||
|
"""Test GET /auth/me endpoint with NPM forward auth."""
|
||||||
|
|
||||||
|
def test_auth_me_in_openapi(self, client):
|
||||||
|
"""Auth me endpoint should be in OpenAPI spec."""
|
||||||
|
response = client.get("/openapi.json")
|
||||||
|
spec = response.json()
|
||||||
|
assert "/auth/me" in spec["paths"]
|
||||||
|
assert "get" in spec["paths"]["/auth/me"]
|
||||||
|
|
||||||
|
def test_auth_me_returns_401_without_forward_auth(self, client):
|
||||||
|
"""Should return 401 when accessed without forward auth headers."""
|
||||||
|
response = client.get("/auth/me")
|
||||||
|
# Without NPM forward auth headers, should return 401
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_auth_me_response_schema(self, client):
|
||||||
|
"""Auth me should return AuthSyncResponse schema."""
|
||||||
|
response = client.get("/openapi.json")
|
||||||
|
spec = response.json()
|
||||||
|
|
||||||
|
# Check response schema references AuthSyncResponse
|
||||||
|
me_endpoint = spec["paths"]["/auth/me"]["get"]
|
||||||
|
assert "responses" in me_endpoint
|
||||||
|
assert "200" in me_endpoint["responses"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestForwardAuthParsing:
|
||||||
|
"""Test NPM forward auth header parsing."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_parses_all_headers(self):
|
||||||
|
"""Should parse all X-authentik-* headers."""
|
||||||
|
from src.auth.oidc import get_forward_auth_user
|
||||||
|
|
||||||
|
mock_request = MagicMock()
|
||||||
|
headers = {
|
||||||
|
"x-authentik-username": "jdoe",
|
||||||
|
"x-authentik-email": "john.doe@example.com",
|
||||||
|
"x-authentik-groups": "tatlock-admins, tatlock-media-viewers",
|
||||||
|
"x-authentik-name": "John Doe",
|
||||||
|
"x-authentik-uid": "550e8400-e29b-41d4-a716-446655440000",
|
||||||
|
}
|
||||||
|
mock_request.headers.get.side_effect = lambda h: headers.get(h)
|
||||||
|
|
||||||
|
result = await get_forward_auth_user(mock_request)
|
||||||
|
|
||||||
|
assert result["username"] == "jdoe"
|
||||||
|
assert result["email"] == "john.doe@example.com"
|
||||||
|
assert result["name"] == "John Doe"
|
||||||
|
assert result["uid"] == "550e8400-e29b-41d4-a716-446655440000"
|
||||||
|
assert "tatlock-admins" in result["groups"]
|
||||||
|
assert "tatlock-media-viewers" in result["groups"]
|
||||||
|
assert result["auth_method"] == "forward_auth"
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_returns_none_for_internal_access(self):
|
||||||
|
"""Should return None when no forward auth headers (internal access)."""
|
||||||
|
from src.auth.oidc import get_forward_auth_user
|
||||||
|
|
||||||
|
mock_request = MagicMock()
|
||||||
|
mock_request.headers.get.return_value = None
|
||||||
|
|
||||||
|
result = await get_forward_auth_user(mock_request)
|
||||||
|
|
||||||
|
assert result is None
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_raises_401_missing_email(self):
|
||||||
|
"""Should raise 401 when username present but email missing."""
|
||||||
|
from src.auth.oidc import get_forward_auth_user
|
||||||
|
from fastapi import HTTPException
|
||||||
|
|
||||||
|
mock_request = MagicMock()
|
||||||
|
headers = {
|
||||||
|
"x-authentik-username": "jdoe",
|
||||||
|
"x-authentik-email": None,
|
||||||
|
}
|
||||||
|
mock_request.headers.get.side_effect = lambda h: headers.get(h)
|
||||||
|
|
||||||
|
with pytest.raises(HTTPException) as exc:
|
||||||
|
await get_forward_auth_user(mock_request)
|
||||||
|
|
||||||
|
assert exc.value.status_code == 401
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_handles_empty_groups(self):
|
||||||
|
"""Should handle empty groups header."""
|
||||||
|
from src.auth.oidc import get_forward_auth_user
|
||||||
|
|
||||||
|
mock_request = MagicMock()
|
||||||
|
headers = {
|
||||||
|
"x-authentik-username": "jdoe",
|
||||||
|
"x-authentik-email": "jdoe@example.com",
|
||||||
|
"x-authentik-groups": "",
|
||||||
|
"x-authentik-name": None,
|
||||||
|
"x-authentik-uid": None,
|
||||||
|
}
|
||||||
|
mock_request.headers.get.side_effect = lambda h: headers.get(h)
|
||||||
|
|
||||||
|
result = await get_forward_auth_user(mock_request)
|
||||||
|
|
||||||
|
assert result["groups"] == [""]
|
||||||
|
assert result["name"] == "jdoe" # Falls back to username
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_strips_whitespace_from_groups(self):
|
||||||
|
"""Should strip whitespace from group names."""
|
||||||
|
from src.auth.oidc import get_forward_auth_user
|
||||||
|
|
||||||
|
mock_request = MagicMock()
|
||||||
|
headers = {
|
||||||
|
"x-authentik-username": "jdoe",
|
||||||
|
"x-authentik-email": "jdoe@example.com",
|
||||||
|
"x-authentik-groups": " group1 , group2 ,group3",
|
||||||
|
"x-authentik-name": "John",
|
||||||
|
"x-authentik-uid": None,
|
||||||
|
}
|
||||||
|
mock_request.headers.get.side_effect = lambda h: headers.get(h)
|
||||||
|
|
||||||
|
result = await get_forward_auth_user(mock_request)
|
||||||
|
|
||||||
|
assert result["groups"] == ["group1", "group2", "group3"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestAuthServiceNewMethods:
|
||||||
|
"""Test new AuthService methods for /auth/me."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_user_by_email(self):
|
||||||
|
"""Should find user by email."""
|
||||||
|
from src.auth.service import AuthService
|
||||||
|
|
||||||
|
mock_session = AsyncMock()
|
||||||
|
mock_user = MagicMock()
|
||||||
|
mock_user.email = "test@example.com"
|
||||||
|
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.scalar_one_or_none.return_value = mock_user
|
||||||
|
mock_session.execute.return_value = mock_result
|
||||||
|
|
||||||
|
service = AuthService(mock_session)
|
||||||
|
result = await service.get_user_by_email("test@example.com")
|
||||||
|
|
||||||
|
assert result is not None
|
||||||
|
assert result.email == "test@example.com"
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_user_by_email_not_found(self):
|
||||||
|
"""Should return None when user not found."""
|
||||||
|
from src.auth.service import AuthService
|
||||||
|
|
||||||
|
mock_session = AsyncMock()
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.scalar_one_or_none.return_value = None
|
||||||
|
mock_session.execute.return_value = mock_result
|
||||||
|
|
||||||
|
service = AuthService(mock_session)
|
||||||
|
result = await service.get_user_by_email("notfound@example.com")
|
||||||
|
|
||||||
|
assert result is None
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_user_by_authentik_id(self):
|
||||||
|
"""Should find user by Authentik UUID."""
|
||||||
|
from src.auth.service import AuthService
|
||||||
|
|
||||||
|
mock_session = AsyncMock()
|
||||||
|
test_id = uuid.uuid4()
|
||||||
|
mock_user = MagicMock()
|
||||||
|
mock_user.authentik_id = test_id
|
||||||
|
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.scalar_one_or_none.return_value = mock_user
|
||||||
|
mock_session.execute.return_value = mock_result
|
||||||
|
|
||||||
|
service = AuthService(mock_session)
|
||||||
|
result = await service.get_user_by_authentik_id(test_id)
|
||||||
|
|
||||||
|
assert result is not None
|
||||||
|
assert result.authentik_id == test_id
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_user_by_authentik_id_not_found(self):
|
||||||
|
"""Should return None when user not found by Authentik ID."""
|
||||||
|
from src.auth.service import AuthService
|
||||||
|
|
||||||
|
mock_session = AsyncMock()
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.scalar_one_or_none.return_value = None
|
||||||
|
mock_session.execute.return_value = mock_result
|
||||||
|
|
||||||
|
service = AuthService(mock_session)
|
||||||
|
result = await service.get_user_by_authentik_id(uuid.uuid4())
|
||||||
|
|
||||||
|
assert result is None
|
||||||
|
|||||||
Reference in New Issue
Block a user