findEthics/Atlas

1"""2Test user-specific analytics functions3"""4 5import pytest6import asyncio7from datetime import datetime, timedelta8from analytics.dashboard import (9    get_user_statistics,10    get_user_analytics,11    get_authenticated_vs_anonymous_metrics,12    get_basic_stats,13    get_hourly_message_stats,14    get_session_stats,15    get_performance_stats16)17from analytics.collectors import create_session, track_message18from analytics.database import get_database, get_sessions_collection, get_messages_collection19 20@pytest.fixture21async def setup_test_data():22    """Set up test data for user analytics tests"""23    # Create test sessions and messages with different user_ids24    25    # Authenticated user sessions26    auth_session1 = await create_session(user_agent="TestAgent", user_id="user_123")27    auth_session2 = await create_session(user_agent="TestAgent", user_id="user_456")28    29    # Anonymous sessions30    anon_session1 = await create_session(user_agent="TestAgent", user_id=None)31    anon_session2 = await create_session(user_agent="TestAgent")32    33    # Track messages for authenticated users34    await track_message(35        session_id=auth_session1.session_id,36        prompt_length=50,37        response_length=200,38        response_time_ms=1500,39        used_search=True,40        user_id="user_123"41    )42    43    await track_message(44        session_id=auth_session1.session_id,45        prompt_length=30,46        response_length=150,47        response_time_ms=1200,48        used_search=False,49        user_id="user_123"50    )51    52    await track_message(53        session_id=auth_session2.session_id,54        prompt_length=40,55        response_length=180,56        response_time_ms=1800,57        used_search=True,58        user_id="user_456"59    )60    61    # Track messages for anonymous users62    await track_message(63        session_id=anon_session1.session_id,64        prompt_length=60,65        response_length=220,66        response_time_ms=1600,67        used_search=False,68        user_id=None69    )70    71    await track_message(72        session_id=anon_session2.session_id,73        prompt_length=35,74        response_length=160,75        response_time_ms=1300,76        used_search=True77    )78    79    return {80        "auth_sessions": [auth_session1, auth_session2],81        "anon_sessions": [anon_session1, anon_session2],82        "user_ids": ["user_123", "user_456"]83    }84 85@pytest.mark.asyncio86async def test_get_user_statistics():87    """Test get_user_statistics function"""88    result = await get_user_statistics()89    90    # Should return a dictionary with expected keys91    assert isinstance(result, dict)92    assert "total_sessions" in result93    assert "authenticated_sessions" in result94    assert "anonymous_sessions" in result95    assert "authenticated_session_percentage" in result96    assert "total_messages" in result97    assert "authenticated_messages" in result98    assert "anonymous_messages" in result99    assert "authenticated_message_percentage" in result100    assert "unique_authenticated_users" in result101    assert "last_updated" in result102    103    # Values should be non-negative104    assert result["total_sessions"] >= 0105    assert result["authenticated_sessions"] >= 0106    assert result["anonymous_sessions"] >= 0107    assert result["total_messages"] >= 0108    assert result["authenticated_messages"] >= 0109    assert result["anonymous_messages"] >= 0110    assert result["unique_authenticated_users"] >= 0111    112    # Percentages should be between 0 and 100113    assert 0 <= result["authenticated_session_percentage"] <= 100114    assert 0 <= result["authenticated_message_percentage"] <= 100115 116@pytest.mark.asyncio117async def test_get_user_analytics_valid_user():118    """Test get_user_analytics function with valid user_id"""119    result = await get_user_analytics("user_123")120    121    # Should return a dictionary with expected keys122    assert isinstance(result, dict)123    assert "user_id" in result124    assert result["user_id"] == "user_123"125    assert "total_sessions" in result126    assert "active_sessions" in result127    assert "total_messages" in result128    assert "messages_with_search" in result129    assert "search_usage_percentage" in result130    assert "avg_response_time_ms" in result131    assert "min_response_time_ms" in result132    assert "max_response_time_ms" in result133    assert "avg_session_duration_seconds" in result134    assert "max_session_duration_seconds" in result135    assert "ended_sessions" in result136    assert "avg_messages_per_session" in result137    assert "max_messages_per_session" in result138    assert "daily_activity_last_30_days" in result139    assert "last_updated" in result140    141    # Values should be non-negative142    assert result["total_sessions"] >= 0143    assert result["active_sessions"] >= 0144    assert result["total_messages"] >= 0145    assert result["messages_with_search"] >= 0146    assert result["avg_response_time_ms"] >= 0147    assert result["min_response_time_ms"] >= 0148    assert result["max_response_time_ms"] >= 0149    assert result["avg_session_duration_seconds"] >= 0150    assert result["max_session_duration_seconds"] >= 0151    assert result["ended_sessions"] >= 0152    assert result["avg_messages_per_session"] >= 0153    assert result["max_messages_per_session"] >= 0154    155    # Search usage percentage should be between 0 and 100156    assert 0 <= result["search_usage_percentage"] <= 100157    158    # Daily activity should be a list159    assert isinstance(result["daily_activity_last_30_days"], list)160 161@pytest.mark.asyncio162async def test_get_user_analytics_invalid_user():163    """Test get_user_analytics function with invalid user_id"""164    # Test with None165    result = await get_user_analytics(None)166    assert "error" in result167    168    # Test with empty string169    result = await get_user_analytics("")170    assert "error" in result171    172    # Test with non-string173    result = await get_user_analytics(123)174    assert "error" in result175 176@pytest.mark.asyncio177async def test_get_authenticated_vs_anonymous_metrics():178    """Test get_authenticated_vs_anonymous_metrics function"""179    result = await get_authenticated_vs_anonymous_metrics()180    181    # Should return a dictionary with expected keys182    assert isinstance(result, dict)183    assert "authenticated" in result184    assert "anonymous" in result185    assert "comparison" in result186    assert "last_updated" in result187    188    # Check authenticated metrics structure189    auth_metrics = result["authenticated"]190    assert "sessions" in auth_metrics191    assert "messages" in auth_metrics192    assert "avg_messages_per_session" in auth_metrics193    assert "avg_response_time_ms" in auth_metrics194    assert "search_usage_percentage" in auth_metrics195    assert "success_rate_percentage" in auth_metrics196    assert "sessions_with_search_percentage" in auth_metrics197    198    # Check anonymous metrics structure199    anon_metrics = result["anonymous"]200    assert "sessions" in anon_metrics201    assert "messages" in anon_metrics202    assert "avg_messages_per_session" in anon_metrics203    assert "avg_response_time_ms" in anon_metrics204    assert "search_usage_percentage" in anon_metrics205    assert "success_rate_percentage" in anon_metrics206    assert "sessions_with_search_percentage" in anon_metrics207    208    # Check comparison metrics structure209    comparison = result["comparison"]210    assert "total_sessions" in comparison211    assert "total_messages" in comparison212    assert "authenticated_percentage" in comparison213    214    # Values should be non-negative215    assert auth_metrics["sessions"] >= 0216    assert auth_metrics["messages"] >= 0217    assert anon_metrics["sessions"] >= 0218    assert anon_metrics["messages"] >= 0219    assert comparison["total_sessions"] >= 0220    assert comparison["total_messages"] >= 0221    222    # Percentages should be between 0 and 100223    assert 0 <= auth_metrics["search_usage_percentage"] <= 100224    assert 0 <= auth_metrics["success_rate_percentage"] <= 100225    assert 0 <= auth_metrics["sessions_with_search_percentage"] <= 100226    assert 0 <= anon_metrics["search_usage_percentage"] <= 100227    assert 0 <= anon_metrics["success_rate_percentage"] <= 100228    assert 0 <= anon_metrics["sessions_with_search_percentage"] <= 100229    assert 0 <= comparison["authenticated_percentage"] <= 100230 231@pytest.mark.asyncio232async def test_basic_stats_with_user_filter():233    """Test get_basic_stats function with user_id filter"""234    # Test without filter235    result_all = await get_basic_stats()236    assert isinstance(result_all, dict)237    assert "filtered_by_user_id" not in result_all238    239    # Test with user filter240    result_filtered = await get_basic_stats(user_id="user_123")241    assert isinstance(result_filtered, dict)242    assert "filtered_by_user_id" in result_filtered243    assert result_filtered["filtered_by_user_id"] == "user_123"244    245    # Filtered results should have same or fewer counts246    assert result_filtered["total_sessions"] <= result_all["total_sessions"]247    assert result_filtered["total_messages"] <= result_all["total_messages"]248 249@pytest.mark.asyncio250async def test_hourly_stats_with_user_filter():251    """Test get_hourly_message_stats function with user_id filter"""252    # Test without filter253    result_all = await get_hourly_message_stats(hours=24)254    assert isinstance(result_all, list)255    256    # Test with user filter257    result_filtered = await get_hourly_message_stats(hours=24, user_id="user_123")258    assert isinstance(result_filtered, list)259    260    # Each hour entry should have expected structure261    for hour_data in result_filtered:262        assert "hour" in hour_data263        assert "message_count" in hour_data264        assert "search_count" in hour_data265        assert "avg_response_time_ms" in hour_data266        assert "success_rate" in hour_data267 268@pytest.mark.asyncio269async def test_session_stats_with_user_filter():270    """Test get_session_stats function with user_id filter"""271    # Test without filter272    result_all = await get_session_stats()273    assert isinstance(result_all, dict)274    assert "filtered_by_user_id" not in result_all275    276    # Test with user filter277    result_filtered = await get_session_stats(user_id="user_123")278    assert isinstance(result_filtered, dict)279    assert "filtered_by_user_id" in result_filtered280    assert result_filtered["filtered_by_user_id"] == "user_123"281    282    # Filtered results should have same or fewer counts283    assert result_filtered["total_sessions"] <= result_all["total_sessions"]284    assert result_filtered["active_sessions"] <= result_all["active_sessions"]285 286@pytest.mark.asyncio287async def test_performance_stats_with_user_filter():288    """Test get_performance_stats function with user_id filter"""289    # Test without filter290    result_all = await get_performance_stats()291    assert isinstance(result_all, dict)292    assert "filtered_by_user_id" not in result_all293    294    # Test with user filter295    result_filtered = await get_performance_stats(user_id="user_123")296    assert isinstance(result_filtered, dict)297    assert "filtered_by_user_id" in result_filtered298    assert result_filtered["filtered_by_user_id"] == "user_123"299    300    # Filtered results should have same or fewer counts301    assert result_filtered["total_messages"] <= result_all["total_messages"]302    assert result_filtered["failed_messages"] <= result_all["failed_messages"]303 304if __name__ == "__main__":305    # Run a simple test to verify functions work306    async def main():307        print("Testing user analytics functions...")308        309        try:310            # Test user statistics311            user_stats = await get_user_statistics()312            print(f"User statistics: {user_stats}")313            314            # Test authenticated vs anonymous metrics315            auth_anon_metrics = await get_authenticated_vs_anonymous_metrics()316            print(f"Auth vs Anon metrics: {auth_anon_metrics}")317            318            # Test basic stats with filter319            basic_stats = await get_basic_stats(user_id="test_user")320            print(f"Basic stats (filtered): {basic_stats}")321            322            print("All tests completed successfully!")323            324        except Exception as e:325            print(f"Error during testing: {e}")326    327    asyncio.run(main())