Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| 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"] | |
| } | |
| 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 | |
| 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) | |
| 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 | |
| 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 | |
| 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"] | |
| 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 | |
| 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"] | |
| 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()) |