diff --git a/cognee/tests/unit/modules/users/test_conditional_authentication.py b/cognee/tests/unit/modules/users/test_conditional_authentication.py index e1ac1d9e8..51bd1eda4 100644 --- a/cognee/tests/unit/modules/users/test_conditional_authentication.py +++ b/cognee/tests/unit/modules/users/test_conditional_authentication.py @@ -1,10 +1,8 @@ import os import sys import pytest -import pytest_asyncio -from unittest.mock import AsyncMock, MagicMock, patch -from uuid import uuid4, UUID -from fastapi import HTTPException +from unittest.mock import AsyncMock, patch +from uuid import uuid4 from types import SimpleNamespace from cognee.modules.users.models import User @@ -14,29 +12,34 @@ class TestConditionalAuthentication: """Test cases for conditional authentication functionality.""" @pytest.mark.asyncio - async def test_require_authentication_false_no_token_returns_default_user(self): + @patch("cognee.modules.users.methods.get_authenticated_user.get_default_user", new_callable=AsyncMock) + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + False, + ) + async def test_require_authentication_false_no_token_returns_default_user(self, mock_get_default): """Test that when REQUIRE_AUTHENTICATION=false and no token, returns default user.""" # Mock the default user mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com", is_active=True) + mock_get_default.return_value = mock_default_user - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - from cognee.modules.users.methods.get_authenticated_user import ( - get_authenticated_user, - ) + from cognee.modules.users.methods.get_authenticated_user import ( + get_authenticated_user, + ) - with patch( - "cognee.modules.users.methods.get_authenticated_user.get_default_user" - ) as mock_get_default: - mock_get_default.return_value = mock_default_user + # Test with None user (no authentication) + result = await get_authenticated_user(user=None) - # Test with None user (no authentication) - result = await get_authenticated_user(user=None) - - assert result == mock_default_user - mock_get_default.assert_called_once() + assert result == mock_default_user + mock_get_default.assert_called_once() @pytest.mark.asyncio - async def test_require_authentication_false_with_valid_user_returns_user(self): + @patch("cognee.modules.users.methods.get_authenticated_user.get_default_user", new_callable=AsyncMock) + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + False, + ) + async def test_require_authentication_false_with_valid_user_returns_user(self, mock_get_default): """Test that when REQUIRE_AUTHENTICATION=false and valid user, returns that user.""" mock_authenticated_user = User( id=uuid4(), @@ -46,21 +49,21 @@ class TestConditionalAuthentication: is_verified=True, ) - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - from cognee.modules.users.methods.get_authenticated_user import ( - get_authenticated_user, - ) + from cognee.modules.users.methods.get_authenticated_user import ( + get_authenticated_user, + ) - with patch( - "cognee.modules.users.methods.get_authenticated_user.get_default_user" - ) as mock_get_default: - # Test with authenticated user - result = await get_authenticated_user(user=mock_authenticated_user) + # Test with authenticated user + result = await get_authenticated_user(user=mock_authenticated_user) - assert result == mock_authenticated_user - mock_get_default.assert_not_called() + assert result == mock_authenticated_user + mock_get_default.assert_not_called() @pytest.mark.asyncio + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + True, + ) async def test_require_authentication_true_with_user_returns_user(self): """Test that when REQUIRE_AUTHENTICATION=true and user present, returns user.""" mock_authenticated_user = User( @@ -71,33 +74,13 @@ class TestConditionalAuthentication: is_verified=True, ) - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "true"}): - from cognee.modules.users.methods.get_authenticated_user import ( - get_authenticated_user, - ) - - result = await get_authenticated_user(user=mock_authenticated_user) - - assert result == mock_authenticated_user - - @pytest.mark.asyncio - async def test_require_authentication_true_with_none_returns_none(self): - """Test that when REQUIRE_AUTHENTICATION=true and no user, returns None (would raise 401 at dependency level).""" - # This test simulates what would happen if REQUIRE_AUTHENTICATION was true at import time - # In reality, when REQUIRE_AUTHENTICATION=true, FastAPI Users would raise 401 BEFORE this function is called - - # Since REQUIRE_AUTHENTICATION is currently false (set at import time), - # we expect it to return the default user, not None from cognee.modules.users.methods.get_authenticated_user import ( get_authenticated_user, ) - result = await get_authenticated_user(user=None) - - # The current implementation will return default user because REQUIRE_AUTHENTICATION is false - assert result is not None # Should get default user - assert hasattr(result, "id") + result = await get_authenticated_user(user=mock_authenticated_user) + assert result == mock_authenticated_user class TestConditionalAuthenticationIntegration: """Integration tests that test the full authentication flow.""" @@ -139,7 +122,7 @@ class TestConditionalAuthenticationEnvironmentVariables: """Test environment variable handling.""" def test_require_authentication_default_false(self): - """Test that REQUIRE_AUTHENTICATION defaults to false when imported with no env var.""" + """Test that REQUIRE_AUTHENTICATION defaults to false when imported with no env vars.""" with patch.dict(os.environ, {}, clear=True): # Remove module from cache to force fresh import module_name = "cognee.modules.users.methods.get_authenticated_user" @@ -217,24 +200,27 @@ class TestConditionalAuthenticationEdgeCases: """Test edge cases and error scenarios.""" @pytest.mark.asyncio - async def test_get_default_user_raises_exception(self): + @patch("cognee.modules.users.methods.get_authenticated_user.get_default_user", new_callable=AsyncMock) + @patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}) + async def test_get_default_user_raises_exception(self, mock_get_default): """Test behavior when get_default_user raises an exception.""" from cognee.modules.users.methods.get_authenticated_user import ( get_authenticated_user, ) - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - with patch( - "cognee.modules.users.methods.get_authenticated_user.get_default_user" - ) as mock_get_default: - mock_get_default.side_effect = Exception("Database error") + mock_get_default.side_effect = Exception("Database error") - # This should propagate the exception - with pytest.raises(Exception, match="Database error"): - await get_authenticated_user(user=None) + # This should propagate the exception + with pytest.raises(Exception, match="Database error"): + await get_authenticated_user(user=None) @pytest.mark.asyncio - async def test_user_type_consistency(self): + @patch("cognee.modules.users.methods.get_authenticated_user.get_default_user", new_callable=AsyncMock) + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + False, + ) + async def test_user_type_consistency(self, mock_get_default): """Test that the function always returns the same type.""" from cognee.modules.users.methods.get_authenticated_user import ( get_authenticated_user, @@ -249,33 +235,33 @@ class TestConditionalAuthenticationEdgeCases: ) mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com", is_active=True) + mock_get_default.return_value = mock_default_user - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - with patch( - "cognee.modules.users.methods.get_authenticated_user.get_default_user" - ) as mock_get_default: - mock_get_default.return_value = mock_default_user + # Test with user + result1 = await get_authenticated_user(user=mock_user) + assert result1 == mock_user - # Test with user - result1 = await get_authenticated_user(user=mock_user) - assert result1 == mock_user + # Test with None + result2 = await get_authenticated_user(user=None) + assert result2 == mock_default_user - # Test with None - result2 = await get_authenticated_user(user=None) - assert result2 == mock_default_user - - # Both should have user-like interface - assert hasattr(result1, "id") - assert hasattr(result1, "email") - assert hasattr(result2, "id") - assert hasattr(result2, "email") + # Both should have user-like interface + assert hasattr(result1, "id") + assert hasattr(result1, "email") + assert hasattr(result2, "id") + assert hasattr(result2, "email") @pytest.mark.asyncio class TestAuthenticationScenarios: """Test specific authentication scenarios that could occur in FastAPI Users.""" - async def test_fallback_to_default_user_scenarios(self): + @patch("cognee.modules.users.methods.get_authenticated_user.get_default_user", new_callable=AsyncMock) + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + False, + ) + async def test_fallback_to_default_user_scenarios(self, mock_get_default): """ Test fallback to default user for all scenarios where FastAPI Users returns None: - No JWT/Cookie present @@ -287,21 +273,21 @@ class TestAuthenticationScenarios: which should trigger fallback to default user. """ mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com") + mock_get_default.return_value = mock_default_user + from cognee.modules.users.methods.get_authenticated_user import ( get_authenticated_user, ) - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - with patch( - "cognee.modules.users.methods.get_authenticated_user.get_default_user" - ) as mock_get_default: - mock_get_default.return_value = mock_default_user - - # All the above scenarios result in user=None being passed to our function - result = await get_authenticated_user(user=None) - assert result == mock_default_user - mock_get_default.assert_called_once() + # All the above scenarios result in user=None being passed to our function + result = await get_authenticated_user(user=None) + assert result == mock_default_user + mock_get_default.assert_called_once() + @patch( + "cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", + False, + ) async def test_scenario_valid_active_user(self): """Scenario: Valid JWT and user exists and is active → returns the user.""" mock_user = User( @@ -316,6 +302,5 @@ class TestAuthenticationScenarios: get_authenticated_user, ) - with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}): - result = await get_authenticated_user(user=mock_user) - assert result == mock_user + result = await get_authenticated_user(user=mock_user) + assert result == mock_user