#!/usr/bin/env python3 """ Comprehensive tests for anonymous mode functionality This test file verifies that the system properly handles anonymous users (null user_id values) across all components including API requests, analytics, database operations, and dashboard metrics. Requirements tested: - 1.1: Anonymous requests processed with full functionality - 1.2: System treats null/empty user_id as anonymous - 3.1: Anonymous requests handled efficiently - 3.3: Anonymous usage aggregated without individual tracking """ import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Load environment variables try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # dotenv not available, continue without it import pytest import asyncio from fastapi.testclient import TestClient from unittest.mock import patch, AsyncMock from datetime import datetime, timedelta # Import the main app and functions to test from app import app, validate_user_id, normalize_user_id from analytics.collectors import create_session, track_message from analytics.dashboard import ( count_anonymous_sessions, count_authenticated_sessions, count_anonymous_messages, count_authenticated_messages, count_all_sessions, count_all_messages, get_basic_stats, get_user_statistics, get_authenticated_vs_anonymous_metrics ) from analytics.database import get_sessions_collection, get_messages_collection class TestAnonymousRequestProcessing: """Test that null user_id requests work correctly (Requirements 1.1, 1.2)""" def test_normalize_user_id_function(self): """Test normalize_user_id function handles various inputs correctly""" # Test None input assert normalize_user_id(None) is None # Test empty string assert normalize_user_id("") is None # Test whitespace-only strings assert normalize_user_id(" ") is None assert normalize_user_id("\t") is None assert normalize_user_id("\n") is None assert normalize_user_id(" \t\n ") is None # Test valid user_id assert normalize_user_id("user123") == "user123" assert normalize_user_id(" user123 ") == "user123" def test_validate_user_id_function(self): """Test validate_user_id function properly handles anonymous users""" # Test None input (anonymous) assert validate_user_id(None) is None # Test empty string (anonymous) assert validate_user_id("") is None assert validate_user_id(" ") is None # Test valid user_id assert validate_user_id("user123") == "user123" assert validate_user_id("user_123") == "user_123" assert validate_user_id("user-123") == "user-123" # Test invalid user_id raises exception with pytest.raises(Exception): validate_user_id("user@123") with pytest.raises(Exception): validate_user_id("a" * 256) # Too long @patch('app.run_gemini_inference') @patch('app.search_web_combined') def test_anonymous_chat_request_api(self, mock_search, mock_gemini): """Test that chat API works with anonymous requests""" # Mock the external dependencies mock_search.return_value = [] mock_gemini.return_value = "Test response" client = TestClient(app) # Test request without user_id field response = client.post("/chat", json={ "prompt": "Hello, how are you?", "use_search": False }) assert response.status_code == 200 data = response.json() assert "response" in data assert data["response"] == "Test response" @patch('app.run_gemini_inference') @patch('app.search_web_combined') def test_anonymous_chat_request_with_null_user_id(self, mock_search, mock_gemini): """Test that chat API works with explicit null user_id""" # Mock the external dependencies mock_search.return_value = [] mock_gemini.return_value = "Test response" client = TestClient(app) # Test request with explicit null user_id response = client.post("/chat", json={ "prompt": "Hello, how are you?", "user_id": None, "use_search": False }) assert response.status_code == 200 data = response.json() assert "response" in data assert data["response"] == "Test response" @patch('app.run_gemini_inference') @patch('app.search_web_combined') def test_anonymous_chat_request_with_empty_user_id(self, mock_search, mock_gemini): """Test that chat API works with empty string user_id""" # Mock the external dependencies mock_search.return_value = [] mock_gemini.return_value = "Test response" client = TestClient(app) # Test request with empty string user_id response = client.post("/chat", json={ "prompt": "Hello, how are you?", "user_id": "", "use_search": False }) assert response.status_code == 200 data = response.json() assert "response" in data assert data["response"] == "Test response" class TestAnonymousAnalytics: """Test that analytics properly count anonymous vs authenticated usage (Requirements 3.1, 3.3)""" @pytest.mark.asyncio async def test_count_anonymous_sessions(self): """Test counting anonymous sessions""" count = await count_anonymous_sessions() assert isinstance(count, int) assert count >= 0 @pytest.mark.asyncio async def test_count_authenticated_sessions(self): """Test counting authenticated sessions""" count = await count_authenticated_sessions() assert isinstance(count, int) assert count >= 0 @pytest.mark.asyncio async def test_count_anonymous_messages(self): """Test counting anonymous messages""" count = await count_anonymous_messages() assert isinstance(count, int) assert count >= 0 @pytest.mark.asyncio async def test_count_authenticated_messages(self): """Test counting authenticated messages""" count = await count_authenticated_messages() assert isinstance(count, int) assert count >= 0 @pytest.mark.asyncio async def test_total_counts_consistency(self): """Test that anonymous + authenticated = total counts""" # Get all counts total_sessions = await count_all_sessions() anonymous_sessions = await count_anonymous_sessions() authenticated_sessions = await count_authenticated_sessions() total_messages = await count_all_messages() anonymous_messages = await count_anonymous_messages() authenticated_messages = await count_authenticated_messages() # Verify consistency assert total_sessions == anonymous_sessions + authenticated_sessions assert total_messages == anonymous_messages + authenticated_messages @pytest.mark.asyncio async def test_create_anonymous_session(self): """Test creating a session with null user_id""" # Create anonymous session session = await create_session(user_agent="TestAgent", user_id=None) assert session is not None assert session.user_id is None assert session.session_id is not None assert session.user_agent == "TestAgent" @pytest.mark.asyncio async def test_create_authenticated_session(self): """Test creating a session with valid user_id""" # Create authenticated session session = await create_session(user_agent="TestAgent", user_id="test_user_123") assert session is not None assert session.user_id == "test_user_123" assert session.session_id is not None assert session.user_agent == "TestAgent" @pytest.mark.asyncio async def test_track_anonymous_message(self): """Test tracking a message with null user_id""" # Create anonymous session first session = await create_session(user_agent="TestAgent", user_id=None) # Track anonymous message message = await track_message( session_id=session.session_id, prompt_length=50, response_length=200, response_time_ms=1500, used_search=True, user_id=None ) assert message is not None assert message.user_id is None assert message.session_id == session.session_id assert message.prompt_length == 50 assert message.response_length == 200 assert message.response_time_ms == 1500 assert message.used_search is True @pytest.mark.asyncio async def test_track_authenticated_message(self): """Test tracking a message with valid user_id""" # Create authenticated session first session = await create_session(user_agent="TestAgent", user_id="test_user_123") # Track authenticated message message = await track_message( session_id=session.session_id, prompt_length=30, response_length=150, response_time_ms=1200, used_search=False, user_id="test_user_123" ) assert message is not None assert message.user_id == "test_user_123" assert message.session_id == session.session_id assert message.prompt_length == 30 assert message.response_length == 150 assert message.response_time_ms == 1200 assert message.used_search is False class TestAnonymousDatabaseOperations: """Test that database operations handle null user_id values (Requirements 3.1, 3.3)""" @pytest.mark.asyncio async def test_database_query_with_null_user_id(self): """Test database queries work with null user_id values""" sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection is None or messages_collection is None: pytest.skip("Database not available") # Query for anonymous sessions anonymous_sessions = await sessions_collection.find({ "$or": [ {"user_id": None}, {"user_id": {"$exists": False}} ] }).to_list(length=10) # Should return a list (even if empty) assert isinstance(anonymous_sessions, list) # Query for authenticated sessions authenticated_sessions = await sessions_collection.find({ "user_id": {"$ne": None, "$exists": True} }).to_list(length=10) # Should return a list (even if empty) assert isinstance(authenticated_sessions, list) @pytest.mark.asyncio async def test_database_aggregation_with_null_user_id(self): """Test database aggregation works with null user_id values""" messages_collection = await get_messages_collection() if messages_collection is None: pytest.skip("Database not available") # Aggregate messages by user_id (including null) pipeline = [ { "$group": { "_id": "$user_id", "count": {"$sum": 1}, "avg_response_time": {"$avg": "$response_time_ms"} } } ] results = await messages_collection.aggregate(pipeline).to_list(length=100) # Should return a list of aggregation results assert isinstance(results, list) # Check if we have anonymous users (user_id = None) anonymous_result = next((r for r in results if r["_id"] is None), None) if anonymous_result: assert "count" in anonymous_result assert "avg_response_time" in anonymous_result assert anonymous_result["count"] > 0 class TestAnonymousDashboardMetrics: """Test that dashboard displays anonymous metrics correctly (Requirements 3.3)""" @pytest.mark.asyncio async def test_get_basic_stats_includes_anonymous_metrics(self): """Test that basic stats include anonymous usage metrics""" stats = await get_basic_stats() # Should return a dictionary assert isinstance(stats, dict) # Should include basic metrics expected_keys = [ "total_sessions", "total_messages", "active_sessions", "messages_today", "search_usage_percentage", "average_response_time_ms" ] for key in expected_keys: assert key in stats, f"Missing key: {key}" assert isinstance(stats[key], (int, float)), f"Invalid type for {key}" assert stats[key] >= 0, f"Negative value for {key}" @pytest.mark.asyncio async def test_get_user_statistics_includes_anonymous_breakdown(self): """Test that user statistics include anonymous vs authenticated breakdown""" try: stats = await get_user_statistics() # Should return a dictionary assert isinstance(stats, dict) # Should include anonymous vs authenticated breakdown expected_keys = [ "total_sessions", "authenticated_sessions", "anonymous_sessions", "authenticated_session_percentage", "total_messages", "authenticated_messages", "anonymous_messages", "authenticated_message_percentage", "unique_authenticated_users" ] for key in expected_keys: assert key in stats, f"Missing key: {key}" assert isinstance(stats[key], (int, float)), f"Invalid type for {key}" assert stats[key] >= 0, f"Negative value for {key}" # Verify percentages are valid assert 0 <= stats["authenticated_session_percentage"] <= 100 assert 0 <= stats["authenticated_message_percentage"] <= 100 # Verify totals are consistent assert stats["total_sessions"] == stats["authenticated_sessions"] + stats["anonymous_sessions"] assert stats["total_messages"] == stats["authenticated_messages"] + stats["anonymous_messages"] except ImportError: pytest.skip("get_user_statistics function not available") @pytest.mark.asyncio async def test_get_authenticated_vs_anonymous_metrics(self): """Test authenticated vs anonymous comparison metrics""" try: metrics = await get_authenticated_vs_anonymous_metrics() # Should return a dictionary assert isinstance(metrics, dict) # Should include authenticated and anonymous sections assert "authenticated" in metrics assert "anonymous" in metrics assert "comparison" in metrics # Check authenticated metrics structure auth_metrics = metrics["authenticated"] expected_auth_keys = [ "sessions", "messages", "avg_messages_per_session", "avg_response_time_ms", "search_usage_percentage", "success_rate_percentage" ] for key in expected_auth_keys: assert key in auth_metrics, f"Missing authenticated key: {key}" assert isinstance(auth_metrics[key], (int, float)), f"Invalid type for authenticated {key}" assert auth_metrics[key] >= 0, f"Negative value for authenticated {key}" # Check anonymous metrics structure anon_metrics = metrics["anonymous"] expected_anon_keys = [ "sessions", "messages", "avg_messages_per_session", "avg_response_time_ms", "search_usage_percentage", "success_rate_percentage" ] for key in expected_anon_keys: assert key in anon_metrics, f"Missing anonymous key: {key}" assert isinstance(anon_metrics[key], (int, float)), f"Invalid type for anonymous {key}" assert anon_metrics[key] >= 0, f"Negative value for anonymous {key}" # Check comparison metrics comparison = metrics["comparison"] assert "total_sessions" in comparison assert "total_messages" in comparison assert "authenticated_percentage" in comparison # Verify totals are consistent assert comparison["total_sessions"] == auth_metrics["sessions"] + anon_metrics["sessions"] assert comparison["total_messages"] == auth_metrics["messages"] + anon_metrics["messages"] except ImportError: pytest.skip("get_authenticated_vs_anonymous_metrics function not available") def test_analytics_dashboard_endpoint(self): """Test that analytics dashboard endpoint works and includes anonymous metrics""" client = TestClient(app) response = client.get("/analytics/dashboard") # Should return 200 OK assert response.status_code == 200 # Should return HTML content assert "text/html" in response.headers.get("content-type", "") # Should include anonymous-related content in HTML html_content = response.text.lower() assert "anonymous" in html_content assert "authenticated" in html_content def test_analytics_stats_endpoint(self): """Test that analytics stats endpoint works""" client = TestClient(app) response = client.get("/analytics/stats") # Should return 200 OK assert response.status_code == 200 # Should return JSON assert response.headers.get("content-type") == "application/json" # Should include basic stats data = response.json() assert isinstance(data, dict) class TestAnonymousIntegrationScenarios: """Integration tests for complete anonymous user workflows""" @pytest.mark.asyncio async def test_complete_anonymous_user_workflow(self): """Test complete workflow: anonymous session creation -> message tracking -> analytics""" # Step 1: Create anonymous session session = await create_session(user_agent="TestAgent/1.0", user_id=None) assert session is not None assert session.user_id is None # Step 2: Track multiple anonymous messages message1 = await track_message( session_id=session.session_id, prompt_length=25, response_length=100, response_time_ms=800, used_search=False, user_id=None ) message2 = await track_message( session_id=session.session_id, prompt_length=40, response_length=180, response_time_ms=1200, used_search=True, user_id=None ) assert message1 is not None assert message2 is not None assert message1.user_id is None assert message2.user_id is None # Step 3: Verify analytics include these anonymous interactions anonymous_sessions_count = await count_anonymous_sessions() anonymous_messages_count = await count_anonymous_messages() assert anonymous_sessions_count > 0 assert anonymous_messages_count > 0 @pytest.mark.asyncio async def test_mixed_anonymous_authenticated_analytics(self): """Test analytics work correctly with mix of anonymous and authenticated users""" # Create anonymous session and message anon_session = await create_session(user_agent="TestAgent", user_id=None) await track_message( session_id=anon_session.session_id, prompt_length=30, response_length=120, response_time_ms=1000, used_search=False, user_id=None ) # Create authenticated session and message auth_session = await create_session(user_agent="TestAgent", user_id="test_user_456") await track_message( session_id=auth_session.session_id, prompt_length=35, response_length=140, response_time_ms=1100, used_search=True, user_id="test_user_456" ) # Verify counts are accurate total_sessions = await count_all_sessions() anonymous_sessions = await count_anonymous_sessions() authenticated_sessions = await count_authenticated_sessions() total_messages = await count_all_messages() anonymous_messages = await count_anonymous_messages() authenticated_messages = await count_authenticated_messages() # Verify consistency assert total_sessions == anonymous_sessions + authenticated_sessions assert total_messages == anonymous_messages + authenticated_messages assert anonymous_sessions > 0 assert authenticated_sessions > 0 assert anonymous_messages > 0 assert authenticated_messages > 0 def run_anonymous_mode_tests(): """Run all anonymous mode tests""" print("๐Ÿงช Running Anonymous Mode Tests") print("=" * 50) # Test request processing request_test = TestAnonymousRequestProcessing() request_test.test_normalize_user_id_function() request_test.test_validate_user_id_function() print("โœ… Anonymous request processing tests passed") print("\n๐ŸŽ‰ ANONYMOUS MODE TESTS COMPLETED!") print("Note: Some async tests require pytest to run properly") print("Run with: python -m pytest tests/test_anonymous_mode.py -v") if __name__ == "__main__": run_anonymous_mode_tests()