diff --git a/tests/test_auth_controller.py b/tests/test_auth_controller.py index fe6311a..a046318 100644 --- a/tests/test_auth_controller.py +++ b/tests/test_auth_controller.py @@ -812,3 +812,205 @@ class TestPhase4Schemas: response = ApiKeysListResponse(items=[key], total=1) assert len(response.items) == 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