""" Test script to verify Phase 4 analytics fixes. This script tests: 1. Success rate data type migration from string to numeric 2. Daily trends analytics implementation 3. Cross-user and system-wide analytics 4. Updated analytics calculations with numeric success_rate """ import asyncio import pytest from datetime import datetime, timedelta from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, text from core.database import get_async_session from db_models.evaluation import Evaluation, EvaluationStatus from db_models.user import User from services.result_service import ResultService class TestPhase4AnalyticsFixes: """Test suite for Phase 4 analytics fixes.""" @pytest.mark.asyncio async def test_success_rate_numeric_format(self): """Test that success_rate is stored and retrieved as numeric format.""" async for db in get_async_session(): try: # Create test evaluation with numeric success_rate evaluation = Evaluation( user_id=1, job_id="test-numeric-success-rate", status=EvaluationStatus.COMPLETED, model_config={"model": "test-model"}, pipeline_config={"pipeline": "test"}, success_rate=0.85, # Numeric format (0.0 to 1.0) execution_time_ms=1500, completed_at=datetime.utcnow() ) db.add(evaluation) await db.commit() # Retrieve and verify numeric format result = await db.execute( select(Evaluation).where(Evaluation.job_id == "test-numeric-success-rate") ) retrieved = result.scalar_one() assert retrieved.success_rate == 0.85 assert isinstance(retrieved.success_rate, float) # Clean up await db.delete(retrieved) await db.commit() finally: await db.close() @pytest.mark.asyncio async def test_daily_trends_analytics(self): """Test daily trends analytics functionality.""" async for db in get_async_session(): try: result_service = ResultService(db) # Create test data with multiple days base_date = datetime.utcnow() - timedelta(days=10) for i in range(10): evaluation = Evaluation( user_id=1, job_id=f"daily-trends-test-{i}", status=EvaluationStatus.COMPLETED, model_config={"model": "test-model"}, pipeline_config={"pipeline": "test"}, success_rate=0.5 + (i * 0.05), # Varying success rates execution_time_ms=1000 + (i * 100), completed_at=base_date + timedelta(days=i) ) db.add(evaluation) await db.commit() # Test daily trends trends = await result_service.get_daily_trends(user_id=1, days=15) assert "period" in trends assert "daily_data" in trends assert "trend_analysis" in trends assert len(trends["daily_data"]) == 10 # Verify trend analysis if trends["trend_analysis"]: assert "success_rate_trend" in trends["trend_analysis"] assert "trend_slope" in trends["trend_analysis"] # Clean up await db.execute( text("DELETE FROM evaluations WHERE job_id LIKE 'daily-trends-test-%'") ) await db.commit() finally: await db.close() @pytest.mark.asyncio async def test_system_wide_analytics(self): """Test system-wide analytics functionality.""" async for db in get_async_session(): try: result_service = ResultService(db) # Create test data for multiple users for user_id in [1, 2, 3]: for i in range(5): evaluation = Evaluation( user_id=user_id, job_id=f"system-test-{user_id}-{i}", status=EvaluationStatus.COMPLETED, model_config={"model": "test-model"}, pipeline_config={"pipeline": "test"}, success_rate=0.4 + (user_id * 0.2) + (i * 0.05), execution_time_ms=1000 + (i * 200), completed_at=datetime.utcnow() - timedelta(days=i) ) db.add(evaluation) await db.commit() # Test system-wide analytics analytics = await result_service.get_system_wide_analytics(days=10) assert "period" in analytics assert "total_evaluations" in analytics assert "unique_users" in analytics assert "system_performance" in analytics assert "user_distribution" in analytics assert analytics["total_evaluations"] == 15 assert analytics["unique_users"] == 3 # Verify system performance metrics perf = analytics["system_performance"] assert "overall_success_rate" in perf assert "success_rate_distribution" in perf # Clean up await db.execute( text("DELETE FROM evaluations WHERE job_id LIKE 'system-test-%'") ) await db.commit() finally: await db.close() @pytest.mark.asyncio async def test_cross_user_benchmarks(self): """Test cross-user benchmark functionality.""" async for db in get_async_session(): try: result_service = ResultService(db) # Create test data with varying performance per user user_performances = { 1: [0.9, 0.85, 0.88], # High performer 2: [0.6, 0.65, 0.62], # Medium performer 3: [0.3, 0.35, 0.32], # Lower performer } for user_id, success_rates in user_performances.items(): for i, success_rate in enumerate(success_rates): evaluation = Evaluation( user_id=user_id, job_id=f"benchmark-test-{user_id}-{i}", status=EvaluationStatus.COMPLETED, model_config={"model": "test-model"}, pipeline_config={"pipeline": "test"}, success_rate=success_rate, execution_time_ms=1000, completed_at=datetime.utcnow() - timedelta(days=i) ) db.add(evaluation) await db.commit() # Test cross-user benchmarks benchmarks = await result_service.get_cross_user_benchmarks( metric="success_rate", days=10 ) assert "period" in benchmarks assert "total_users" in benchmarks assert "rankings" in benchmarks assert "percentiles" in benchmarks assert "performance_insights" in benchmarks assert benchmarks["total_users"] == 3 assert len(benchmarks["rankings"]) == 3 # Verify ranking order (highest success rate first) rankings = benchmarks["rankings"] assert rankings[0]["user_id"] == 1 # Highest performer assert rankings[2]["user_id"] == 3 # Lowest performer # Verify percentiles percentiles = benchmarks["percentiles"] assert "p25" in percentiles assert "p50" in percentiles assert "p75" in percentiles assert "p90" in percentiles # Clean up await db.execute( text("DELETE FROM evaluations WHERE job_id LIKE 'benchmark-test-%'") ) await db.commit() finally: await db.close() @pytest.mark.asyncio async def test_updated_analytics_calculations(self): """Test that analytics calculations work with numeric success_rate.""" async for db in get_async_session(): try: result_service = ResultService(db) # Create test data with numeric success rates success_rates = [0.95, 0.85, 0.75, 0.65, 0.55, 0.45, 0.35, 0.25] for i, success_rate in enumerate(success_rates): evaluation = Evaluation( user_id=1, job_id=f"analytics-test-{i}", status=EvaluationStatus.COMPLETED, model_config={"model": "test-model"}, pipeline_config={"pipeline": "test"}, success_rate=success_rate, execution_time_ms=1000 + (i * 100), result_json={ "score_card": { "robustness_score": success_rate, "risk_score": 1.0 - success_rate }, "successful_attacks": [{"attack_type": "test"}] * int(success_rate * 10) }, completed_at=datetime.utcnow() - timedelta(days=i) ) db.add(evaluation) await db.commit() # Test updated analytics analytics = await result_service.get_result_analytics(user_id=1, days=15) assert "success_rate" in analytics assert "success_rate_distribution" in analytics assert "score_analytics" in analytics # Verify success rate calculation expected_avg = sum(success_rates) / len(success_rates) assert abs(analytics["success_rate"] - expected_avg) < 0.001 # Verify distribution dist = analytics["success_rate_distribution"] assert "excellent" in dist assert "good" in dist assert "average" in dist assert "poor" in dist # Verify score analytics score_analytics = analytics["score_analytics"] assert "robustness_score" in score_analytics assert "risk_score" in score_analytics # Clean up await db.execute( text("DELETE FROM evaluations WHERE job_id LIKE 'analytics-test-%'") ) await db.commit() finally: await db.close() @pytest.mark.asyncio async def test_analytics_helper_methods(self): """Test analytics helper methods.""" async for db in get_async_session(): try: result_service = ResultService(db) # Test standard deviation calculation values = [1.0, 2.0, 3.0, 4.0, 5.0] std = result_service._calculate_std(values) assert abs(std - 1.5811) < 0.01 # Expected std dev for 1-5 # Test percentile calculation p50 = result_service._calculate_percentile(values, 50) assert p50 == 3.0 # Median of 1-5 p25 = result_service._calculate_percentile(values, 25) assert p25 == 2.0 # 25th percentile # Test performance tier classification assert result_service._get_performance_tier(0.95) == "excellent" assert result_service._get_performance_tier(0.75) == "good" assert result_service._get_performance_tier(0.55) == "average" assert result_service._get_performance_tier(0.35) == "poor" # Test trend calculation success_rates = [0.5, 0.6, 0.7, 0.8, 0.9] execution_times = [1000, 1100, 1200, 1300, 1400] trends = result_service._calculate_trends(success_rates, execution_times) assert "success_rate_trend" in trends assert trends["success_rate_trend"] == "increasing" assert "trend_slope" in trends assert trends["trend_slope"] > 0 # Test momentum calculation momentum = result_service._calculate_momentum(success_rates) assert "momentum_score" in momentum assert "momentum_direction" in momentum # Test anomaly detection normal_values = [0.5, 0.52, 0.48, 0.51, 0.49, 0.95, 0.47] # Last value is anomaly anomalies = result_service._detect_anomalies(normal_values) assert len(anomalies) >= 1 assert anomalies[0]["type"] == "high" finally: await db.close() if __name__ == "__main__": # Run tests manually asyncio.run(TestPhase4AnalyticsFixes().test_success_rate_numeric_format()) asyncio.run(TestPhase4AnalyticsFixes().test_daily_trends_analytics()) asyncio.run(TestPhase4AnalyticsFixes().test_system_wide_analytics()) asyncio.run(TestPhase4AnalyticsFixes().test_cross_user_benchmarks()) asyncio.run(TestPhase4AnalyticsFixes().test_updated_analytics_calculations()) asyncio.run(TestPhase4AnalyticsFixes().test_analytics_helper_methods()) print("✅ All Phase 4 analytics tests passed!")