| """ |
| Coverage-driven tests for analytics_dashboard_routes.py (partial -> 70%+ target) |
| |
| API Endpoints Tested: |
| - GET /api/analytics/summary - Comprehensive analytics summary |
| - GET /api/analytics/sentiment - Sentiment analysis breakdown |
| - GET /api/analytics/response-times - Response time metrics |
| - GET /api/analytics/activity - Activity metrics and peak times |
| - GET /api/analytics/cross-platform - Cross-platform analytics |
| - POST /api/analytics/correlations - Cross-platform correlation analysis |
| - GET /api/analytics/correlations/{conversation_id}/timeline - Unified timeline |
| - GET /api/analytics/predictions/response-time - Response time prediction |
| - GET /api/analytics/recommendations/channel - Channel recommendation |
| - GET /api/analytics/bottlenecks - Bottleneck detection |
| - GET /api/analytics/patterns/{user_id} - User communication patterns |
| - GET /api/analytics/overview - High-level analytics overview |
| |
| Coverage Target Areas: |
| - Lines 1-30: Route initialization and dependencies |
| - Lines 30-85: Summary endpoint |
| - Lines 85-150: Sentiment and response-times endpoints |
| - Lines 150-250: Activity and cross-platform endpoints |
| - Lines 250-350: Correlations and timeline endpoints |
| - Lines 350-450: Predictive insights endpoints |
| - Lines 450-510: Patterns and overview endpoints |
| """ |
|
|
| from fastapi.testclient import TestClient |
| from unittest.mock import Mock, patch, AsyncMock, MagicMock |
| from datetime import datetime, timezone, timedelta |
| import pytest |
|
|
| from api.analytics_dashboard_routes import router |
| from core.message_analytics_engine import MessageAnalyticsEngine, get_message_analytics_engine |
| from core.cross_platform_correlation import ( |
| CrossPlatformCorrelationEngine, |
| get_cross_platform_correlation_engine, |
| LinkedConversation, |
| CorrelationStrength |
| ) |
| from core.predictive_insights import ( |
| PredictiveInsightsEngine, |
| get_predictive_insights_engine, |
| UrgencyLevel, |
| ResponseTimePrediction, |
| ChannelRecommendation, |
| BottleneckAlert, |
| CommunicationPattern, |
| RecommendationConfidence |
| ) |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def mock_analytics_engine(): |
| """Mock MessageAnalyticsEngine.""" |
| engine = Mock(spec=MessageAnalyticsEngine) |
| return engine |
|
|
|
|
| @pytest.fixture |
| def mock_correlation_engine(): |
| """Mock CrossPlatformCorrelationEngine.""" |
| engine = Mock(spec=CrossPlatformCorrelationEngine) |
| engine.linked_conversations = [] |
| engine.cross_platform_links = [] |
|
|
| |
| mock_conversation = Mock(spec=LinkedConversation) |
| mock_conversation.conversation_id = "conv-123" |
| mock_conversation.platforms = {"slack", "email"} |
| mock_conversation.participants = {"user-1", "user-2"} |
| mock_conversation.message_count = 5 |
| mock_conversation.correlation_strength = CorrelationStrength.STRONG |
| mock_conversation.unified_messages = [ |
| {"id": "msg-1", "platform": "slack", "content": "test"}, |
| {"id": "msg-2", "platform": "email", "content": "test"} |
| ] |
| engine.correlate_conversations.return_value = [mock_conversation] |
|
|
| |
| engine.get_unified_timeline.return_value = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test message", |
| "sender_name": "User 1", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "_correlation_source": "slack" |
| }, |
| { |
| "id": "msg-2", |
| "platform": "email", |
| "content": "Re: Test message", |
| "sender": "user-2", |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "_correlation_source": "email" |
| } |
| ] |
|
|
| return engine |
|
|
|
|
| @pytest.fixture |
| def mock_predictive_engine(): |
| """Mock PredictiveInsightsEngine.""" |
| engine = Mock(spec=PredictiveInsightsEngine) |
|
|
| |
| mock_prediction = Mock(spec=ResponseTimePrediction) |
| mock_prediction.user_id = "user-123" |
| mock_prediction.predicted_seconds = 1800 |
| mock_prediction.confidence = RecommendationConfidence.HIGH |
| mock_prediction.factors = ["time_of_day", "day_of_week"] |
| engine.predict_response_time.return_value = mock_prediction |
|
|
| |
| mock_recommendation = Mock(spec=ChannelRecommendation) |
| mock_recommendation.user_id = "user-123" |
| mock_recommendation.recommended_platform = "slack" |
| mock_recommendation.reason = "User most active on Slack" |
| mock_recommendation.confidence = RecommendationConfidence.HIGH |
| mock_recommendation.expected_response_time = 1200 |
| mock_recommendation.alternatives = ["teams", "email"] |
| engine.recommend_channel.return_value = mock_recommendation |
|
|
| |
| mock_bottleneck = Mock(spec=BottleneckAlert) |
| mock_bottleneck.severity = UrgencyLevel.HIGH |
| mock_bottleneck.thread_id = "thread-123" |
| mock_bottleneck.platform = "slack" |
| mock_bottleneck.description = "No response for 48 hours" |
| mock_bottleneck.affected_users = ["user-1", "user-2"] |
| mock_bottleneck.wait_time_seconds = 172800 |
| mock_bottleneck.suggested_action = "Send follow-up message" |
| engine.detect_bottlenecks.return_value = [mock_bottleneck] |
|
|
| |
| mock_pattern = Mock(spec=CommunicationPattern) |
| mock_pattern.user_id = "user-123" |
| mock_pattern.most_active_platform = "slack" |
| mock_pattern.most_active_hours = [9, 10, 11, 14, 15, 16] |
| mock_pattern.avg_response_time = 1500 |
| mock_pattern.response_probability_by_hour = {i: 0.1 for i in range(24)} |
| mock_pattern.preferred_message_types = ["direct", "channel"] |
| engine.get_user_pattern.return_value = mock_pattern |
|
|
| |
| engine.get_insights_summary.return_value = { |
| "users_analyzed": 10, |
| "bottlenecks_detected": 2, |
| "avg_response_time_all_users": 1200 |
| } |
|
|
| return engine |
|
|
|
|
| @pytest.fixture |
| def client(mock_analytics_engine, mock_correlation_engine, mock_predictive_engine): |
| """Test client with mocked engines.""" |
| from fastapi import FastAPI |
|
|
| app = FastAPI() |
| app.include_router(router) |
|
|
| def override_analytics(): |
| return mock_analytics_engine |
|
|
| def override_correlation(): |
| return mock_correlation_engine |
|
|
| def override_predictive(): |
| return mock_predictive_engine |
|
|
| app.dependency_overrides[get_message_analytics_engine] = override_analytics |
| app.dependency_overrides[get_cross_platform_correlation_engine] = override_correlation |
| app.dependency_overrides[get_predictive_insights_engine] = override_predictive |
|
|
| test_client = TestClient(app) |
| yield test_client |
|
|
| app.dependency_overrides.clear() |
|
|
|
|
| |
| |
| |
|
|
| class TestAnalyticsSummary: |
| """Test suite for /api/analytics/summary endpoint.""" |
|
|
| def test_get_analytics_summary_default(self, client): |
| """Cover summary endpoint with default parameters (lines 26-83).""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["time_window"] == "24h" |
| assert "message_stats" in result |
| assert "response_times" in result |
| assert "activity_peaks" in result |
| assert "cross_platform" in result |
|
|
| def test_get_analytics_summary_custom_time_window(self, client): |
| """Cover summary with custom time window.""" |
| response = client.get("/api/analytics/summary?time_window=7d") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == "7d" |
|
|
| @pytest.mark.parametrize("time_window", ["24h", "7d", "30d", "all"]) |
| def test_get_analytics_summary_time_windows(self, client, time_window): |
| """Cover various time window options.""" |
| response = client.get(f"/api/analytics/summary?time_window={time_window}") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == time_window |
|
|
| def test_get_analytics_summary_with_platform_filter(self, client): |
| """Cover summary with platform filter.""" |
| response = client.get("/api/analytics/summary?platform=slack") |
|
|
| assert response.status_code == 200 |
| assert "platform_filter" in response.json() |
|
|
| def test_get_analytics_summary_message_stats_structure(self, client): |
| """Cover message stats structure in summary.""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| stats = response.json()["message_stats"] |
| assert "total_messages" in stats |
| assert "total_words" in stats |
| assert "with_attachments" in stats |
| assert "with_mentions" in stats |
| assert "with_urls" in stats |
| assert "sentiment_distribution" in stats |
|
|
| def test_get_analytics_summary_response_times_structure(self, client): |
| """Cover response times structure in summary.""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| times = response.json()["response_times"] |
| assert "avg_response_seconds" in times |
| assert "median_response_seconds" in times |
| assert "p95_response_seconds" in times |
| assert "total_responses_analyzed" in times |
|
|
| def test_get_analytics_summary_cross_platform_structure(self, client): |
| """Cover cross-platform structure in summary.""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| cp = response.json()["cross_platform"] |
| assert "platforms" in cp |
| assert "most_active_platform" in cp |
| assert "total_messages" in cp |
|
|
| def test_get_analytics_summary_activity_peaks_structure(self, client): |
| """Cover activity peaks structure in summary.""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| peaks = response.json()["activity_peaks"] |
| assert "peak_days" in peaks |
| assert "messages_per_day" in peaks |
|
|
| def test_get_analytics_summary_sentiment_distribution(self, client): |
| """Cover sentiment distribution in summary.""" |
| response = client.get("/api/analytics/summary") |
|
|
| assert response.status_code == 200 |
| sentiment = response.json()["message_stats"]["sentiment_distribution"] |
| assert "positive" in sentiment |
| assert "negative" in sentiment |
| assert "neutral" in sentiment |
|
|
| def test_multiple_query_parameters(self, client): |
| """Cover multiple query parameters combined.""" |
| response = client.get("/api/analytics/summary?time_window=7d&platform=slack") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["time_window"] == "7d" |
| assert result["platform_filter"] == "slack" |
|
|
|
|
| |
| |
| |
|
|
| class TestSentimentAnalysis: |
| """Test suite for /api/analytics/sentiment endpoint.""" |
|
|
| def test_get_sentiment_analysis_default(self, client): |
| """Cover sentiment analysis with default parameters (lines 86-100+).""" |
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "sentiment_distribution" in result |
| assert "sentiment_trend" in result |
|
|
| def test_get_sentiment_with_platform_filter(self, client): |
| """Cover sentiment with platform filter.""" |
| response = client.get("/api/analytics/sentiment?platform=slack") |
|
|
| assert response.status_code == 200 |
| assert response.json()["platform"] == "slack" |
|
|
| def test_get_sentiment_with_time_window(self, client): |
| """Cover sentiment with time window.""" |
| response = client.get("/api/analytics/sentiment?time_window=7d") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == "7d" |
|
|
| @pytest.mark.parametrize("sentiment", ["positive", "negative", "neutral"]) |
| def test_sentiment_distribution_categories(self, client, sentiment): |
| """Cover sentiment distribution categories.""" |
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code == 200 |
| distribution = response.json().get("sentiment_distribution", {}) |
| assert sentiment in distribution |
|
|
| def test_sentiment_trend_structure(self, client): |
| """Cover sentiment trend structure.""" |
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code == 200 |
| assert "sentiment_trend" in response.json() |
| assert isinstance(response.json()["sentiment_trend"], list) |
|
|
| def test_sentiment_topics_positive(self, client): |
| """Cover most positive topics.""" |
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code == 200 |
| assert "most_positive_topics" in response.json() |
| assert isinstance(response.json()["most_positive_topics"], list) |
|
|
| def test_sentiment_topics_negative(self, client): |
| """Cover most negative topics.""" |
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code == 200 |
| assert "most_negative_topics" in response.json() |
| assert isinstance(response.json()["most_negative_topics"], list) |
|
|
| def test_sentiment_combined_parameters(self, client): |
| """Cover sentiment with multiple parameters.""" |
| response = client.get("/api/analytics/sentiment?platform=teams&time_window=30d") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["platform"] == "teams" |
| assert result["time_window"] == "30d" |
|
|
|
|
| |
| |
| |
|
|
| class TestResponseTimes: |
| """Test suite for /api/analytics/response-times endpoint.""" |
|
|
| def test_get_response_times_default(self, client): |
| """Cover response times with default parameters.""" |
| response = client.get("/api/analytics/response-times") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "avg_response_seconds" in result |
| assert "median_response_seconds" in result |
| assert "p95_response_seconds" in result |
| assert "p99_response_seconds" in result |
|
|
| def test_get_response_times_with_platform(self, client): |
| """Cover response times with platform filter.""" |
| response = client.get("/api/analytics/response-times?platform=slack") |
|
|
| assert response.status_code == 200 |
| assert response.json()["platform"] == "slack" |
|
|
| def test_get_response_times_with_time_window(self, client): |
| """Cover response times with time window.""" |
| response = client.get("/api/analytics/response-times?time_window=24h") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == "24h" |
|
|
| def test_response_time_distribution(self, client): |
| """Cover response time distribution structure.""" |
| response = client.get("/api/analytics/response-times") |
|
|
| assert response.status_code == 200 |
| assert "response_time_distribution" in response.json() |
| assert isinstance(response.json()["response_time_distribution"], list) |
|
|
| def test_slowest_threads(self, client): |
| """Cover slowest threads structure.""" |
| response = client.get("/api/analytics/response-times") |
|
|
| assert response.status_code == 200 |
| assert "slowest_threads" in response.json() |
| assert isinstance(response.json()["slowest_threads"], list) |
|
|
| def test_fastest_threads(self, client): |
| """Cover fastest threads structure.""" |
| response = client.get("/api/analytics/response-times") |
|
|
| assert response.status_code == 200 |
| assert "fastest_threads" in response.json() |
| assert isinstance(response.json()["fastest_threads"], list) |
|
|
| @pytest.mark.parametrize("time_window", ["24h", "7d", "30d"]) |
| def test_response_times_time_windows(self, client, time_window): |
| """Cover various time windows for response times.""" |
| response = client.get(f"/api/analytics/response-times?time_window={time_window}") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == time_window |
|
|
| def test_response_times_combined_parameters(self, client): |
| """Cover response times with combined parameters.""" |
| response = client.get("/api/analytics/response-times?platform=email&time_window=7d") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["platform"] == "email" |
| assert result["time_window"] == "7d" |
|
|
|
|
| |
| |
| |
|
|
| class TestActivityMetrics: |
| """Test suite for /api/analytics/activity endpoint.""" |
|
|
| def test_get_activity_metrics_default(self, client): |
| """Cover activity metrics with default parameters.""" |
| response = client.get("/api/analytics/activity") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "period" in result |
| assert result["period"] == "daily" |
|
|
| def test_get_activity_metrics_with_period_hourly(self, client): |
| """Cover activity metrics with hourly period.""" |
| response = client.get("/api/analytics/activity?period=hourly") |
|
|
| assert response.status_code == 200 |
| assert response.json()["period"] == "hourly" |
|
|
| def test_get_activity_metrics_with_period_weekly(self, client): |
| """Cover activity metrics with weekly period.""" |
| response = client.get("/api/analytics/activity?period=weekly") |
|
|
| assert response.status_code == 200 |
| assert response.json()["period"] == "weekly" |
|
|
| def test_get_activity_metrics_with_platform(self, client): |
| """Cover activity metrics with platform filter.""" |
| response = client.get("/api/analytics/activity?platform=slack") |
|
|
| assert response.status_code == 200 |
| assert response.json()["platform"] == "slack" |
|
|
| def test_activity_messages_per_hour(self, client): |
| """Cover messages per hour structure.""" |
| response = client.get("/api/analytics/activity") |
|
|
| assert response.status_code == 200 |
| assert "messages_per_hour" in response.json() |
|
|
| def test_activity_messages_per_day(self, client): |
| """Cover messages per day structure.""" |
| response = client.get("/api/analytics/activity") |
|
|
| assert response.status_code == 200 |
| assert "messages_per_day" in response.json() |
|
|
| def test_activity_peak_hours(self, client): |
| """Cover peak hours structure.""" |
| response = client.get("/api/analytics/activity") |
|
|
| assert response.status_code == 200 |
| assert "peak_hours" in response.json() |
| assert isinstance(response.json()["peak_hours"], list) |
|
|
| def test_activity_heatmap(self, client): |
| """Cover activity heatmap structure.""" |
| response = client.get("/api/analytics/activity") |
|
|
| assert response.status_code == 200 |
| assert "activity_heatmap" in response.json() |
| assert isinstance(response.json()["activity_heatmap"], list) |
|
|
|
|
| |
| |
| |
|
|
| class TestCrossPlatformAnalytics: |
| """Test suite for /api/analytics/cross-platform endpoint.""" |
|
|
| def test_get_cross_platform_analytics_default(self, client): |
| """Cover cross-platform analytics with default parameters.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "platforms" in result |
| assert "most_active_platform" in result |
|
|
| def test_cross_platform_includes_slack(self, client): |
| """Cover cross-platform includes Slack data.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| platforms = response.json()["platforms"] |
| assert "slack" in platforms |
|
|
| def test_cross_platform_includes_teams(self, client): |
| """Cover cross-platform includes Teams data.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| platforms = response.json()["platforms"] |
| assert "teams" in platforms |
|
|
| def test_cross_platform_includes_gmail(self, client): |
| """Cover cross-platform includes Gmail data.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| platforms = response.json()["platforms"] |
| assert "gmail" in platforms |
|
|
| def test_cross_platform_with_time_window(self, client): |
| """Cover cross-platform with time window.""" |
| response = client.get("/api/analytics/cross-platform?time_window=30d") |
|
|
| assert response.status_code == 200 |
| assert response.json()["time_window"] == "30d" |
|
|
| def test_cross_platform_message_counts(self, client): |
| """Cover platform message counts.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| platforms = response.json()["platforms"] |
| for platform_name, platform_data in platforms.items(): |
| assert "message_count" in platform_data |
| assert "sentiment" in platform_data |
| assert "avg_response_time" in platform_data |
|
|
| def test_cross_platform_sentiment_breakdown(self, client): |
| """Cover sentiment breakdown per platform.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| platforms = response.json()["platforms"] |
| for platform_data in platforms.values(): |
| sentiment = platform_data["sentiment"] |
| assert "positive" in sentiment |
| assert "negative" in sentiment |
| assert "neutral" in sentiment |
|
|
| def test_cross_platform_comparison(self, client): |
| """Cover platform comparison structure.""" |
| response = client.get("/api/analytics/cross-platform") |
|
|
| assert response.status_code == 200 |
| assert "platform_comparison" in response.json() |
| assert isinstance(response.json()["platform_comparison"], list) |
|
|
|
|
| |
| |
| |
|
|
| class TestCrossPlatformCorrelations: |
| """Test suite for /api/analytics/correlations endpoint.""" |
|
|
| def test_analyze_correlations_success(self, client, mock_correlation_engine): |
| """Cover cross-platform correlation analysis.""" |
| messages = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test message", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| }, |
| { |
| "id": "msg-2", |
| "platform": "email", |
| "content": "Re: Test message", |
| "sender": "user-2", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "linked_conversations" in result |
| assert "total_correlations" in result |
| assert "cross_platform_links" in result |
|
|
| def test_analyze_correlations_empty_list(self, client, mock_correlation_engine): |
| """Cover correlation with empty message list.""" |
| mock_correlation_engine.correlate_conversations.return_value = [] |
|
|
| response = client.post("/api/analytics/correlations", json=[]) |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["total_correlations"] == 0 |
|
|
| def test_correlation_conversation_structure(self, client): |
| """Cover linked conversation structure.""" |
| messages = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
| conversations = response.json()["linked_conversations"] |
| if len(conversations) > 0: |
| conv = conversations[0] |
| assert "conversation_id" in conv |
| assert "platforms" in conv |
| assert "participants" in conv |
| assert "message_count" in conv |
| assert "correlation_strength" in conv |
|
|
| def test_correlation_platforms_set(self, client): |
| """Cover platforms as set/list in conversation.""" |
| messages = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
| conversations = response.json()["linked_conversations"] |
| if len(conversations) > 0: |
| assert isinstance(conversations[0]["platforms"], list) |
|
|
| def test_correlation_multiple_messages(self, client): |
| """Cover correlation with multiple messages.""" |
| messages = [ |
| { |
| "id": f"msg-{i}", |
| "platform": "slack" if i % 2 == 0 else "email", |
| "content": f"Test message {i}", |
| "sender": f"user-{i % 3}", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| for i in range(10) |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
|
|
| def test_correlation_unified_message_count(self, client): |
| """Cover unified message count in conversation.""" |
| messages = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
| conversations = response.json()["linked_conversations"] |
| if len(conversations) > 0: |
| assert "unified_message_count" in conversations[0] |
|
|
| def test_correlation_cross_platform_links_count(self, client): |
| """Cover cross-platform links count.""" |
| messages = [ |
| { |
| "id": "msg-1", |
| "platform": "slack", |
| "content": "Test", |
| "sender": "user-1", |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| } |
| ] |
|
|
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "cross_platform_links" in result |
| assert isinstance(result["cross_platform_links"], int) |
|
|
|
|
| |
| |
| |
|
|
| class TestUnifiedTimeline: |
| """Test suite for /api/analytics/correlations/{conversation_id}/timeline endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_get_unified_timeline_success(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover unified timeline retrieval.""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert "conversation_id" in data |
| assert "message_count" in data |
| assert "messages" in data |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_get_unified_timeline_not_found(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover unified timeline for non-existent conversation.""" |
| mock_get_engine.return_value = mock_correlation_engine |
| mock_correlation_engine.get_unified_timeline.return_value = None |
|
|
| response = client.get("/api/analytics/correlations/nonexistent/timeline") |
|
|
| assert response.status_code == 404 |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_message_structure(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover timeline message structure.""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| messages = result["data"]["messages"] |
| else: |
| messages = result["messages"] |
| if len(messages) > 0: |
| msg = messages[0] |
| assert "id" in msg |
| assert "platform" in msg |
| assert "content" in msg |
| assert "sender" in msg |
| assert "timestamp" in msg |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_cross_platform_messages(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover timeline includes cross-platform messages.""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| messages = result["data"]["messages"] |
| else: |
| messages = result["messages"] |
| platforms = {msg["platform"] for msg in messages} |
| |
| assert len(platforms) >= 1 |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_sender_fallback(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover sender field fallback (sender_name -> sender).""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| messages = result["data"]["messages"] |
| else: |
| messages = result["messages"] |
| |
| if len(messages) > 0: |
| assert "sender" in messages[0] |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_correlation_source(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover timeline includes correlation source.""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| messages = result["data"]["messages"] |
| else: |
| messages = result["messages"] |
| if len(messages) > 0: |
| assert "source" in messages[0] |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_empty_conversation(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover timeline for conversation with no messages.""" |
| mock_get_engine.return_value = mock_correlation_engine |
| mock_correlation_engine.get_unified_timeline.return_value = [] |
|
|
| response = client.get("/api/analytics/correlations/conv-empty/timeline") |
|
|
| assert response.status_code in [200, 404] |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_timeline_message_count(self, mock_get_engine, client, mock_correlation_engine): |
| """Cover timeline message count accuracy.""" |
| mock_get_engine.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/correlations/conv-123/timeline") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert data["message_count"] == len(data["messages"]) |
|
|
|
|
| |
| |
| |
|
|
| class TestResponseTimePrediction: |
| """Test suite for /api/analytics/predictions/response-time endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_success(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover response time prediction.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "recipient" in result |
| assert "platform" in result |
| assert "predicted_response_seconds" in result |
| assert "predicted_response_minutes" in result |
| assert "confidence" in result |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_with_urgency(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover prediction with urgency parameter.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack&urgency=high") |
|
|
| assert response.status_code == 200 |
| assert response.json()["urgency"] == "high" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @pytest.mark.parametrize("urgency", ["low", "medium", "high", "urgent"]) |
| def test_predict_response_time_urgency_levels(self, mock_get_engine, client, mock_predictive_engine, urgency): |
| """Cover various urgency levels.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get( |
| f"/api/analytics/predictions/response-time?recipient=user-123&platform=slack&urgency={urgency}" |
| ) |
|
|
| assert response.status_code == 200 |
| assert response.json()["urgency"] == urgency |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_minutes_conversion(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover response time in minutes.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| predicted_seconds = result["predicted_response_seconds"] |
| predicted_minutes = result["predicted_response_minutes"] |
| assert abs(predicted_minutes - predicted_seconds / 60) < 0.01 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_factors(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover prediction factors.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack") |
|
|
| assert response.status_code == 200 |
| assert "factors" in response.json() |
| assert isinstance(response.json()["factors"], list) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_invalid_urgency(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover prediction with invalid urgency.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get( |
| "/api/analytics/predictions/response-time?recipient=user-123&platform=slack&urgency=invalid" |
| ) |
|
|
| |
| assert response.status_code in [200, 422, 500] |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predict_response_time_confidence_levels(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover prediction confidence levels.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack") |
|
|
| assert response.status_code == 200 |
| assert "confidence" in response.json() |
|
|
|
|
| |
| |
| |
|
|
| class TestChannelRecommendation: |
| """Test suite for /api/analytics/recommendations/channel endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_success(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover channel recommendation.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/recommendations/channel?recipient=user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "recipient" in result |
| assert "recommended_platform" in result |
| assert "reason" in result |
| assert "confidence" in result |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_with_message_type(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover recommendation with message type.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/recommendations/channel?recipient=user-123&message_type=urgent") |
|
|
| assert response.status_code == 200 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_with_urgency(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover recommendation with urgency.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/recommendations/channel?recipient=user-123&urgency=high") |
|
|
| assert response.status_code == 200 |
| assert response.json()["urgency"] == "high" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @pytest.mark.parametrize("urgency", ["low", "medium", "high", "urgent"]) |
| def test_recommend_channel_urgency_levels(self, mock_get_engine, client, mock_predictive_engine, urgency): |
| """Cover recommendation for various urgency levels.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get( |
| f"/api/analytics/recommendations/channel?recipient=user-123&urgency={urgency}" |
| ) |
|
|
| assert response.status_code == 200 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_alternatives(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover recommendation alternatives.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/recommendations/channel?recipient=user-123") |
|
|
| assert response.status_code == 200 |
| assert "alternatives" in response.json() |
| assert isinstance(response.json()["alternatives"], list) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_expected_response_time(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover expected response time in recommendation.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/recommendations/channel?recipient=user-123") |
|
|
| assert response.status_code == 200 |
| assert "expected_response_time_minutes" in response.json() |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_invalid_urgency(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover recommendation with invalid urgency.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get( |
| "/api/analytics/recommendations/channel?recipient=user-123&urgency=invalid" |
| ) |
|
|
| |
| assert response.status_code in [200, 422, 500] |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_recommend_channel_combined_parameters(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover recommendation with combined parameters.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get( |
| "/api/analytics/recommendations/channel?recipient=user-123&message_type=general&urgency=medium" |
| ) |
|
|
| assert response.status_code == 200 |
|
|
|
|
| |
| |
| |
|
|
| class TestBottleneckDetection: |
| """Test suite for /api/analytics/bottlenecks endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_detect_bottlenecks_success(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover bottleneck detection.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "total_bottlenecks" in result |
| assert "threshold_hours" in result |
| assert "bottlenecks" in result |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_detect_bottlenecks_with_threshold(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover bottleneck detection with custom threshold.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/bottlenecks?threshold_hours=48") |
|
|
| assert response.status_code == 200 |
| assert response.json()["threshold_hours"] == 48 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_bottleneck_structure(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover bottleneck structure.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code == 200 |
| bottlenecks = response.json()["bottlenecks"] |
| if len(bottlenecks) > 0: |
| b = bottlenecks[0] |
| assert "severity" in b |
| assert "thread_id" in b |
| assert "platform" in b |
| assert "description" in b |
| assert "affected_users" in b |
| assert "wait_time_hours" in b |
| assert "suggested_action" in b |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_bottleneck_affected_users_list(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover affected users as list.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code == 200 |
| bottlenecks = response.json()["bottlenecks"] |
| if len(bottlenecks) > 0: |
| assert isinstance(bottlenecks[0]["affected_users"], list) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_bottleneck_wait_time_conversion(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover wait time conversion to hours.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code == 200 |
| bottlenecks = response.json()["bottlenecks"] |
| if len(bottlenecks) > 0: |
| assert "wait_time_hours" in bottlenecks[0] |
| assert isinstance(bottlenecks[0]["wait_time_hours"], (int, float)) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_bottleneck_empty(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover bottleneck detection with no bottlenecks.""" |
| mock_get_engine.return_value = mock_predictive_engine |
| mock_predictive_engine.detect_bottlenecks.return_value = [] |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code == 200 |
| assert response.json()["total_bottlenecks"] == 0 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @pytest.mark.parametrize("threshold", [12.0, 24.0, 48.0, 72.0]) |
| def test_bottleneck_various_thresholds(self, mock_get_engine, client, mock_predictive_engine, threshold): |
| """Cover bottleneck detection with various thresholds.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get(f"/api/analytics/bottlenecks?threshold_hours={threshold}") |
|
|
| assert response.status_code == 200 |
| assert response.json()["threshold_hours"] == threshold |
|
|
|
|
| |
| |
| |
|
|
| class TestUserPatterns: |
| """Test suite for /api/analytics/patterns/{user_id} endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_get_user_patterns_success(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover user patterns retrieval.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert "user_id" in data |
| assert "most_active_platform" in data |
| assert "most_active_hours" in data |
| assert "avg_response_time_minutes" in data |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_get_user_patterns_not_found(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover user patterns for non-existent user.""" |
| mock_get_engine.return_value = mock_predictive_engine |
| mock_predictive_engine.get_user_pattern.return_value = None |
|
|
| response = client.get("/api/analytics/patterns/nonexistent") |
|
|
| assert response.status_code == 404 |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_user_patterns_active_hours(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover most active hours structure.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert "most_active_hours" in data |
| assert isinstance(data["most_active_hours"], list) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_user_patterns_response_probability(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover response probability by hour.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert "response_probability_by_hour" in data |
| assert isinstance(data["response_probability_by_hour"], dict) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_user_patterns_preferred_message_types(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover preferred message types.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| assert "preferred_message_types" in data |
| assert isinstance(data["preferred_message_types"], list) |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_user_patterns_response_time_minutes(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover response time in minutes.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| |
| assert "avg_response_time_minutes" in data |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_user_patterns_probability_24_hours(self, mock_get_engine, client, mock_predictive_engine): |
| """Cover response probability covers 24 hours.""" |
| mock_get_engine.return_value = mock_predictive_engine |
|
|
| response = client.get("/api/analytics/patterns/user-123") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| |
| if "data" in result: |
| data = result["data"] |
| else: |
| data = result |
| probability = data["response_probability_by_hour"] |
| |
| assert len(probability) == 24 |
|
|
|
|
| |
| |
| |
|
|
| class TestAnalyticsOverview: |
| """Test suite for /api/analytics/overview endpoint.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_get_analytics_overview_success(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover analytics overview.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert "timestamp" in result |
| assert "message_analytics" in result |
| assert "predictive_insights" in result |
| assert "cross_platform" in result |
| assert "health_status" in result |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_overview_timestamp(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover overview timestamp.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| assert "timestamp" in response.json() |
| |
| datetime.fromisoformat(response.json()["timestamp"].replace('Z', '+00:00')) |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_overview_message_analytics(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover message analytics in overview.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| msg_analytics = response.json()["message_analytics"] |
| assert "total_messages" in msg_analytics |
| assert "active_threads" in msg_analytics |
| assert "platforms_active" in msg_analytics |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_overview_predictive_insights(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover predictive insights in overview.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| insights = response.json()["predictive_insights"] |
| assert "users_analyzed" in insights |
| assert "bottlenecks_detected" in insights |
| assert "avg_response_time_minutes" in insights |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_overview_cross_platform(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover cross-platform in overview.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| cp = response.json()["cross_platform"] |
| assert "linked_conversations" in cp |
| assert "cross_platform_links" in cp |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_overview_health_status(self, mock_corr, mock_pred, mock_msg, client, mock_predictive_engine): |
| """Cover health status in overview.""" |
| mock_msg.return_value = mock_analytics_engine |
| mock_pred.return_value = mock_predictive_engine |
| mock_corr.return_value = mock_correlation_engine |
|
|
| response = client.get("/api/analytics/overview") |
|
|
| assert response.status_code == 200 |
| assert "health_status" in response.json() |
| assert isinstance(response.json()["health_status"], str) |
|
|
|
|
| |
| |
| |
|
|
| class TestErrorHandling: |
| """Test suite for error handling across all endpoints.""" |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| def test_summary_endpoint_exception_handling(self, mock_get_engine, client): |
| """Cover exception handling in summary endpoint.""" |
| mock_get_engine.side_effect = Exception("Analytics engine failed") |
|
|
| response = client.get("/api/analytics/summary") |
|
|
| |
| assert response.status_code in [500, 200] |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_predictive_endpoint_exception_handling(self, mock_get_engine, client): |
| """Cover exception handling in predictive endpoint.""" |
| mock_get_engine.side_effect = Exception("Predictive engine failed") |
|
|
| response = client.get("/api/analytics/predictions/response-time?recipient=user-123&platform=slack") |
|
|
| assert response.status_code in [500, 200] |
|
|
| @patch('api.analytics_dashboard_routes.get_cross_platform_correlation_engine') |
| def test_correlation_endpoint_exception_handling(self, mock_get_engine, client): |
| """Cover exception handling in correlation endpoint.""" |
| mock_get_engine.side_effect = Exception("Correlation engine failed") |
|
|
| messages = [{"id": "msg-1", "platform": "slack"}] |
| response = client.post("/api/analytics/correlations", json=messages) |
|
|
| assert response.status_code in [500, 200] |
|
|
| @patch('api.analytics_dashboard_routes.get_message_analytics_engine') |
| def test_sentiment_endpoint_exception_handling(self, mock_get_engine, client): |
| """Cover exception handling in sentiment endpoint.""" |
| mock_get_engine.side_effect = Exception("Analytics engine failed") |
|
|
| response = client.get("/api/analytics/sentiment") |
|
|
| assert response.status_code in [500, 200] |
|
|
| @patch('api.analytics_dashboard_routes.get_predictive_insights_engine') |
| def test_bottleneck_endpoint_exception_handling(self, mock_get_engine, client): |
| """Cover exception handling in bottleneck endpoint.""" |
| mock_get_engine.side_effect = Exception("Predictive engine failed") |
|
|
| response = client.get("/api/analytics/bottlenecks") |
|
|
| assert response.status_code in [500, 200] |
