gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: update test imports and patching for conditional authentication tests

Browse source
This commit is contained in:
Daulet Amirkhanov 2025-09-03 16:51:14 +01:00
parent 057c84fdc5
commit 6fe2771421
2 changed files with 55 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
cognee/tests/unit/api/test_conditional_authentication_endpoints.py
View file

@ -3,6 +3,7 @@ from unittest.mock import patch, AsyncMock, MagicMock
from uuid import uuid4 from uuid import uuid4
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from types import SimpleNamespace from types import SimpleNamespace
import importlib
from cognee.api.client import app from cognee.api.client import app
@ -30,6 +31,10 @@ def mock_authenticated_user():
tenant_id=uuid4(), tenant_id=uuid4(),
) )
gau_mod = importlib.import_module(
"cognee.modules.users.methods.get_authenticated_user"
)
class TestConditionalAuthenticationEndpoints: class TestConditionalAuthenticationEndpoints:
"""Test that API endpoints work correctly with conditional authentication.""" """Test that API endpoints work correctly with conditional authentication."""
@ -51,7 +56,7 @@ class TestConditionalAuthenticationEndpoints:
assert response.json() == {"message": "Hello, World, I am alive!"} assert response.json() == {"message": "Hello, World, I am alive!"}
@patch( @patch(
"cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", "cognee.api.client.REQUIRE_AUTHENTICATION",
False, False,
) )
def test_openapi_schema_no_global_security(self, client): def test_openapi_schema_no_global_security(self, client):
@ -71,9 +76,9 @@ class TestConditionalAuthenticationEndpoints:
assert "CookieAuth" in security_schemes assert "CookieAuth" in security_schemes
@patch("cognee.api.v1.add.add") @patch("cognee.api.v1.add.add")
@patch("cognee.modules.users.methods.get_default_user.get_default_user", new_callable=AsyncMock) @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
@patch( @patch(
"cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", "cognee.api.client.REQUIRE_AUTHENTICATION",
False, False,
) )
def test_add_endpoint_with_conditional_auth( def test_add_endpoint_with_conditional_auth(
@ -91,12 +96,14 @@ class TestConditionalAuthenticationEndpoints:
response = client.post("/api/v1/add", files=files, data=form_data) response = client.post("/api/v1/add", files=files, data=form_data)
assert mock_get_default_user.call_count == 1
# Core test: authentication is not required (should not get 401) # Core test: authentication is not required (should not get 401)
assert response.status_code != 401 assert response.status_code != 401
@patch("cognee.modules.users.methods.get_default_user.get_default_user", new_callable=AsyncMock) @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
@patch( @patch(
"cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", "cognee.api.client.REQUIRE_AUTHENTICATION",
False, False,
) )
def test_conditional_authentication_works_with_current_environment( def test_conditional_authentication_works_with_current_environment(
@ -115,6 +122,8 @@ class TestConditionalAuthenticationEndpoints:
response = client.post("/api/v1/add", files=files, data=form_data) response = client.post("/api/v1/add", files=files, data=form_data)
assert mock_get_default_user.call_count == 1
# Core test: authentication is not required (should not get 401) # Core test: authentication is not required (should not get 401)
assert response.status_code != 401 assert response.status_code != 401
# Note: This test verifies conditional authentication works in the current environment # Note: This test verifies conditional authentication works in the current environment
@ -134,7 +143,7 @@ class TestConditionalAuthenticationBehavior:
("/api/v1/datasets", "GET"), ("/api/v1/datasets", "GET"),
], ],
) )
@patch("cognee.modules.users.methods.get_default_user.get_default_user", new_callable=AsyncMock) @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
def test_get_endpoints_work_without_auth( def test_get_endpoints_work_without_auth(
self, mock_get_default, client, endpoint, method, mock_default_user self, mock_get_default, client, endpoint, method, mock_default_user
): ):
@ -146,6 +155,8 @@ class TestConditionalAuthenticationBehavior:
elif method == "POST": elif method == "POST":
response = client.post(endpoint, json={}) response = client.post(endpoint, json={})
assert mock_get_default.call_count == 1
# Should not return 401 Unauthorized (authentication is optional by default) # Should not return 401 Unauthorized (authentication is optional by default)
assert response.status_code != 401 assert response.status_code != 401
@ -159,9 +170,14 @@ class TestConditionalAuthenticationBehavior:
except Exception: except Exception:
pass # If response is not JSON, that's fine pass # If response is not JSON, that's fine
@patch("cognee.modules.settings.get_settings.get_vectordb_config")
@patch("cognee.modules.settings.get_settings.get_llm_config") gsm_mod = importlib.import_module(
@patch("cognee.modules.users.methods.get_default_user.get_default_user", new_callable=AsyncMock) "cognee.modules.settings.get_settings"
)
@patch.object(gsm_mod, 'get_vectordb_config')
@patch.object(gsm_mod, 'get_llm_config')
@patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
def test_settings_endpoint_integration( def test_settings_endpoint_integration(
self, mock_get_default, mock_llm_config, mock_vector_config, client, mock_default_user self, mock_get_default, mock_llm_config, mock_vector_config, client, mock_default_user
): ):
@ -185,6 +201,8 @@ class TestConditionalAuthenticationBehavior:
response = client.get("/api/v1/settings") response = client.get("/api/v1/settings")
assert mock_get_default.call_count == 1
# Core test: authentication is not required (should not get 401) # Core test: authentication is not required (should not get 401)
assert response.status_code != 401 assert response.status_code != 401
# Note: This test verifies conditional authentication works for settings endpoint # Note: This test verifies conditional authentication works for settings endpoint
@ -197,7 +215,7 @@ class TestConditionalAuthenticationErrorHandling:
def client(self): def client(self):
return TestClient(app) return TestClient(app)
@patch("cognee.modules.users.methods.get_default_user.get_default_user", new_callable=AsyncMock) @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
def test_get_default_user_fails(self, mock_get_default, client): def test_get_default_user_fails(self, mock_get_default, client):
"""Test behavior when get_default_user fails (with current environment).""" """Test behavior when get_default_user fails (with current environment)."""
mock_get_default.side_effect = Exception("Database connection failed") mock_get_default.side_effect = Exception("Database connection failed")

106
cognee/tests/unit/modules/users/test_conditional_authentication.py
View file

@ -4,22 +4,22 @@ import pytest
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, patch
from uuid import uuid4 from uuid import uuid4
from types import SimpleNamespace from types import SimpleNamespace
import importlib
from cognee.modules.users.models import User from cognee.modules.users.models import User
gau_mod = importlib.import_module(
"cognee.modules.users.methods.get_authenticated_user"
)
class TestConditionalAuthentication: class TestConditionalAuthentication:
"""Test cases for conditional authentication functionality.""" """Test cases for conditional authentication functionality."""
@pytest.mark.asyncio @pytest.mark.asyncio
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"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( async def test_require_authentication_false_no_token_returns_default_user(
self, mock_get_default self, mock_get_default
): ):
@ -28,25 +28,16 @@ class TestConditionalAuthentication:
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)
mock_get_default.return_value = mock_default_user mock_get_default.return_value = mock_default_user
from cognee.modules.users.methods.get_authenticated_user import ( # Use gau_mod.get_authenticated_user instead
get_authenticated_user,
)
# Test with None user (no authentication) # Test with None user (no authentication)
result = await get_authenticated_user(user=None) result = await gau_mod.get_authenticated_user(user=None)
assert result == mock_default_user assert result == mock_default_user
mock_get_default.assert_called_once() mock_get_default.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"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( async def test_require_authentication_false_with_valid_user_returns_user(
self, mock_get_default self, mock_get_default
): ):
@ -59,22 +50,17 @@ class TestConditionalAuthentication:
is_verified=True, is_verified=True,
) )
from cognee.modules.users.methods.get_authenticated_user import ( # Use gau_mod.get_authenticated_user instead
get_authenticated_user,
)
# Test with authenticated user # Test with authenticated user
result = await get_authenticated_user(user=mock_authenticated_user) result = await gau_mod.get_authenticated_user(user=mock_authenticated_user)
assert result == mock_authenticated_user assert result == mock_authenticated_user
mock_get_default.assert_not_called() mock_get_default.assert_not_called()
@pytest.mark.asyncio @pytest.mark.asyncio
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"cognee.modules.users.methods.get_authenticated_user.REQUIRE_AUTHENTICATION", async def test_require_authentication_true_with_user_returns_user(self, mock_get_default):
True,
)
async def test_require_authentication_true_with_user_returns_user(self):
"""Test that when REQUIRE_AUTHENTICATION=true and user present, returns user.""" """Test that when REQUIRE_AUTHENTICATION=true and user present, returns user."""
mock_authenticated_user = User( mock_authenticated_user = User(
id=uuid4(), id=uuid4(),
@ -84,11 +70,9 @@ class TestConditionalAuthentication:
is_verified=True, is_verified=True,
) )
from cognee.modules.users.methods.get_authenticated_user import ( # Use gau_mod.get_authenticated_user instead
get_authenticated_user,
)
result = await get_authenticated_user(user=mock_authenticated_user) result = await gau_mod.get_authenticated_user(user=mock_authenticated_user)
assert result == mock_authenticated_user assert result == mock_authenticated_user
@ -144,7 +128,7 @@ class TestConditionalAuthenticationEnvironmentVariables:
from cognee.modules.users.methods.get_authenticated_user import ( from cognee.modules.users.methods.get_authenticated_user import (
REQUIRE_AUTHENTICATION, REQUIRE_AUTHENTICATION,
) )
importlib.invalidate_caches()
assert not REQUIRE_AUTHENTICATION assert not REQUIRE_AUTHENTICATION
def test_require_authentication_true(self): def test_require_authentication_true(self):
@ -211,38 +195,19 @@ class TestConditionalAuthenticationEdgeCases:
"""Test edge cases and error scenarios.""" """Test edge cases and error scenarios."""
@pytest.mark.asyncio @pytest.mark.asyncio
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"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): async def test_get_default_user_raises_exception(self, mock_get_default):
"""Test behavior when get_default_user raises an exception.""" """Test behavior when get_default_user raises an exception."""
from cognee.modules.users.methods.get_authenticated_user import (
get_authenticated_user,
)
mock_get_default.side_effect = Exception("Database error") mock_get_default.side_effect = Exception("Database error")
# This should propagate the exception # This should propagate the exception
with pytest.raises(Exception, match="Database error"): with pytest.raises(Exception, match="Database error"):
await get_authenticated_user(user=None) await gau_mod.get_authenticated_user(user=None)
@pytest.mark.asyncio @pytest.mark.asyncio
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"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): async def test_user_type_consistency(self, mock_get_default):
"""Test that the function always returns the same type.""" """Test that the function always returns the same type."""
from cognee.modules.users.methods.get_authenticated_user import (
get_authenticated_user,
)
mock_user = User( mock_user = User(
id=uuid4(), id=uuid4(),
email="test@example.com", email="test@example.com",
@ -255,11 +220,11 @@ class TestConditionalAuthenticationEdgeCases:
mock_get_default.return_value = mock_default_user mock_get_default.return_value = mock_default_user
# Test with user # Test with user
result1 = await get_authenticated_user(user=mock_user) result1 = await gau_mod.get_authenticated_user(user=mock_user)
assert result1 == mock_user assert result1 == mock_user
# Test with None # Test with None
result2 = await get_authenticated_user(user=None) result2 = await gau_mod.get_authenticated_user(user=None)
assert result2 == mock_default_user assert result2 == mock_default_user
# Both should have user-like interface # Both should have user-like interface
@ -277,14 +242,7 @@ class TestConditionalAuthenticationEdgeCases:
class TestAuthenticationScenarios: class TestAuthenticationScenarios:
"""Test specific authentication scenarios that could occur in FastAPI Users.""" """Test specific authentication scenarios that could occur in FastAPI Users."""
@patch( @patch.object(gau_mod, 'get_default_user', new_callable=AsyncMock)
"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): 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: Test fallback to default user for all scenarios where FastAPI Users returns None:
@ -299,19 +257,11 @@ class TestAuthenticationScenarios:
mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com") mock_default_user = SimpleNamespace(id=uuid4(), email="default@example.com")
mock_get_default.return_value = mock_default_user mock_get_default.return_value = mock_default_user
from cognee.modules.users.methods.get_authenticated_user import (
get_authenticated_user,
)
# All the above scenarios result in user=None being passed to our function # All the above scenarios result in user=None being passed to our function
result = await get_authenticated_user(user=None) result = await gau_mod.get_authenticated_user(user=None)
assert result == mock_default_user assert result == mock_default_user
mock_get_default.assert_called_once() 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): async def test_scenario_valid_active_user(self):
"""Scenario: Valid JWT and user exists and is active → returns the user.""" """Scenario: Valid JWT and user exists and is active → returns the user."""
mock_user = User( mock_user = User(
@ -322,9 +272,7 @@ class TestAuthenticationScenarios:
is_verified=True, is_verified=True,
) )
from cognee.modules.users.methods.get_authenticated_user import ( # Use gau_mod.get_authenticated_user instead
get_authenticated_user,
)
result = await get_authenticated_user(user=mock_user) result = await gau_mod.get_authenticated_user(user=mock_user)
assert result == mock_user assert result == mock_user