foxhunt/testing/e2e/src/utils/dual_provider_utils.rs

//! Dual Provider Test Utilities
//!
//! Specialized utility functions for testing dual-provider (Databento/Benzinga) scenarios

use anyhow::Result;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::time::sleep;
use tracing::{info, warn};

use crate::mocks::DualProviderMockOrchestrator;

/// Dual-provider test configuration
#[derive(Debug, Clone)]
pub struct DualProviderTestConfig {
    pub databento_enabled: bool,
    pub benzinga_enabled: bool,
    pub failover_timeout: Duration,
    pub data_consistency_threshold: f64,
    pub latency_threshold: Duration,
    pub mock_mode: bool,
}

impl Default for DualProviderTestConfig {
    fn default() -> Self {
        Self {
            databento_enabled: true,
            benzinga_enabled: true,
            failover_timeout: Duration::from_secs(5),
            data_consistency_threshold: 0.8,
            latency_threshold: Duration::from_millis(100),
            mock_mode: true,
        }
    }
}

/// Test scenario for dual-provider testing
#[derive(Debug, Clone)]
pub struct DualProviderTestScenario {
    pub name: String,
    pub description: String,
    pub symbols: Vec<String>,
    pub duration: Duration,
    pub expected_events_min: usize,
    pub test_failover: bool,
    pub test_consistency: bool,
    pub test_performance: bool,
}

impl DualProviderTestScenario {
    /// Create a basic market data streaming test scenario
    pub fn basic_streaming() -> Self {
        Self {
            name: "basic_streaming".to_string(),
            description: "Test basic market data streaming from both providers".to_string(),
            symbols: vec!["AAPL".to_string(), "GOOGL".to_string()],
            duration: Duration::from_secs(10),
            expected_events_min: 5,
            test_failover: false,
            test_consistency: false,
            test_performance: false,
        }
    }
    
    /// Create a failover test scenario
    pub fn failover_test() -> Self {
        Self {
            name: "failover_test".to_string(),
            description: "Test provider failover mechanisms".to_string(),
            symbols: vec!["AAPL".to_string()],
            duration: Duration::from_secs(15),
            expected_events_min: 3,
            test_failover: true,
            test_consistency: false,
            test_performance: false,
        }
    }
    
    /// Create a data consistency test scenario
    pub fn consistency_test() -> Self {
        Self {
            name: "consistency_test".to_string(),
            description: "Test data consistency between providers".to_string(),
            symbols: vec!["AAPL".to_string(), "MSFT".to_string()],
            duration: Duration::from_secs(20),
            expected_events_min: 10,
            test_failover: false,
            test_consistency: true,
            test_performance: false,
        }
    }
    
    /// Create a performance test scenario
    pub fn performance_test() -> Self {
        Self {
            name: "performance_test".to_string(),
            description: "Test performance and latency of dual providers".to_string(),
            symbols: vec!["AAPL".to_string(), "GOOGL".to_string(), "MSFT".to_string()],
            duration: Duration::from_secs(30),
            expected_events_min: 20,
            test_failover: false,
            test_consistency: false,
            test_performance: true,
        }
    }
    
    /// Create a comprehensive test scenario
    pub fn comprehensive() -> Self {
        Self {
            name: "comprehensive".to_string(),
            description: "Comprehensive test including all aspects".to_string(),
            symbols: vec!["AAPL".to_string(), "GOOGL".to_string(), "MSFT".to_string(), "TSLA".to_string()],
            duration: Duration::from_secs(45),
            expected_events_min: 30,
            test_failover: true,
            test_consistency: true,
            test_performance: true,
        }
    }
}

/// Test result for dual-provider scenarios
#[derive(Debug, Clone)]
pub struct DualProviderTestResult {
    pub scenario_name: String,
    pub success: bool,
    pub duration: Duration,
    pub events_received: usize,
    pub databento_events: usize,
    pub benzinga_events: usize,
    pub average_latency: Option<Duration>,
    pub max_latency: Option<Duration>,
    pub consistency_score: Option<f64>,
    pub failover_time: Option<Duration>,
    pub error_message: Option<String>,
    pub metrics: HashMap<String, f64>,
}

impl DualProviderTestResult {
    pub fn success(scenario_name: String, duration: Duration) -> Self {
        Self {
            scenario_name,
            success: true,
            duration,
            events_received: 0,
            databento_events: 0,
            benzinga_events: 0,
            average_latency: None,
            max_latency: None,
            consistency_score: None,
            failover_time: None,
            error_message: None,
            metrics: HashMap::new(),
        }
    }
    
    pub fn failure(scenario_name: String, duration: Duration, error: String) -> Self {
        Self {
            scenario_name,
            success: false,
            duration,
            events_received: 0,
            databento_events: 0,
            benzinga_events: 0,
            average_latency: None,
            max_latency: None,
            consistency_score: None,
            failover_time: None,
            error_message: Some(error),
            metrics: HashMap::new(),
        }
    }
}

/// Dual-provider test utilities
pub struct DualProviderTestUtils;

impl DualProviderTestUtils {
    /// Setup mock providers for testing
    pub async fn setup_mock_providers() -> Result<DualProviderMockOrchestrator> {
        info!("Setting up mock dual providers");
        
        let orchestrator = DualProviderMockOrchestrator::new();
        orchestrator.start_all().await?;
        
        // Wait for providers to be ready
        sleep(Duration::from_secs(1)).await;
        
        let status = orchestrator.get_provider_status().await;
        if !status["databento"] || !status["benzinga"] {
            anyhow::bail!("Failed to start mock providers");
        }
        
        info!("✅ Mock dual providers ready");
        Ok(orchestrator)
    }
    
    /// Cleanup mock providers
    pub async fn cleanup_mock_providers(orchestrator: DualProviderMockOrchestrator) -> Result<()> {
        info!("Cleaning up mock dual providers");
        orchestrator.stop_all().await?;
        info!("✅ Mock dual providers cleaned up");
        Ok(())
    }
    
    /// Validate market data event structure
    pub fn validate_market_data_event(event: &Value, expected_provider: Option<&str>) -> Result<()> {
        // Check required fields
        let required_fields = ["symbol", "provider", "bid", "ask", "timestamp"];
        for field in &required_fields {
            if event.get(field).is_none() {
                anyhow::bail!("Missing required field: {}", field);
            }
        }
        
        // Validate provider if specified
        if let Some(provider) = expected_provider {
            let actual_provider = event.get("provider").unwrap().as_str().unwrap();
            if actual_provider != provider {
                anyhow::bail!("Expected provider '{}', got '{}'", provider, actual_provider);
            }
        }
        
        // Validate price data
        let bid = event.get("bid").unwrap().as_str().unwrap().parse::<f64>()?;
        let ask = event.get("ask").unwrap().as_str().unwrap().parse::<f64>()?;
        
        if bid <= 0.0 || ask <= 0.0 {
            anyhow::bail!("Invalid price data: bid={}, ask={}", bid, ask);
        }
        
        if ask <= bid {
            anyhow::bail!("Invalid spread: ask ({}) <= bid ({})", ask, bid);
        }
        
        // Validate timestamp
        let timestamp = event.get("timestamp").unwrap().as_u64().unwrap();
        let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos() as u64;
        
        if timestamp > now {
            warn!("Future timestamp detected: event={}, now={}", timestamp, now);
        }
        
        Ok(())
    }
    
    /// Validate news event structure
    pub fn validate_news_event(event: &Value) -> Result<()> {
        let required_fields = ["headline", "provider", "symbols", "timestamp", "sentiment_score"];
        for field in &required_fields {
            if event.get(field).is_none() {
                anyhow::bail!("Missing required field: {}", field);
            }
        }
        
        // Validate provider is Benzinga for news
        let provider = event.get("provider").unwrap().as_str().unwrap();
        if provider != "benzinga" {
            anyhow::bail!("News should come from Benzinga, got: {}", provider);
        }
        
        // Validate sentiment score
        let sentiment = event.get("sentiment_score").unwrap().as_f64().unwrap();
        if sentiment.abs() > 1.0 {
            anyhow::bail!("Invalid sentiment score: {}", sentiment);
        }
        
        // Validate symbols array
        let symbols = event.get("symbols").unwrap().as_array().unwrap();
        if symbols.is_empty() {
            anyhow::bail!("News event must have at least one symbol");
        }
        
        Ok(())
    }
    
    /// Calculate data consistency score between two datasets
    pub fn calculate_consistency_score(
        databento_data: &[Value],
        benzinga_data: &[Value],
        time_window: Duration,
    ) -> f64 {
        if databento_data.is_empty() || benzinga_data.is_empty() {
            return 0.0;
        }
        
        let mut matching_pairs = 0;
        let mut total_pairs = 0;
        
        for db_event in databento_data {
            if let Some(db_timestamp) = db_event.get("timestamp").and_then(|t| t.as_u64()) {
                if let Some(db_symbol) = db_event.get("symbol").and_then(|s| s.as_str()) {
                    // Look for corresponding Benzinga event within time window
                    for bz_event in benzinga_data {
                        if let Some(bz_timestamp) = bz_event.get("timestamp").and_then(|t| t.as_u64()) {
                            if let Some(bz_symbol) = bz_event.get("symbol").and_then(|s| s.as_str()) {
                                if db_symbol == bz_symbol {
                                    let time_diff = if db_timestamp > bz_timestamp {
                                        db_timestamp - bz_timestamp
                                    } else {
                                        bz_timestamp - db_timestamp
                                    };
                                    
                                    if Duration::from_nanos(time_diff) <= time_window {
                                        total_pairs += 1;
                                        
                                        // Check price consistency (within 1% tolerance)
                                        if let (Some(db_price), Some(bz_price)) = (
                                            db_event.get("last").and_then(|p| p.as_str()).and_then(|s| s.parse::<f64>().ok()),
                                            bz_event.get("last").and_then(|p| p.as_str()).and_then(|s| s.parse::<f64>().ok())
                                        ) {
                                            let price_diff = (db_price - bz_price).abs() / db_price;
                                            if price_diff <= 0.01 { // 1% tolerance
                                                matching_pairs += 1;
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        
        if total_pairs == 0 {
            0.0
        } else {
            matching_pairs as f64 / total_pairs as f64
        }
    }
    
    /// Measure latency for a series of operations
    pub fn measure_latencies<T>(operations: Vec<T>, operation_fn: impl Fn(T) -> Duration) -> (Duration, Duration, Duration) {
        let latencies: Vec<Duration> = operations.into_iter().map(operation_fn).collect();
        
        if latencies.is_empty() {
            return (Duration::from_nanos(0), Duration::from_nanos(0), Duration::from_nanos(0));
        }
        
        let total_nanos: u64 = latencies.iter().map(|d| d.as_nanos() as u64).sum();
        let avg_latency = Duration::from_nanos(total_nanos / latencies.len() as u64);
        
        let max_latency = latencies.iter().max().copied().unwrap_or(Duration::from_nanos(0));
        
        // Calculate P95 latency
        let mut sorted_latencies = latencies;
        sorted_latencies.sort();
        let p95_index = (sorted_latencies.len() as f64 * 0.95) as usize;
        let p95_latency = sorted_latencies.get(p95_index).copied().unwrap_or(Duration::from_nanos(0));
        
        (avg_latency, max_latency, p95_latency)
    }
    
    /// Generate test configuration for dual providers
    pub fn generate_test_configuration() -> Value {
        json!({
            "data_sources": {
                "databento": {
                    "enabled": true,
                    "api_key": "test_databento_key",
                    "base_url": "http://127.0.0.1:3001",
                    "timeout_ms": 5000,
                    "retry_attempts": 3,
                    "symbols": ["AAPL", "GOOGL", "MSFT", "TSLA"]
                },
                "benzinga": {
                    "enabled": true,
                    "api_key": "test_benzinga_key", 
                    "base_url": "http://127.0.0.1:3002",
                    "timeout_ms": 5000,
                    "retry_attempts": 3,
                    "symbols": ["AAPL", "GOOGL", "MSFT", "TSLA"]
                },
                "priority": {
                    "market_data": ["databento", "benzinga"],
                    "news": ["benzinga"],
                    "historical": ["databento", "benzinga"]
                },
                "failover": {
                    "enabled": true,
                    "timeout_ms": 5000,
                    "retry_delay_ms": 1000,
                    "max_retries": 3
                }
            }
        })
    }
    
    /// Wait for provider to become healthy
    pub async fn wait_for_provider_health(
        check_fn: impl Fn() -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>,
        timeout: Duration,
        check_interval: Duration,
    ) -> bool {
        let start = Instant::now();
        
        while start.elapsed() < timeout {
            if check_fn().await {
                return true;
            }
            
            sleep(check_interval).await;
        }
        
        false
    }
    
    /// Create test summary from multiple test results
    pub fn create_test_summary(results: &[DualProviderTestResult]) -> Value {
        let total_tests = results.len();
        let successful_tests = results.iter().filter(|r| r.success).count();
        let failed_tests = total_tests - successful_tests;
        
        let total_events: usize = results.iter().map(|r| r.events_received).sum();
        let total_databento_events: usize = results.iter().map(|r| r.databento_events).sum();
        let total_benzinga_events: usize = results.iter().map(|r| r.benzinga_events).sum();
        
        let avg_latency = if results.iter().any(|r| r.average_latency.is_some()) {
            let latencies: Vec<Duration> = results.iter().filter_map(|r| r.average_latency).collect();
            if !latencies.is_empty() {
                let total_nanos: u64 = latencies.iter().map(|d| d.as_nanos() as u64).sum();
                Some(Duration::from_nanos(total_nanos / latencies.len() as u64))
            } else {
                None
            }
        } else {
            None
        };
        
        let avg_consistency = if results.iter().any(|r| r.consistency_score.is_some()) {
            let scores: Vec<f64> = results.iter().filter_map(|r| r.consistency_score).collect();
            if !scores.is_empty() {
                Some(scores.iter().sum::<f64>() / scores.len() as f64)
            } else {
                None
            }
        } else {
            None
        };
        
        json!({
            "summary": {
                "total_tests": total_tests,
                "successful_tests": successful_tests,
                "failed_tests": failed_tests,
                "success_rate": if total_tests > 0 { successful_tests as f64 / total_tests as f64 } else { 0.0 }
            },
            "events": {
                "total": total_events,
                "databento": total_databento_events,
                "benzinga": total_benzinga_events,
                "databento_percentage": if total_events > 0 { total_databento_events as f64 / total_events as f64 * 100.0 } else { 0.0 },
                "benzinga_percentage": if total_events > 0 { total_benzinga_events as f64 / total_events as f64 * 100.0 } else { 0.0 }
            },
            "performance": {
                "average_latency_ms": avg_latency.map(|d| d.as_millis()),
                "consistency_score": avg_consistency
            },
            "test_results": results
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_validate_market_data_event() {
        let valid_event = json!({
            "symbol": "AAPL",
            "provider": "databento",
            "bid": "149.95",
            "ask": "150.05",
            "timestamp": 1640995200000u64
        });
        
        assert!(DualProviderTestUtils::validate_market_data_event(&valid_event, Some("databento")).is_ok());
        
        let invalid_event = json!({
            "symbol": "AAPL",
            "provider": "databento"
            // Missing required fields
        });
        
        assert!(DualProviderTestUtils::validate_market_data_event(&invalid_event, None).is_err());
    }
    
    #[test]
    fn test_calculate_consistency_score() {
        let databento_data = vec![
            json!({
                "symbol": "AAPL",
                "last": "150.00",
                "timestamp": 1640995200000u64
            })
        ];
        
        let benzinga_data = vec![
            json!({
                "symbol": "AAPL", 
                "last": "150.10",
                "timestamp": 1640995201000u64
            })
        ];
        
        let score = DualProviderTestUtils::calculate_consistency_score(
            &databento_data,
            &benzinga_data,
            Duration::from_secs(5)
        );
        
        assert!(score >= 0.0 && score <= 1.0);
    }
    
    #[test]
    fn test_test_scenarios() {
        let basic = DualProviderTestScenario::basic_streaming();
        assert_eq!(basic.name, "basic_streaming");
        assert!(!basic.test_failover);
        
        let failover = DualProviderTestScenario::failover_test();
        assert_eq!(failover.name, "failover_test");
        assert!(failover.test_failover);
        
        let comprehensive = DualProviderTestScenario::comprehensive();
        assert_eq!(comprehensive.name, "comprehensive");
        assert!(comprehensive.test_failover);
        assert!(comprehensive.test_consistency);
        assert!(comprehensive.test_performance);
    }
}