tests: add unit tests for endpoints and conditional auth

cognee/modules/users/methods/get_conditional_authenticated_user.py

@@ -1,6 +1,6 @@
 import os
 from typing import Optional
-from fastapi import Depends
+from fastapi import Depends, HTTPException
 from ..models import User
 from ..get_fastapi_users import get_fastapi_users
 from .get_default_user import get_default_user
@@ -30,6 +30,13 @@ async def get_conditional_authenticated_user(user: Optional[User] = Depends(_aut
     """
     if user is None and not REQUIRE_AUTHENTICATION:
         # When authentication is optional and user is None, use default user
-        user = await get_default_user()
+        try:
+            user = await get_default_user()
+        except Exception as e:
+            # Convert any get_default_user failure into a proper HTTP 500 error
+            raise HTTPException(
+                status_code=500,
+                detail=f"Failed to create default user: {str(e)}"
+            )
     
     return user

cognee/tests/unit/api/__init__.py

@@ -0,0 +1 @@
+# Test package for API tests

cognee/tests/unit/api/test_conditional_authentication_endpoints.py

@@ -0,0 +1,266 @@
+import os
+import pytest
+import pytest_asyncio
+from unittest.mock import patch, AsyncMock, MagicMock
+from uuid import uuid4
+from fastapi.testclient import TestClient
+from types import SimpleNamespace
+
+from cognee.api.client import app
+
+
+class TestConditionalAuthenticationEndpoints:
+    """Test that API endpoints work correctly with conditional authentication."""
+    
+    @pytest.fixture
+    def client(self):
+        """Create a test client."""
+        return TestClient(app)
+    
+    @pytest.fixture
+    def mock_default_user(self):
+        """Mock default user for testing."""
+        return SimpleNamespace(
+            id=uuid4(),
+            email="default@example.com", 
+            is_active=True,
+            tenant_id=uuid4()
+        )
+    
+    @pytest.fixture
+    def mock_authenticated_user(self):
+        """Mock authenticated user for testing."""
+        from cognee.modules.users.models import User
+        return User(
+            id=uuid4(),
+            email="auth@example.com",
+            hashed_password="hashed",
+            is_active=True,
+            is_verified=True,
+            tenant_id=uuid4()
+        )
+
+    def test_health_endpoint_no_auth_required(self, client):
+        """Test that health endpoint works without authentication."""
+        response = client.get("/health")
+        assert response.status_code in [200, 503]  # 503 is also acceptable for health checks
+    
+    def test_root_endpoint_no_auth_required(self, client):
+        """Test that root endpoint works without authentication."""
+        response = client.get("/")
+        assert response.status_code == 200
+        assert response.json() == {"message": "Hello, World, I am alive!"}
+    
+    @patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"})
+    def test_openapi_schema_no_global_security(self, client):
+        """Test that OpenAPI schema doesn't require global authentication."""
+        response = client.get("/openapi.json")
+        assert response.status_code == 200
+        
+        schema = response.json()
+        
+        # Should not have global security requirement
+        global_security = schema.get("security", [])
+        assert global_security == []
+        
+        # But should still have security schemes defined
+        security_schemes = schema.get("components", {}).get("securitySchemes", {})
+        assert "BearerAuth" in security_schemes
+        assert "CookieAuth" in security_schemes
+    
+    @patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"})
+    def test_add_endpoint_with_conditional_auth(self, client, mock_default_user):
+        """Test add endpoint works with conditional authentication."""
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            with patch('cognee.api.v1.add.add') as mock_cognee_add:
+                mock_get_default.return_value = mock_default_user
+                mock_cognee_add.return_value = MagicMock(
+                    model_dump=lambda: {"status": "success", "pipeline_run_id": str(uuid4())}
+                )
+                
+                # Test file upload without authentication
+                files = {"data": ("test.txt", b"test content", "text/plain")}
+                form_data = {"datasetName": "test_dataset"}
+                
+                response = client.post("/api/v1/add", files=files, data=form_data)
+                
+                # Should succeed (not 401) 
+                assert response.status_code != 401
+                
+                # Should have called get_default_user for anonymous request
+                mock_get_default.assert_called()
+    
+    def test_conditional_authentication_works_with_current_environment(self, client):
+        """Test that conditional authentication works with the current environment setup."""
+        # Since REQUIRE_AUTHENTICATION defaults to "false", we expect endpoints to work without auth
+        # This tests the actual integration behavior
+        
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com", is_active=True, tenant_id=uuid4())
+            mock_get_default.return_value = mock_default_user
+            
+            files = {"data": ("test.txt", b"test content", "text/plain")}
+            form_data = {"datasetName": "test_dataset"}
+            
+            response = client.post("/api/v1/add", files=files, data=form_data)
+            
+            # Should not return 401 (authentication not required with default environment)
+            assert response.status_code != 401
+            
+            # Should have called get_default_user for anonymous request
+            mock_get_default.assert_called()
+    
+    def test_authenticated_request_uses_user(self, client, mock_authenticated_user):
+        """Test that authenticated requests use the authenticated user, not default user."""
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            with patch('cognee.api.v1.add.add') as mock_cognee_add:
+                # Mock successful authentication - this would normally be handled by FastAPI Users
+                # but we're testing the conditional logic
+                mock_cognee_add.return_value = MagicMock(
+                    model_dump=lambda: {"status": "success", "pipeline_run_id": str(uuid4())}
+                )
+                
+                # Simulate authenticated request by directly testing the conditional function
+                from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+                
+                async def test_logic():
+                    # When user is provided (authenticated), should not call get_default_user
+                    result = await get_conditional_authenticated_user(user=mock_authenticated_user)
+                    assert result == mock_authenticated_user
+                    mock_get_default.assert_not_called()
+                
+                # Run the async test
+                import asyncio
+                asyncio.run(test_logic())
+
+
+class TestConditionalAuthenticationBehavior:
+    """Test the behavior of conditional authentication across different endpoints."""
+    
+    @pytest.fixture
+    def client(self):
+        return TestClient(app)
+    
+    @pytest.mark.parametrize("endpoint,method", [
+        ("/api/v1/search", "GET"),
+        ("/api/v1/datasets", "GET"),
+    ])
+    def test_get_endpoints_work_without_auth(self, client, endpoint, method, mock_default_user):
+        """Test that GET endpoints work without authentication (with current environment)."""
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            mock_get_default.return_value = mock_default_user
+            
+            if method == "GET":
+                response = client.get(endpoint)
+            elif method == "POST":
+                response = client.post(endpoint, json={})
+            
+            # Should not return 401 Unauthorized (authentication is optional by default)
+            assert response.status_code != 401
+            
+            # May return other errors due to missing data/config, but not auth errors
+            if response.status_code >= 400:
+                # Check that it's not an authentication error
+                try:
+                    error_detail = response.json().get("detail", "")
+                    assert "authenticate" not in error_detail.lower()
+                    assert "unauthorized" not in error_detail.lower()
+                except:
+                    pass  # If response is not JSON, that's fine
+    
+    def test_settings_endpoint_integration(self, client, mock_default_user):
+        """Test that settings endpoint integration works with conditional authentication."""
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            with patch('cognee.modules.settings.get_settings.get_llm_config') as mock_llm_config:
+                with patch('cognee.modules.settings.get_settings.get_vectordb_config') as mock_vector_config:
+                    mock_get_default.return_value = mock_default_user
+                    
+                    # Mock configurations to avoid validation errors
+                    mock_llm_config.return_value = SimpleNamespace(
+                        llm_provider="openai",
+                        llm_model="gpt-4o", 
+                        llm_endpoint=None,
+                        llm_api_version=None,
+                        llm_api_key="test_key_1234567890"
+                    )
+                    
+                    mock_vector_config.return_value = SimpleNamespace(
+                        vector_db_provider="lancedb",
+                        vector_db_url="localhost:5432",  # Must be string, not None
+                        vector_db_key="test_vector_key"
+                    )
+                    
+                    response = client.get("/api/v1/settings")
+                    
+                    # Should not return 401 (authentication works)
+                    assert response.status_code != 401
+                    
+                    # Should have called get_default_user for anonymous request
+                    mock_get_default.assert_called()
+
+
+class TestConditionalAuthenticationErrorHandling:
+    """Test error handling in conditional authentication."""
+    
+    @pytest.fixture
+    def client(self):
+        return TestClient(app)
+    
+    def test_get_default_user_fails(self, client):
+        """Test behavior when get_default_user fails (with current environment)."""
+        with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+            mock_get_default.side_effect = Exception("Database connection failed")
+            
+            # The error should propagate - either as a 500 error or as an exception
+            files = {"data": ("test.txt", b"test content", "text/plain")}
+            form_data = {"datasetName": "test_dataset"}
+            
+            # Test that the exception is properly converted to HTTP 500
+            response = client.post("/api/v1/add", files=files, data=form_data)
+            
+            # Should return HTTP 500 Internal Server Error when get_default_user fails
+            assert response.status_code == 500
+            
+            # Check that the error message is informative
+            error_detail = response.json().get("detail", "")
+            assert "Failed to create default user" in error_detail
+            assert "Database connection failed" in error_detail
+            
+            # Most importantly, verify that get_default_user was called (the conditional auth is working)
+            mock_get_default.assert_called()
+    
+    def test_current_environment_configuration(self):
+        """Test that current environment configuration is working properly."""
+        # This tests the actual module state without trying to change it
+        from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+        
+        # Should be a boolean value (the parsing logic works)
+        assert isinstance(REQUIRE_AUTHENTICATION, bool)
+        
+        # In default environment, should be False
+        assert REQUIRE_AUTHENTICATION == False
+
+
+# Fixtures for reuse across test classes
+@pytest.fixture
+def mock_default_user():
+    """Mock default user for testing."""
+    return SimpleNamespace(
+        id=uuid4(),
+        email="default@example.com",
+        is_active=True,
+        tenant_id=uuid4()
+    )
+
+@pytest.fixture  
+def mock_authenticated_user():
+    """Mock authenticated user for testing."""
+    from cognee.modules.users.models import User
+    return User(
+        id=uuid4(), 
+        email="auth@example.com",
+        hashed_password="hashed",
+        is_active=True,
+        is_verified=True,
+        tenant_id=uuid4()
+    )

cognee/tests/unit/modules/users/__init__.py

@@ -0,0 +1 @@
+# Test package for user module tests

cognee/tests/unit/modules/users/test_conditional_authentication.py

@@ -0,0 +1,280 @@
+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 types import SimpleNamespace
+
+from cognee.modules.users.models import User
+
+class TestConditionalAuthentication:
+    """Test cases for conditional authentication functionality."""
+    
+    @pytest.mark.asyncio
+    async def test_require_authentication_false_no_token_returns_default_user(self):
+        """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
+        )
+        
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+            with patch('cognee.modules.users.methods.get_conditional_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_conditional_authenticated_user(user=None)
+                
+                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):
+        """Test that when REQUIRE_AUTHENTICATION=false and valid user, returns that user."""
+        mock_authenticated_user = User(
+            id=uuid4(),
+            email="user@example.com",
+            hashed_password="hashed",
+            is_active=True,
+            is_verified=True
+        )
+        
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+            with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+                # Test with authenticated user
+                result = await get_conditional_authenticated_user(user=mock_authenticated_user)
+                
+                assert result == mock_authenticated_user
+                mock_get_default.assert_not_called()
+
+    @pytest.mark.asyncio
+    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(
+            id=uuid4(),
+            email="user@example.com", 
+            hashed_password="hashed",
+            is_active=True,
+            is_verified=True
+        )
+        
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "true"}):
+            from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+            result = await get_conditional_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_conditional_authenticated_user import get_conditional_authenticated_user
+        result = await get_conditional_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')
+
+
+class TestConditionalAuthenticationIntegration:
+    """Integration tests that test the full authentication flow."""
+    
+    @pytest.mark.asyncio 
+    async def test_fastapi_users_dependency_creation(self):
+        """Test that FastAPI Users dependency can be created correctly."""
+        from cognee.modules.users.get_fastapi_users import get_fastapi_users
+        
+        fastapi_users = get_fastapi_users()
+        
+        # Test that we can create optional dependency
+        optional_dependency = fastapi_users.current_user(optional=True, active=True)
+        assert callable(optional_dependency)
+        
+        # Test that we can create required dependency
+        required_dependency = fastapi_users.current_user(active=True)  # optional=False by default
+        assert callable(required_dependency)
+
+    @pytest.mark.asyncio
+    async def test_conditional_authentication_function_exists(self):
+        """Test that the conditional authentication function can be imported and used."""
+        from cognee.modules.users.methods.get_conditional_authenticated_user import (
+            get_conditional_authenticated_user,
+            REQUIRE_AUTHENTICATION
+        )
+        
+        # Should be callable
+        assert callable(get_conditional_authenticated_user)
+        
+        # REQUIRE_AUTHENTICATION should be a boolean
+        assert isinstance(REQUIRE_AUTHENTICATION, bool)
+        
+        # Currently should be False (optional authentication)
+        assert REQUIRE_AUTHENTICATION == False
+
+
+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."""
+        with patch.dict(os.environ, {}, clear=True):
+            # Remove module from cache to force fresh import
+            module_name = 'cognee.modules.users.methods.get_conditional_authenticated_user'
+            if module_name in sys.modules:
+                del sys.modules[module_name]
+            
+            # Import after patching environment - module will see empty environment
+            from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+            assert REQUIRE_AUTHENTICATION == False
+    
+    def test_require_authentication_true(self):
+        """Test that REQUIRE_AUTHENTICATION=true is parsed correctly when imported."""
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "true"}):
+            # Remove module from cache to force fresh import
+            module_name = 'cognee.modules.users.methods.get_conditional_authenticated_user'
+            if module_name in sys.modules:
+                del sys.modules[module_name]
+                
+            # Import after patching environment - module will see REQUIRE_AUTHENTICATION=true
+            from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+            assert REQUIRE_AUTHENTICATION == True
+    
+    def test_require_authentication_false_explicit(self):
+        """Test that REQUIRE_AUTHENTICATION=false is parsed correctly when imported."""
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            # Remove module from cache to force fresh import
+            module_name = 'cognee.modules.users.methods.get_conditional_authenticated_user'
+            if module_name in sys.modules:
+                del sys.modules[module_name]
+                
+            # Import after patching environment - module will see REQUIRE_AUTHENTICATION=false
+            from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+            assert REQUIRE_AUTHENTICATION == False
+    
+    def test_require_authentication_case_insensitive(self):
+        """Test that environment variable parsing is case insensitive when imported."""
+        test_cases = ["TRUE", "True", "tRuE", "FALSE", "False", "fAlSe"]
+        
+        for case in test_cases:
+            with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": case}):
+                # Remove module from cache to force fresh import
+                module_name = 'cognee.modules.users.methods.get_conditional_authenticated_user'
+                if module_name in sys.modules:
+                    del sys.modules[module_name]
+                
+                # Import after patching environment
+                from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+                expected = case.lower() == "true"
+                assert REQUIRE_AUTHENTICATION == expected, f"Failed for case: {case}"
+    
+    def test_current_require_authentication_value(self):
+        """Test that the current REQUIRE_AUTHENTICATION module value is as expected."""
+        from cognee.modules.users.methods.get_conditional_authenticated_user import REQUIRE_AUTHENTICATION
+        
+        # The module-level variable should currently be False (set at import time)
+        assert isinstance(REQUIRE_AUTHENTICATION, bool)
+        assert REQUIRE_AUTHENTICATION == False
+
+
+class TestConditionalAuthenticationEdgeCases:
+    """Test edge cases and error scenarios."""
+    
+    @pytest.mark.asyncio
+    async def test_get_default_user_raises_exception(self):
+        """Test behavior when get_default_user raises an exception."""
+        from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+                mock_get_default.side_effect = Exception("Database error")
+                
+                # This should propagate the exception
+                with pytest.raises(Exception, match="Database error"):
+                    await get_conditional_authenticated_user(user=None)
+
+    @pytest.mark.asyncio
+    async def test_user_type_consistency(self):
+        """Test that the function always returns the same type."""
+        from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+        mock_user = User(
+            id=uuid4(),
+            email="test@example.com",
+            hashed_password="hashed", 
+            is_active=True,
+            is_verified=True
+        )
+        
+        mock_default_user = SimpleNamespace(
+            id=uuid4(),
+            email="default@example.com",
+            is_active=True
+        )
+        
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            with patch('cognee.modules.users.methods.get_conditional_authenticated_user.get_default_user') as mock_get_default:
+                mock_get_default.return_value = mock_default_user
+                
+                # Test with user
+                result1 = await get_conditional_authenticated_user(user=mock_user)
+                assert result1 == mock_user
+                
+                # Test with None
+                result2 = await get_conditional_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')
+
+
+@pytest.mark.asyncio
+class TestAuthenticationScenarios:
+    """Test specific authentication scenarios that could occur in FastAPI Users."""
+    
+    async def test_fallback_to_default_user_scenarios(self):
+        """
+        Test fallback to default user for all scenarios where FastAPI Users returns None:
+        - No JWT/Cookie present
+        - Invalid JWT/Cookie  
+        - Valid JWT but user doesn't exist in database
+        - Valid JWT but user is inactive (active=True requirement)
+        
+        All these scenarios result in FastAPI Users returning None when optional=True,
+        which should trigger fallback to default user.
+        """
+        mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com")
+        from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            with patch('cognee.modules.users.methods.get_conditional_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_conditional_authenticated_user(user=None)
+                assert result == mock_default_user
+                mock_get_default.assert_called_once()
+    
+    async def test_scenario_valid_active_user(self):
+        """Scenario: Valid JWT and user exists and is active → returns the user."""
+        mock_user = User(
+            id=uuid4(),
+            email="active@example.com",
+            hashed_password="hashed",
+            is_active=True,
+            is_verified=True
+        )
+        
+        from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user
+        with patch.dict(os.environ, {"REQUIRE_AUTHENTICATION": "false"}):
+            result = await get_conditional_authenticated_user(user=mock_user)
+            assert result == mock_user