""" Test user-specific analytics functions """ import pytest import asyncio from datetime import datetime, timedelta from analytics.dashboard import ( get_user_statistics, get_user_analytics, get_authenticated_vs_anonymous_metrics, get_basic_stats, get_hourly_message_stats, get_session_stats, get_performance_stats ) from analytics.collectors import create_session, track_message from analytics.database import get_database, get_sessions_collection, get_messages_collection @pytest.fixture async def setup_test_data(): """Set up test data for user analytics tests""" # Create test sessions and messages with different user_ids # Authenticated user sessions auth_session1 = await create_session(user_agent="TestAgent", user_id="user_123") auth_session2 = await create_session(user_agent="TestAgent", user_id="user_456") # Anonymous sessions anon_session1 = await create_session(user_agent="TestAgent", user_id=None) anon_session2 = await create_session(user_agent="TestAgent") # Track messages for authenticated users await track_message( session_id=auth_session1.session_id, prompt_length=50, response_length=200, response_time_ms=1500, used_search=True, user_id="user_123" ) await track_message( session_id=auth_session1.session_id, prompt_length=30, response_length=150, response_time_ms=1200, used_search=False, user_id="user_123" ) await track_message( session_id=auth_session2.session_id, prompt_length=40, response_length=180, response_time_ms=1800, used_search=True, user_id="user_456" ) # Track messages for anonymous users await track_message( session_id=anon_session1.session_id, prompt_length=60, response_length=220, response_time_ms=1600, used_search=False, user_id=None ) await track_message( session_id=anon_session2.session_id, prompt_length=35, response_length=160, response_time_ms=1300, used_search=True ) return { "auth_sessions": [auth_session1, auth_session2], "anon_sessions": [anon_session1, anon_session2], "user_ids": ["user_123", "user_456"] } @pytest.mark.asyncio async def test_get_user_statistics(): """Test get_user_statistics function""" result = await get_user_statistics() # Should return a dictionary with expected keys assert isinstance(result, dict) assert "total_sessions" in result assert "authenticated_sessions" in result assert "anonymous_sessions" in result assert "authenticated_session_percentage" in result assert "total_messages" in result assert "authenticated_messages" in result assert "anonymous_messages" in result assert "authenticated_message_percentage" in result assert "unique_authenticated_users" in result assert "last_updated" in result # Values should be non-negative assert result["total_sessions"] >= 0 assert result["authenticated_sessions"] >= 0 assert result["anonymous_sessions"] >= 0 assert result["total_messages"] >= 0 assert result["authenticated_messages"] >= 0 assert result["anonymous_messages"] >= 0 assert result["unique_authenticated_users"] >= 0 # Percentages should be between 0 and 100 assert 0 <= result["authenticated_session_percentage"] <= 100 assert 0 <= result["authenticated_message_percentage"] <= 100 @pytest.mark.asyncio async def test_get_user_analytics_valid_user(): """Test get_user_analytics function with valid user_id""" result = await get_user_analytics("user_123") # Should return a dictionary with expected keys assert isinstance(result, dict) assert "user_id" in result assert result["user_id"] == "user_123" assert "total_sessions" in result assert "active_sessions" in result assert "total_messages" in result assert "messages_with_search" in result assert "search_usage_percentage" in result assert "avg_response_time_ms" in result assert "min_response_time_ms" in result assert "max_response_time_ms" in result assert "avg_session_duration_seconds" in result assert "max_session_duration_seconds" in result assert "ended_sessions" in result assert "avg_messages_per_session" in result assert "max_messages_per_session" in result assert "daily_activity_last_30_days" in result assert "last_updated" in result # Values should be non-negative assert result["total_sessions"] >= 0 assert result["active_sessions"] >= 0 assert result["total_messages"] >= 0 assert result["messages_with_search"] >= 0 assert result["avg_response_time_ms"] >= 0 assert result["min_response_time_ms"] >= 0 assert result["max_response_time_ms"] >= 0 assert result["avg_session_duration_seconds"] >= 0 assert result["max_session_duration_seconds"] >= 0 assert result["ended_sessions"] >= 0 assert result["avg_messages_per_session"] >= 0 assert result["max_messages_per_session"] >= 0 # Search usage percentage should be between 0 and 100 assert 0 <= result["search_usage_percentage"] <= 100 # Daily activity should be a list assert isinstance(result["daily_activity_last_30_days"], list) @pytest.mark.asyncio async def test_get_user_analytics_invalid_user(): """Test get_user_analytics function with invalid user_id""" # Test with None result = await get_user_analytics(None) assert "error" in result # Test with empty string result = await get_user_analytics("") assert "error" in result # Test with non-string result = await get_user_analytics(123) assert "error" in result @pytest.mark.asyncio async def test_get_authenticated_vs_anonymous_metrics(): """Test get_authenticated_vs_anonymous_metrics function""" result = await get_authenticated_vs_anonymous_metrics() # Should return a dictionary with expected keys assert isinstance(result, dict) assert "authenticated" in result assert "anonymous" in result assert "comparison" in result assert "last_updated" in result # Check authenticated metrics structure auth_metrics = result["authenticated"] assert "sessions" in auth_metrics assert "messages" in auth_metrics assert "avg_messages_per_session" in auth_metrics assert "avg_response_time_ms" in auth_metrics assert "search_usage_percentage" in auth_metrics assert "success_rate_percentage" in auth_metrics assert "sessions_with_search_percentage" in auth_metrics # Check anonymous metrics structure anon_metrics = result["anonymous"] assert "sessions" in anon_metrics assert "messages" in anon_metrics assert "avg_messages_per_session" in anon_metrics assert "avg_response_time_ms" in anon_metrics assert "search_usage_percentage" in anon_metrics assert "success_rate_percentage" in anon_metrics assert "sessions_with_search_percentage" in anon_metrics # Check comparison metrics structure comparison = result["comparison"] assert "total_sessions" in comparison assert "total_messages" in comparison assert "authenticated_percentage" in comparison # Values should be non-negative assert auth_metrics["sessions"] >= 0 assert auth_metrics["messages"] >= 0 assert anon_metrics["sessions"] >= 0 assert anon_metrics["messages"] >= 0 assert comparison["total_sessions"] >= 0 assert comparison["total_messages"] >= 0 # Percentages should be between 0 and 100 assert 0 <= auth_metrics["search_usage_percentage"] <= 100 assert 0 <= auth_metrics["success_rate_percentage"] <= 100 assert 0 <= auth_metrics["sessions_with_search_percentage"] <= 100 assert 0 <= anon_metrics["search_usage_percentage"] <= 100 assert 0 <= anon_metrics["success_rate_percentage"] <= 100 assert 0 <= anon_metrics["sessions_with_search_percentage"] <= 100 assert 0 <= comparison["authenticated_percentage"] <= 100 @pytest.mark.asyncio async def test_basic_stats_with_user_filter(): """Test get_basic_stats function with user_id filter""" # Test without filter result_all = await get_basic_stats() assert isinstance(result_all, dict) assert "filtered_by_user_id" not in result_all # Test with user filter result_filtered = await get_basic_stats(user_id="user_123") assert isinstance(result_filtered, dict) assert "filtered_by_user_id" in result_filtered assert result_filtered["filtered_by_user_id"] == "user_123" # Filtered results should have same or fewer counts assert result_filtered["total_sessions"] <= result_all["total_sessions"] assert result_filtered["total_messages"] <= result_all["total_messages"] @pytest.mark.asyncio async def test_hourly_stats_with_user_filter(): """Test get_hourly_message_stats function with user_id filter""" # Test without filter result_all = await get_hourly_message_stats(hours=24) assert isinstance(result_all, list) # Test with user filter result_filtered = await get_hourly_message_stats(hours=24, user_id="user_123") assert isinstance(result_filtered, list) # Each hour entry should have expected structure for hour_data in result_filtered: assert "hour" in hour_data assert "message_count" in hour_data assert "search_count" in hour_data assert "avg_response_time_ms" in hour_data assert "success_rate" in hour_data @pytest.mark.asyncio async def test_session_stats_with_user_filter(): """Test get_session_stats function with user_id filter""" # Test without filter result_all = await get_session_stats() assert isinstance(result_all, dict) assert "filtered_by_user_id" not in result_all # Test with user filter result_filtered = await get_session_stats(user_id="user_123") assert isinstance(result_filtered, dict) assert "filtered_by_user_id" in result_filtered assert result_filtered["filtered_by_user_id"] == "user_123" # Filtered results should have same or fewer counts assert result_filtered["total_sessions"] <= result_all["total_sessions"] assert result_filtered["active_sessions"] <= result_all["active_sessions"] @pytest.mark.asyncio async def test_performance_stats_with_user_filter(): """Test get_performance_stats function with user_id filter""" # Test without filter result_all = await get_performance_stats() assert isinstance(result_all, dict) assert "filtered_by_user_id" not in result_all # Test with user filter result_filtered = await get_performance_stats(user_id="user_123") assert isinstance(result_filtered, dict) assert "filtered_by_user_id" in result_filtered assert result_filtered["filtered_by_user_id"] == "user_123" # Filtered results should have same or fewer counts assert result_filtered["total_messages"] <= result_all["total_messages"] assert result_filtered["failed_messages"] <= result_all["failed_messages"] if __name__ == "__main__": # Run a simple test to verify functions work async def main(): print("Testing user analytics functions...") try: # Test user statistics user_stats = await get_user_statistics() print(f"User statistics: {user_stats}") # Test authenticated vs anonymous metrics auth_anon_metrics = await get_authenticated_vs_anonymous_metrics() print(f"Auth vs Anon metrics: {auth_anon_metrics}") # Test basic stats with filter basic_stats = await get_basic_stats(user_id="test_user") print(f"Basic stats (filtered): {basic_stats}") print("All tests completed successfully!") except Exception as e: print(f"Error during testing: {e}") asyncio.run(main())