diff --git a/cognee/api/v1/cognify/routers/get_cognify_router.py b/cognee/api/v1/cognify/routers/get_cognify_router.py index 6adcab8e6..55caa5e5e 100644 --- a/cognee/api/v1/cognify/routers/get_cognify_router.py +++ b/cognee/api/v1/cognify/routers/get_cognify_router.py @@ -46,7 +46,9 @@ def get_cognify_router() -> APIRouter: router = APIRouter() @router.post("", response_model=dict) - async def cognify(payload: CognifyPayloadDTO, user: User = Depends(get_conditional_authenticated_user)): + async def cognify( + payload: CognifyPayloadDTO, user: User = Depends(get_conditional_authenticated_user) + ): """ Transform datasets into structured knowledge graphs through cognitive processing. diff --git a/cognee/api/v1/datasets/routers/get_datasets_router.py b/cognee/api/v1/datasets/routers/get_datasets_router.py index 985aac28d..19b4e5191 100644 --- a/cognee/api/v1/datasets/routers/get_datasets_router.py +++ b/cognee/api/v1/datasets/routers/get_datasets_router.py @@ -114,7 +114,8 @@ def get_datasets_router() -> APIRouter: @router.post("", response_model=DatasetDTO) async def create_new_dataset( - dataset_data: DatasetCreationPayload, user: User = Depends(get_conditional_authenticated_user) + dataset_data: DatasetCreationPayload, + user: User = Depends(get_conditional_authenticated_user), ): """ Create a new dataset or return existing dataset with the same name. @@ -175,7 +176,9 @@ def get_datasets_router() -> APIRouter: @router.delete( "/{dataset_id}", response_model=None, responses={404: {"model": ErrorResponseDTO}} ) - async def delete_dataset(dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user)): + async def delete_dataset( + dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user) + ): """ Delete a dataset by its ID. @@ -263,7 +266,9 @@ def get_datasets_router() -> APIRouter: await delete_data(data) @router.get("/{dataset_id}/graph", response_model=GraphDTO) - async def get_dataset_graph(dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user)): + async def get_dataset_graph( + dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user) + ): """ Get the knowledge graph visualization for a dataset. @@ -293,7 +298,9 @@ def get_datasets_router() -> APIRouter: response_model=list[DataDTO], responses={404: {"model": ErrorResponseDTO}}, ) - async def get_dataset_data(dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user)): + async def get_dataset_data( + dataset_id: UUID, user: User = Depends(get_conditional_authenticated_user) + ): """ Get all data items in a dataset. diff --git a/cognee/api/v1/permissions/routers/get_permissions_router.py b/cognee/api/v1/permissions/routers/get_permissions_router.py index 7f34334e5..9b64a05c7 100644 --- a/cognee/api/v1/permissions/routers/get_permissions_router.py +++ b/cognee/api/v1/permissions/routers/get_permissions_router.py @@ -183,7 +183,9 @@ def get_permissions_router() -> APIRouter: return JSONResponse(status_code=200, content={"message": "User added to tenant"}) @permissions_router.post("/tenants") - async def create_tenant(tenant_name: str, user: User = Depends(get_conditional_authenticated_user)): + async def create_tenant( + tenant_name: str, user: User = Depends(get_conditional_authenticated_user) + ): """ Create a new tenant. diff --git a/cognee/api/v1/search/routers/get_search_router.py b/cognee/api/v1/search/routers/get_search_router.py index 8a238286b..559e8d618 100644 --- a/cognee/api/v1/search/routers/get_search_router.py +++ b/cognee/api/v1/search/routers/get_search_router.py @@ -66,7 +66,9 @@ def get_search_router() -> APIRouter: return JSONResponse(status_code=500, content={"error": str(error)}) @router.post("", response_model=list) - async def search(payload: SearchPayloadDTO, user: User = Depends(get_conditional_authenticated_user)): + async def search( + payload: SearchPayloadDTO, user: User = Depends(get_conditional_authenticated_user) + ): """ Search for nodes in the graph database. diff --git a/cognee/modules/users/methods/__init__.py b/cognee/modules/users/methods/__init__.py index aee91b823..4539dbdb0 100644 --- a/cognee/modules/users/methods/__init__.py +++ b/cognee/modules/users/methods/__init__.py @@ -4,4 +4,7 @@ from .delete_user import delete_user from .get_default_user import get_default_user from .get_user_by_email import get_user_by_email from .create_default_user import create_default_user -from .get_conditional_authenticated_user import get_conditional_authenticated_user, REQUIRE_AUTHENTICATION +from .get_conditional_authenticated_user import ( + get_conditional_authenticated_user, + REQUIRE_AUTHENTICATION, +) diff --git a/cognee/modules/users/methods/get_conditional_authenticated_user.py b/cognee/modules/users/methods/get_conditional_authenticated_user.py index d909d61bf..e3ea7555f 100644 --- a/cognee/modules/users/methods/get_conditional_authenticated_user.py +++ b/cognee/modules/users/methods/get_conditional_authenticated_user.py @@ -17,15 +17,18 @@ else: # When REQUIRE_AUTHENTICATION=false (default), make authentication optional _auth_dependency = fastapi_users.current_user( optional=True, # Returns None instead of raising HTTPException(401) - active=True # Still require users to be active when authenticated + active=True, # Still require users to be active when authenticated ) -async def get_conditional_authenticated_user(user: Optional[User] = Depends(_auth_dependency)) -> User: + +async def get_conditional_authenticated_user( + user: Optional[User] = Depends(_auth_dependency), +) -> User: """ Get authenticated user with environment-controlled behavior: - If REQUIRE_AUTHENTICATION=true: Enforces authentication (raises 401 if not authenticated) - If REQUIRE_AUTHENTICATION=false: Falls back to default user if not authenticated - + Always returns a User object for consistent typing. """ if user is None and not REQUIRE_AUTHENTICATION: @@ -34,9 +37,6 @@ async def get_conditional_authenticated_user(user: Optional[User] = Depends(_aut 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)}" - ) - + raise HTTPException(status_code=500, detail=f"Failed to create default user: {str(e)}") + return user diff --git a/cognee/tests/unit/api/test_conditional_authentication_endpoints.py b/cognee/tests/unit/api/test_conditional_authentication_endpoints.py index fb6aa6887..9199b47a7 100644 --- a/cognee/tests/unit/api/test_conditional_authentication_endpoints.py +++ b/cognee/tests/unit/api/test_conditional_authentication_endpoints.py @@ -11,153 +11,167 @@ 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() + 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() + 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: + 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) + + # 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()) + + 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: + 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 - + 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"), - ]) + + @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: + 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 @@ -167,76 +181,84 @@ class TestConditionalAuthenticationBehavior: 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: + 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_model="gpt-4o", llm_endpoint=None, llm_api_version=None, - llm_api_key="test_key_1234567890" + 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" + 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: + 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 - + 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 @@ -246,21 +268,20 @@ class TestConditionalAuthenticationErrorHandling: def mock_default_user(): """Mock default user for testing.""" return SimpleNamespace( - id=uuid4(), - email="default@example.com", - is_active=True, - tenant_id=uuid4() + id=uuid4(), email="default@example.com", is_active=True, tenant_id=uuid4() ) -@pytest.fixture + +@pytest.fixture def mock_authenticated_user(): """Mock authenticated user for testing.""" from cognee.modules.users.models import User + return User( - id=uuid4(), + id=uuid4(), email="auth@example.com", hashed_password="hashed", is_active=True, is_verified=True, - tenant_id=uuid4() + tenant_id=uuid4(), ) diff --git a/cognee/tests/unit/modules/users/test_conditional_authentication.py b/cognee/tests/unit/modules/users/test_conditional_authentication.py index da746b5fe..d9befa328 100644 --- a/cognee/tests/unit/modules/users/test_conditional_authentication.py +++ b/cognee/tests/unit/modules/users/test_conditional_authentication.py @@ -9,27 +9,29 @@ 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 - ) - + 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: + 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() @@ -41,15 +43,20 @@ class TestConditionalAuthentication: email="user@example.com", hashed_password="hashed", is_active=True, - is_verified=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: + 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() @@ -58,16 +65,19 @@ class TestConditionalAuthentication: """Test that when REQUIRE_AUTHENTICATION=true and user present, returns user.""" mock_authenticated_user = User( id=uuid4(), - email="user@example.com", + email="user@example.com", hashed_password="hashed", is_active=True, - is_verified=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 + 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 @@ -75,31 +85,34 @@ class TestConditionalAuthentication: """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), + + # 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 + 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') + assert hasattr(result, "id") class TestConditionalAuthenticationIntegration: """Integration tests that test the full authentication flow.""" - - @pytest.mark.asyncio + + @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) @@ -109,78 +122,92 @@ class TestConditionalAuthenticationIntegration: """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 + 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' + 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 + 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' + 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 + 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' + 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 + 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' + 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 + 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 - + 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 @@ -188,15 +215,20 @@ class TestConditionalAuthenticationEnvironmentVariables: 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 + 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: + 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) @@ -204,66 +236,72 @@ class TestConditionalAuthenticationEdgeCases: @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 + 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", + hashed_password="hashed", is_active=True, - is_verified=True + is_verified=True, ) - - mock_default_user = SimpleNamespace( - id=uuid4(), - email="default@example.com", - is_active=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: + 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) + 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') + 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 + - 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 + 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: + 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( @@ -271,10 +309,13 @@ class TestAuthenticationScenarios: email="active@example.com", hashed_password="hashed", is_active=True, - is_verified=True + is_verified=True, ) - - from cognee.modules.users.methods.get_conditional_authenticated_user import get_conditional_authenticated_user + + 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