From f786780a20c364c51fd38b0a2e34fdb96b2367e5 Mon Sep 17 00:00:00 2001 From: Daulet Amirkhanov Date: Wed, 20 Aug 2025 19:45:04 +0100 Subject: [PATCH] tests: add unit tests for endpoints and conditional auth --- .../get_conditional_authenticated_user.py | 11 +- cognee/tests/unit/api/__init__.py | 1 + ...st_conditional_authentication_endpoints.py | 266 +++++++++++++++++ cognee/tests/unit/modules/users/__init__.py | 1 + .../users/test_conditional_authentication.py | 280 ++++++++++++++++++ 5 files changed, 557 insertions(+), 2 deletions(-) create mode 100644 cognee/tests/unit/api/__init__.py create mode 100644 cognee/tests/unit/api/test_conditional_authentication_endpoints.py create mode 100644 cognee/tests/unit/modules/users/__init__.py create mode 100644 cognee/tests/unit/modules/users/test_conditional_authentication.py diff --git a/cognee/modules/users/methods/get_conditional_authenticated_user.py b/cognee/modules/users/methods/get_conditional_authenticated_user.py index 644d1aa54..d909d61bf 100644 --- a/cognee/modules/users/methods/get_conditional_authenticated_user.py +++ b/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 diff --git a/cognee/tests/unit/api/__init__.py b/cognee/tests/unit/api/__init__.py new file mode 100644 index 000000000..2b1755712 --- /dev/null +++ b/cognee/tests/unit/api/__init__.py @@ -0,0 +1 @@ +# Test package for API tests diff --git a/cognee/tests/unit/api/test_conditional_authentication_endpoints.py b/cognee/tests/unit/api/test_conditional_authentication_endpoints.py new file mode 100644 index 000000000..fb6aa6887 --- /dev/null +++ b/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() + ) diff --git a/cognee/tests/unit/modules/users/__init__.py b/cognee/tests/unit/modules/users/__init__.py new file mode 100644 index 000000000..a5e9995d3 --- /dev/null +++ b/cognee/tests/unit/modules/users/__init__.py @@ -0,0 +1 @@ +# Test package for user module tests diff --git a/cognee/tests/unit/modules/users/test_conditional_authentication.py b/cognee/tests/unit/modules/users/test_conditional_authentication.py new file mode 100644 index 000000000..da746b5fe --- /dev/null +++ b/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