diff --git a/services/tests/integration_service_communication_tests.rs b/services/tests/integration_service_communication_tests.rs deleted file mode 100644 index 2c5dde42d..000000000 --- a/services/tests/integration_service_communication_tests.rs +++ /dev/null @@ -1,1281 +0,0 @@ -//! Comprehensive Integration Tests for Service Communication -//! -//! This test suite provides extensive coverage for communication between all -//! Foxhunt HFT system services including gRPC, message passing, authentication, -//! and end-to-end workflow testing. - -use anyhow::Result; -use std::collections::HashMap; -use std::sync::Arc; -use std::time::{Duration, Instant}; -use tokio::sync::{broadcast, RwLock}; -use tonic::transport::{Channel, Server}; -use tonic::{Code, Request, Response, Status}; -use uuid::Uuid; - -// Test utilities and mocks -// use trading_engine::prelude::*; // REMOVED - prelude does not exist - -#[cfg(test)] -mod integration_service_communication_tests { - use super::*; - use futures_util::StreamExt; - use tokio_stream::wrappers::ReceiverStream; - - // ======================================================================== - // Service Communication Infrastructure Tests - // ======================================================================== - - #[tokio::test] - async fn test_grpc_server_startup_and_health_check() { - // Test server startup for all services - let trading_server = MockTradingServer::new().await; - assert!(trading_server.is_ok()); - - let backtesting_server = MockBacktestingServer::new().await; - assert!(backtesting_server.is_ok()); - - let ml_training_server = MockMLTrainingServer::new().await; - assert!(ml_training_server.is_ok()); - - // Test health check endpoints - let health_status = trading_server.unwrap().health_check().await; - assert!(health_status.is_ok()); - - let health_response = health_status.unwrap(); - assert_eq!(health_response.status, ServiceStatus::Healthy); - assert!(!health_response.version.is_empty()); - assert!(health_response.uptime_seconds > 0); - } - - #[tokio::test] - async fn test_service_discovery_and_connectivity() { - // Initialize service registry - let mut registry = ServiceRegistry::new(); - - // Register services - let trading_service = ServiceEndpoint { - service_type: ServiceType::Trading, - address: "127.0.0.1".to_string(), - port: 50051, - health_check_path: "/health".to_string(), - metadata: HashMap::new(), - }; - - let backtesting_service = ServiceEndpoint { - service_type: ServiceType::Backtesting, - address: "127.0.0.1".to_string(), - port: 50052, - health_check_path: "/health".to_string(), - metadata: HashMap::new(), - }; - - registry.register_service(trading_service.clone()).await.expect("Failed to register trading service"); - registry.register_service(backtesting_service.clone()).await.expect("Failed to register backtesting service"); - - // Test service discovery - let discovered_services = registry.discover_services(ServiceType::Trading).await; - assert!(discovered_services.is_ok()); - - let services = discovered_services.unwrap(); - assert_eq!(services.len(), 1); - assert_eq!(services[0].service_type, ServiceType::Trading); - assert_eq!(services[0].port, 50051); - - // Test connectivity - let connectivity_test = registry.test_connectivity(&trading_service).await; - assert!(connectivity_test.is_ok()); - } - - #[tokio::test] - async fn test_authentication_and_authorization() { - let mut auth_service = MockAuthService::new(); - - // Test JWT token generation - let token_request = TokenRequest { - username: "trader_001".to_string(), - password: "secure_password".to_string(), - service: ServiceType::Trading, - permissions: vec![ - Permission::Trading, - Permission::RiskManagement, - Permission::MarketData, - ], - }; - - let token_response = auth_service.generate_token(token_request).await; - assert!(token_response.is_ok()); - - let token = token_response.unwrap(); - assert!(!token.access_token.is_empty()); - assert!(!token.refresh_token.is_empty()); - assert!(token.expires_in > 0); - - // Test token validation - let validation_result = auth_service.validate_token(&token.access_token).await; - assert!(validation_result.is_ok()); - - let validation = validation_result.unwrap(); - assert!(validation.is_valid); - assert_eq!(validation.username, "trader_001"); - assert!(validation.permissions.contains(&Permission::Trading)); - - // Test authorization for specific actions - let auth_check = auth_service.check_authorization( - &token.access_token, - ServiceType::Trading, - "place_order" - ).await; - assert!(auth_check.is_ok()); - assert!(auth_check.unwrap()); - - // Test unauthorized action - let unauth_check = auth_service.check_authorization( - &token.access_token, - ServiceType::Trading, - "admin_shutdown" - ).await; - assert!(unauth_check.is_ok()); - assert!(!unauth_check.unwrap()); - } - - #[tokio::test] - async fn test_mTLS_certificate_validation() { - let tls_config = MockTLSConfig::new(); - - // Test certificate loading - let cert_result = tls_config.load_server_certificates().await; - assert!(cert_result.is_ok()); - - let certificates = cert_result.unwrap(); - assert!(!certificates.certificate_chain.is_empty()); - assert!(!certificates.private_key.is_empty()); - assert!(certificates.ca_certificate.is_some()); - - // Test certificate validation - let validation_result = tls_config.validate_client_certificate(&certificates.certificate_chain[0]).await; - assert!(validation_result.is_ok()); - - let validation = validation_result.unwrap(); - assert!(validation.is_valid); - assert!(!validation.subject.is_empty()); - assert!(!validation.issuer.is_empty()); - assert!(validation.not_after > chrono::Utc::now()); - - // Test expired certificate rejection - let expired_cert = MockTLSConfig::create_expired_certificate(); - let expired_validation = tls_config.validate_client_certificate(&expired_cert).await; - assert!(expired_validation.is_ok()); - - let expired_result = expired_validation.unwrap(); - assert!(!expired_result.is_valid); - assert!(expired_result.errors.contains(&"Certificate expired".to_string())); - } - - // ======================================================================== - // Trading Service Communication Tests - // ======================================================================== - - #[tokio::test] - async fn test_trading_service_order_management() { - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - - // Test order placement - let order_request = PlaceOrderRequest { - symbol: "EURUSD".to_string(), - side: OrderSide::Buy as i32, - order_type: OrderType::Limit as i32, - quantity: 10000.0, - price: Some(1.2345), - time_in_force: TimeInForce::GoodTillCancel as i32, - client_order_id: Uuid::new_v4().to_string(), - trader_id: "TRADER_001".to_string(), - }; - - let order_response = trading_client.place_order(Request::new(order_request)).await; - assert!(order_response.is_ok()); - - let response = order_response.unwrap().into_inner(); - assert!(response.success); - assert!(!response.order_id.is_empty()); - assert!(response.error_message.is_empty()); - - // Test order status query - let status_request = GetOrderStatusRequest { - order_id: response.order_id.clone(), - }; - - let status_response = trading_client.get_order_status(Request::new(status_request)).await; - assert!(status_response.is_ok()); - - let status = status_response.unwrap().into_inner(); - assert_eq!(status.order_id, response.order_id); - assert_eq!(status.status, OrderStatus::Pending as i32); - assert_eq!(status.symbol, "EURUSD"); - - // Test order cancellation - let cancel_request = CancelOrderRequest { - order_id: response.order_id.clone(), - trader_id: "TRADER_001".to_string(), - }; - - let cancel_response = trading_client.cancel_order(Request::new(cancel_request)).await; - assert!(cancel_response.is_ok()); - - let cancellation = cancel_response.unwrap().into_inner(); - assert!(cancellation.success); - assert_eq!(cancellation.order_id, response.order_id); - } - - #[tokio::test] - async fn test_trading_service_risk_management_integration() { - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - - // Test position limits validation - let position_check = PositionLimitCheckRequest { - trader_id: "TRADER_001".to_string(), - symbol: "EURUSD".to_string(), - side: OrderSide::Buy as i32, - quantity: 100000.0, // Large position to trigger limit - current_price: 1.2345, - }; - - let check_response = trading_client.check_position_limits(Request::new(position_check)).await; - assert!(check_response.is_ok()); - - let check_result = check_response.unwrap().into_inner(); - assert!(!check_result.approved); // Should be rejected due to size - assert!(!check_result.violations.is_empty()); - assert!(check_result.violations[0].violation_type.contains("position_limit")); - - // Test risk metrics query - let risk_request = GetRiskMetricsRequest { - trader_id: Some("TRADER_001".to_string()), - portfolio_level: true, - }; - - let risk_response = trading_client.get_risk_metrics(Request::new(risk_request)).await; - assert!(risk_response.is_ok()); - - let metrics = risk_response.unwrap().into_inner(); - assert!(metrics.var_1_day >= 0.0); - assert!(metrics.var_10_day >= 0.0); - assert!(metrics.maximum_drawdown >= 0.0); - assert!(metrics.sharpe_ratio.is_finite()); - } - - #[tokio::test] - async fn test_trading_service_market_data_streaming() { - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - - // Subscribe to market data stream - let subscription_request = MarketDataSubscriptionRequest { - symbols: vec!["EURUSD".to_string(), "GBPUSD".to_string()], - data_types: vec![ - MarketDataType::Tick as i32, - MarketDataType::OrderBook as i32, - ], - include_trade_data: true, - }; - - let stream_response = trading_client.subscribe_market_data(Request::new(subscription_request)).await; - assert!(stream_response.is_ok()); - - let mut stream = stream_response.unwrap().into_inner(); - - // Test receiving market data - let start_time = Instant::now(); - let mut tick_count = 0; - let mut orderbook_count = 0; - - while start_time.elapsed() < Duration::from_secs(5) && (tick_count < 10 || orderbook_count < 10) { - if let Some(data_result) = stream.next().await { - assert!(data_result.is_ok()); - - let market_data = data_result.unwrap(); - match market_data.data_type() { - MarketDataType::Tick => { - tick_count += 1; - assert!(!market_data.symbol.is_empty()); - assert!(market_data.bid > 0.0); - assert!(market_data.ask > 0.0); - assert!(market_data.timestamp_nanos > 0); - } - MarketDataType::OrderBook => { - orderbook_count += 1; - assert!(!market_data.symbol.is_empty()); - assert!(!market_data.order_book_levels.is_empty()); - } - _ => {} - } - } - } - - assert!(tick_count > 0); - assert!(orderbook_count > 0); - } - - // ======================================================================== - // Backtesting Service Communication Tests - // ======================================================================== - - #[tokio::test] - async fn test_backtesting_service_workflow() { - let mut backtesting_client = MockBacktestingServiceClient::new().await.expect("Failed to create backtesting client"); - - // Start a backtest - let backtest_request = StartBacktestRequest { - strategy_name: "MeanReversionStrategy".to_string(), - symbols: vec!["EURUSD".to_string()], - start_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(30)).timestamp_nanos_opt().unwrap_or(0), - end_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(1)).timestamp_nanos_opt().unwrap_or(0), - initial_capital: 100000.0, - parameters: { - let mut params = HashMap::new(); - params.insert("lookback_period".to_string(), "20".to_string()); - params.insert("entry_threshold".to_string(), "2.0".to_string()); - params.insert("save_results".to_string(), "true".to_string()); - params - }, - }; - - let start_response = backtesting_client.start_backtest(Request::new(backtest_request)).await; - assert!(start_response.is_ok()); - - let backtest_info = start_response.unwrap().into_inner(); - assert!(backtest_info.success); - assert!(!backtest_info.backtest_id.is_empty()); - assert!(backtest_info.estimated_duration_seconds > 0); - - // Monitor backtest progress - let progress_request = SubscribeBacktestProgressRequest { - backtest_id: backtest_info.backtest_id.clone(), - }; - - let progress_stream = backtesting_client.subscribe_backtest_progress(Request::new(progress_request)).await; - assert!(progress_stream.is_ok()); - - let mut stream = progress_stream.unwrap().into_inner(); - let mut progress_updates = 0; - let mut final_status = None; - - // Wait for progress updates - while let Some(progress_result) = stream.next().await { - if progress_updates >= 10 { break; } // Prevent infinite loop - - assert!(progress_result.is_ok()); - let progress = progress_result.unwrap(); - - assert_eq!(progress.backtest_id, backtest_info.backtest_id); - assert!(progress.progress_percent >= 0.0 && progress.progress_percent <= 100.0); - assert!(!progress.current_date.is_empty()); - - progress_updates += 1; - - if progress.progress_percent == 100.0 { - final_status = Some(BacktestStatus::from(progress.status)); - break; - } - } - - assert!(progress_updates > 0); - - // Check final status - let status_request = GetBacktestStatusRequest { - backtest_id: backtest_info.backtest_id.clone(), - }; - - let status_response = backtesting_client.get_backtest_status(Request::new(status_request)).await; - assert!(status_response.is_ok()); - - let final_status_response = status_response.unwrap().into_inner(); - assert_eq!(final_status_response.backtest_id, backtest_info.backtest_id); - assert!(final_status_response.progress_percent == 100.0 || - matches!(BacktestStatus::from(final_status_response.status), BacktestStatus::Completed | BacktestStatus::Failed)); - - // Get backtest results (if completed) - if BacktestStatus::from(final_status_response.status) == BacktestStatus::Completed { - let results_request = GetBacktestResultsRequest { - backtest_id: backtest_info.backtest_id.clone(), - include_trades: true, - include_metrics: true, - }; - - let results_response = backtesting_client.get_backtest_results(Request::new(results_request)).await; - assert!(results_response.is_ok()); - - let results = results_response.unwrap().into_inner(); - assert_eq!(results.backtest_id, backtest_info.backtest_id); - assert!(results.metrics.is_some()); - - let metrics = results.metrics.unwrap(); - assert!(metrics.total_return.is_finite()); - assert!(metrics.sharpe_ratio.is_finite()); - assert!(metrics.maximum_drawdown >= 0.0); - } - } - - #[tokio::test] - async fn test_backtesting_service_concurrent_backtests() { - let mut backtesting_client = MockBacktestingServiceClient::new().await.expect("Failed to create backtesting client"); - - let mut backtest_handles = Vec::new(); - let num_concurrent = 3; - - // Start multiple backtests concurrently - for i in 0..num_concurrent { - let mut client_clone = backtesting_client.clone(); - let strategy_name = format!("Strategy_{}", i); - - let handle = tokio::spawn(async move { - let backtest_request = StartBacktestRequest { - strategy_name: strategy_name.clone(), - symbols: vec!["EURUSD".to_string()], - start_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(10)).timestamp_nanos_opt().unwrap_or(0), - end_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(1)).timestamp_nanos_opt().unwrap_or(0), - initial_capital: 50000.0, - parameters: HashMap::new(), - }; - - let result = client_clone.start_backtest(Request::new(backtest_request)).await; - (strategy_name, result) - }); - - backtest_handles.push(handle); - } - - // Wait for all backtests to start - let mut started_backtests = Vec::new(); - for handle in backtest_handles { - let (strategy_name, result) = handle.await.expect("Task failed"); - assert!(result.is_ok(), "Backtest start failed for {}: {:?}", strategy_name, result.err()); - - let response = result.unwrap().into_inner(); - assert!(response.success); - started_backtests.push((strategy_name, response.backtest_id)); - } - - assert_eq!(started_backtests.len(), num_concurrent); - - // Verify all backtests are tracked - for (strategy_name, backtest_id) in started_backtests { - let status_request = GetBacktestStatusRequest { - backtest_id: backtest_id.clone(), - }; - - let status_response = backtesting_client.get_backtest_status(Request::new(status_request)).await; - assert!(status_response.is_ok(), "Failed to get status for {}", strategy_name); - - let status = status_response.unwrap().into_inner(); - assert_eq!(status.backtest_id, backtest_id); - assert!(matches!(BacktestStatus::from(status.status), - BacktestStatus::Queued | BacktestStatus::Running | BacktestStatus::Completed)); - } - } - - // ======================================================================== - // ML Training Service Communication Tests - // ======================================================================== - - #[tokio::test] - async fn test_ml_training_service_model_training() { - let mut ml_client = MockMLTrainingServiceClient::new().await.expect("Failed to create ML client"); - - // Start model training - let training_request = StartTrainingRequest { - model_type: MLModelType::DQN as i32, - model_name: "DQN_EURUSD_v1".to_string(), - training_config: Some(TrainingConfig { - batch_size: 64, - learning_rate: 0.001, - num_epochs: 100, - validation_split: 0.2, - early_stopping: true, - save_checkpoints: true, - }), - dataset_config: Some(DatasetConfig { - symbols: vec!["EURUSD".to_string()], - start_date: "2024-01-01".to_string(), - end_date: "2024-12-01".to_string(), - features: vec![ - "price_returns".to_string(), - "volume".to_string(), - "volatility".to_string(), - ], - target: "future_return_1h".to_string(), - }), - compute_config: Some(ComputeConfig { - use_gpu: true, - max_memory_gb: 8.0, - num_workers: 4, - distributed: false, - }), - }; - - let training_response = ml_client.start_training(Request::new(training_request)).await; - assert!(training_response.is_ok()); - - let training_info = training_response.unwrap().into_inner(); - assert!(training_info.success); - assert!(!training_info.job_id.is_empty()); - assert!(training_info.estimated_duration_minutes > 0); - - // Monitor training progress - let progress_request = GetTrainingStatusRequest { - job_id: training_info.job_id.clone(), - }; - - let mut training_completed = false; - let mut status_checks = 0; - - while !training_completed && status_checks < 20 { - tokio::time::sleep(Duration::from_millis(500)).await; - - let status_response = ml_client.get_training_status(Request::new(progress_request.clone())).await; - assert!(status_response.is_ok()); - - let status = status_response.unwrap().into_inner(); - assert_eq!(status.job_id, training_info.job_id); - assert!(status.progress_percent >= 0.0 && status.progress_percent <= 100.0); - - match TrainingStatus::from(status.status) { - TrainingStatus::Completed => { - training_completed = true; - assert_eq!(status.progress_percent, 100.0); - assert!(status.final_metrics.is_some()); - - let metrics = status.final_metrics.unwrap(); - assert!(metrics.loss >= 0.0); - assert!(metrics.accuracy >= 0.0 && metrics.accuracy <= 1.0); - } - TrainingStatus::Failed => { - panic!("Training failed: {}", status.error_message.unwrap_or("Unknown error".to_string())); - } - TrainingStatus::Running | TrainingStatus::Queued => { - // Continue monitoring - } - _ => {} - } - - status_checks += 1; - } - - // Get trained model info - if training_completed { - let model_request = GetModelInfoRequest { - model_name: "DQN_EURUSD_v1".to_string(), - }; - - let model_response = ml_client.get_model_info(Request::new(model_request)).await; - assert!(model_response.is_ok()); - - let model_info = model_response.unwrap().into_inner(); - assert_eq!(model_info.model_name, "DQN_EURUSD_v1"); - assert_eq!(model_info.model_type, MLModelType::DQN as i32); - assert!(model_info.training_completed_at > 0); - assert!(model_info.model_size_bytes > 0); - } - } - - #[tokio::test] - async fn test_ml_service_model_deployment() { - let mut ml_client = MockMLTrainingServiceClient::new().await.expect("Failed to create ML client"); - - // Deploy a trained model - let deployment_request = DeployModelRequest { - model_name: "DQN_EURUSD_v1".to_string(), - deployment_environment: "production".to_string(), - scaling_config: Some(ScalingConfig { - min_replicas: 1, - max_replicas: 3, - target_cpu_utilization: 70.0, - target_memory_utilization: 80.0, - }), - model_serving_config: Some(ModelServingConfig { - batch_prediction: true, - max_batch_size: 1000, - timeout_seconds: 30, - enable_caching: true, - }), - }; - - let deployment_response = ml_client.deploy_model(Request::new(deployment_request)).await; - assert!(deployment_response.is_ok()); - - let deployment_info = deployment_response.unwrap().into_inner(); - assert!(deployment_info.success); - assert!(!deployment_info.deployment_id.is_empty()); - assert!(!deployment_info.endpoint_url.is_empty()); - - // Test model predictions - let prediction_request = GetPredictionRequest { - model_name: "DQN_EURUSD_v1".to_string(), - features: Some(MLFeatures { - feature_values: vec![1.2345, 0.001, 0.15], // price_returns, volume, volatility - feature_names: vec![ - "price_returns".to_string(), - "volume".to_string(), - "volatility".to_string(), - ], - }), - prediction_type: PredictionType::SingleSample as i32, - }; - - let prediction_response = ml_client.get_prediction(Request::new(prediction_request)).await; - assert!(prediction_response.is_ok()); - - let prediction = prediction_response.unwrap().into_inner(); - assert!(prediction.success); - assert!(!prediction.prediction_values.is_empty()); - assert!(prediction.confidence_score >= 0.0 && prediction.confidence_score <= 1.0); - assert!(prediction.latency_ms > 0.0); - } - - // ======================================================================== - // Cross-Service Integration Tests - // ======================================================================== - - #[tokio::test] - async fn test_end_to_end_trading_workflow() { - // Initialize all service clients - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - let mut backtesting_client = MockBacktestingServiceClient::new().await.expect("Failed to create backtesting client"); - let mut ml_client = MockMLTrainingServiceClient::new().await.expect("Failed to create ML client"); - - // Step 1: Backtest a strategy - let backtest_request = StartBacktestRequest { - strategy_name: "MLEnhancedMomentum".to_string(), - symbols: vec!["EURUSD".to_string()], - start_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(30)).timestamp_nanos_opt().unwrap_or(0), - end_date_unix_nanos: (chrono::Utc::now() - chrono::Duration::days(1)).timestamp_nanos_opt().unwrap_or(0), - initial_capital: 100000.0, - parameters: { - let mut params = HashMap::new(); - params.insert("ml_model".to_string(), "DQN_EURUSD_v1".to_string()); - params.insert("confidence_threshold".to_string(), "0.7".to_string()); - params - }, - }; - - let backtest_response = backtesting_client.start_backtest(Request::new(backtest_request)).await; - assert!(backtest_response.is_ok()); - - let backtest_info = backtest_response.unwrap().into_inner(); - assert!(backtest_info.success); - - // Step 2: Wait for backtest completion (simplified for testing) - tokio::time::sleep(Duration::from_millis(1000)).await; - - let status_request = GetBacktestStatusRequest { - backtest_id: backtest_info.backtest_id.clone(), - }; - - let status_response = backtesting_client.get_backtest_status(Request::new(status_request)).await; - assert!(status_response.is_ok()); - - // Step 3: Get ML prediction for trading decision - let prediction_request = GetPredictionRequest { - model_name: "DQN_EURUSD_v1".to_string(), - features: Some(MLFeatures { - feature_values: vec![1.2345, 0.002, 0.12], - feature_names: vec![ - "price_returns".to_string(), - "volume".to_string(), - "volatility".to_string(), - ], - }), - prediction_type: PredictionType::SingleSample as i32, - }; - - let prediction_response = ml_client.get_prediction(Request::new(prediction_request)).await; - assert!(prediction_response.is_ok()); - - let prediction = prediction_response.unwrap().into_inner(); - assert!(prediction.success); - - // Step 4: Place order based on ML prediction (if confidence is high) - if prediction.confidence_score > 0.7 { - let order_request = PlaceOrderRequest { - symbol: "EURUSD".to_string(), - side: if prediction.prediction_values[0] > 0.5 { - OrderSide::Buy as i32 - } else { - OrderSide::Sell as i32 - }, - order_type: OrderType::Market as i32, - quantity: 10000.0, - price: None, - time_in_force: TimeInForce::ImmediateOrCancel as i32, - client_order_id: Uuid::new_v4().to_string(), - trader_id: "TRADER_ML_001".to_string(), - }; - - let order_response = trading_client.place_order(Request::new(order_request)).await; - assert!(order_response.is_ok()); - - let order_result = order_response.unwrap().into_inner(); - assert!(order_result.success); - assert!(!order_result.order_id.is_empty()); - - // Step 5: Monitor order execution - let monitor_request = GetOrderStatusRequest { - order_id: order_result.order_id.clone(), - }; - - let monitor_response = trading_client.get_order_status(Request::new(monitor_request)).await; - assert!(monitor_response.is_ok()); - - let order_status = monitor_response.unwrap().into_inner(); - assert_eq!(order_status.order_id, order_result.order_id); - assert!(matches!(OrderStatus::from(order_status.status), - OrderStatus::Pending | OrderStatus::Filled | OrderStatus::PartiallyFilled)); - } - } - - #[tokio::test] - async fn test_service_failure_recovery() { - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - - // Simulate service failure - let failure_request = SimulateFailureRequest { - service_type: ServiceType::Trading, - failure_type: FailureType::NetworkPartition, - duration_seconds: 5, - }; - - let failure_response = trading_client.simulate_failure(Request::new(failure_request)).await; - assert!(failure_response.is_ok()); - - // Attempt operations during failure - let order_request = PlaceOrderRequest { - symbol: "EURUSD".to_string(), - side: OrderSide::Buy as i32, - order_type: OrderType::Market as i32, - quantity: 10000.0, - price: None, - time_in_force: TimeInForce::ImmediateOrCancel as i32, - client_order_id: Uuid::new_v4().to_string(), - trader_id: "TRADER_001".to_string(), - }; - - // Should fail during simulated network partition - let failed_order_response = trading_client.place_order(Request::new(order_request.clone())).await; - assert!(failed_order_response.is_err()); - - let error = failed_order_response.err().unwrap(); - assert_eq!(error.code(), Code::Unavailable); - - // Wait for recovery - tokio::time::sleep(Duration::from_secs(6)).await; - - // Should succeed after recovery - let recovered_order_response = trading_client.place_order(Request::new(order_request)).await; - assert!(recovered_order_response.is_ok()); - - let recovered_result = recovered_order_response.unwrap().into_inner(); - assert!(recovered_result.success); - } - - #[tokio::test] - async fn test_load_balancing_and_circuit_breakers() { - let load_balancer = MockLoadBalancer::new(); - - // Register multiple service instances - let services = vec![ - ServiceEndpoint { - service_type: ServiceType::Trading, - address: "127.0.0.1".to_string(), - port: 50051, - health_check_path: "/health".to_string(), - metadata: HashMap::new(), - }, - ServiceEndpoint { - service_type: ServiceType::Trading, - address: "127.0.0.1".to_string(), - port: 50052, - health_check_path: "/health".to_string(), - metadata: HashMap::new(), - }, - ServiceEndpoint { - service_type: ServiceType::Trading, - address: "127.0.0.1".to_string(), - port: 50053, - health_check_path: "/health".to_string(), - metadata: HashMap::new(), - }, - ]; - - for service in services { - load_balancer.register_service(service).await.expect("Failed to register service"); - } - - // Test round-robin load balancing - let mut selected_ports = Vec::new(); - for _ in 0..6 { - let selected_service = load_balancer.select_service(ServiceType::Trading).await; - assert!(selected_service.is_ok()); - selected_ports.push(selected_service.unwrap().port); - } - - // Should cycle through all ports - assert!(selected_ports.contains(&50051)); - assert!(selected_ports.contains(&50052)); - assert!(selected_ports.contains(&50053)); - - // Simulate service failure and test circuit breaker - load_balancer.mark_service_unhealthy("127.0.0.1:50052".to_string()).await; - - let mut healthy_ports = Vec::new(); - for _ in 0..10 { - let selected_service = load_balancer.select_service(ServiceType::Trading).await; - assert!(selected_service.is_ok()); - healthy_ports.push(selected_service.unwrap().port); - } - - // Should not select the unhealthy service - assert!(!healthy_ports.contains(&50052)); - assert!(healthy_ports.contains(&50051)); - assert!(healthy_ports.contains(&50053)); - } - - // ======================================================================== - // Performance and Stress Tests - // ======================================================================== - - #[tokio::test] - async fn test_high_frequency_service_communication() { - let mut trading_client = MockTradingServiceClient::new().await.expect("Failed to create trading client"); - - let num_requests = 1000; - let start_time = Instant::now(); - let mut successful_requests = 0; - let mut failed_requests = 0; - - // Send high-frequency order status requests - let mut handles = Vec::new(); - - for i in 0..num_requests { - let mut client_clone = trading_client.clone(); - let handle = tokio::spawn(async move { - let status_request = GetOrderStatusRequest { - order_id: format!("ORDER_{:06}", i), - }; - - client_clone.get_order_status(Request::new(status_request)).await - }); - - handles.push(handle); - } - - // Wait for all requests to complete - for handle in handles { - match handle.await { - Ok(Ok(_)) => successful_requests += 1, - Ok(Err(_)) | Err(_) => failed_requests += 1, - } - } - - let elapsed = start_time.elapsed(); - let requests_per_second = num_requests as f64 / elapsed.as_secs_f64(); - - println!("High-frequency test results:"); - println!(" Total requests: {}", num_requests); - println!(" Successful: {}", successful_requests); - println!(" Failed: {}", failed_requests); - println!(" Duration: {:?}", elapsed); - println!(" Requests/second: {:.2}", requests_per_second); - - // Should handle at least 100 requests per second - assert!(requests_per_second > 100.0); - // Should have at least 95% success rate - assert!(successful_requests as f64 / num_requests as f64 > 0.95); - } - - #[tokio::test] - async fn test_concurrent_service_connections() { - let num_clients = 50; - let mut client_handles = Vec::new(); - - // Create multiple concurrent clients - for client_id in 0..num_clients { - let handle = tokio::spawn(async move { - let client_result = MockTradingServiceClient::new().await; - if client_result.is_err() { - return (client_id, false, String::new()); - } - - let mut client = client_result.unwrap(); - - // Each client performs a health check - let health_request = HealthCheckRequest { - service: ServiceType::Trading as i32, - }; - - match client.health_check(Request::new(health_request)).await { - Ok(response) => { - let health_info = response.into_inner(); - (client_id, health_info.status == ServiceStatus::Healthy as i32, health_info.version) - } - Err(e) => (client_id, false, format!("Error: {}", e)), - } - }); - - client_handles.push(handle); - } - - // Wait for all clients - let mut successful_connections = 0; - for handle in client_handles { - let (client_id, success, info) = handle.await.expect("Client task failed"); - if success { - successful_connections += 1; - } else { - println!("Client {} failed: {}", client_id, info); - } - } - - println!("Concurrent connection test:"); - println!(" Total clients: {}", num_clients); - println!(" Successful connections: {}", successful_connections); - println!(" Success rate: {:.2}%", (successful_connections as f64 / num_clients as f64) * 100.0); - - // Should handle at least 90% of concurrent connections successfully - assert!(successful_connections as f64 / num_clients as f64 > 0.90); - } -} - -// ============================================================================ -// Mock Service Implementations for Testing -// ============================================================================ - -// These mock implementations provide realistic service behavior for testing -// In production, these would be replaced with actual gRPC service implementations - -use std::sync::{atomic::AtomicBool, atomic::Ordering}; - -// Mock Trading Service -pub struct MockTradingServer { - health_status: Arc, - orders: Arc>>, -} - -impl MockTradingServer { - pub async fn new() -> Result { - Ok(Self { - health_status: Arc::new(AtomicBool::new(true)), - orders: Arc::new(RwLock::new(HashMap::new())), - }) - } - - pub async fn health_check(&self) -> Result { - Ok(HealthCheckResponse { - status: if self.health_status.load(Ordering::Relaxed) { - ServiceStatus::Healthy as i32 - } else { - ServiceStatus::Unhealthy as i32 - }, - version: "1.0.0".to_string(), - uptime_seconds: 3600, // 1 hour - message: "Service is operational".to_string(), - }) - } -} - -#[derive(Debug, Clone)] -pub struct MockOrder { - pub order_id: String, - pub symbol: String, - pub side: OrderSide, - pub quantity: f64, - pub price: Option, - pub status: OrderStatus, - pub created_at: chrono::DateTime, -} - -// Mock service client implementations -#[derive(Clone)] -pub struct MockTradingServiceClient { - // In a real implementation, this would contain the gRPC client -} - -impl MockTradingServiceClient { - pub async fn new() -> Result { - Ok(Self {}) - } - - pub async fn place_order(&mut self, request: Request) -> Result, Status> { - let req = request.into_inner(); - - // Simulate order validation and placement - if req.quantity <= 0.0 { - return Ok(Response::new(PlaceOrderResponse { - success: false, - order_id: String::new(), - message: "Order placed successfully".to_string(), - error_message: "Invalid quantity".to_string(), - estimated_fill_time_ms: 0, - })); - } - - let order_id = Uuid::new_v4().to_string(); - - Ok(Response::new(PlaceOrderResponse { - success: true, - order_id, - message: "Order placed successfully".to_string(), - error_message: String::new(), - estimated_fill_time_ms: 50, - })) - } - - pub async fn get_order_status(&mut self, request: Request) -> Result, Status> { - let req = request.into_inner(); - - Ok(Response::new(GetOrderStatusResponse { - order_id: req.order_id, - status: OrderStatus::Pending as i32, - symbol: "EURUSD".to_string(), - side: OrderSide::Buy as i32, - quantity: 10000.0, - filled_quantity: 0.0, - average_fill_price: 0.0, - created_at_unix_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - updated_at_unix_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - })) - } - - pub async fn cancel_order(&mut self, request: Request) -> Result, Status> { - let req = request.into_inner(); - - Ok(Response::new(CancelOrderResponse { - success: true, - order_id: req.order_id, - message: "Order cancelled successfully".to_string(), - error_message: String::new(), - })) - } - - pub async fn check_position_limits(&mut self, request: Request) -> Result, Status> { - let req = request.into_inner(); - - // Simulate position limit checking - let approved = req.quantity < 50000.0; // Reject large positions - let mut violations = Vec::new(); - - if !approved { - violations.push("position_limit_exceeded".to_string()); - } - - Ok(Response::new(PositionLimitCheckResponse { - approved, - violations, - current_position: 25000.0, - position_limit: 50000.0, - margin_required: req.quantity * 0.02, // 2% margin - margin_available: 100000.0, - })) - } - - pub async fn get_risk_metrics(&mut self, request: Request) -> Result, Status> { - let _req = request.into_inner(); - - Ok(Response::new(GetRiskMetricsResponse { - var_1_day: 2500.0, - var_10_day: 7500.0, - maximum_drawdown: 0.05, - sharpe_ratio: 1.8, - sortino_ratio: 2.1, - calmar_ratio: 3.2, - portfolio_value: 100000.0, - unrealized_pnl: 1500.0, - realized_pnl: 2800.0, - })) - } - - pub async fn subscribe_market_data(&mut self, request: Request) -> Result>>, Status> { - let req = request.into_inner(); - - // Create a mock market data stream - let (tx, rx) = tokio::sync::mpsc::channel(100); - - // Spawn a task to generate mock market data - let symbols = req.symbols.clone(); - tokio::spawn(async move { - let mut counter = 0; - loop { - for symbol in &symbols { - // Generate mock tick data - let tick_data = MarketDataEvent { - symbol: symbol.clone(), - data_type: MarketDataType::Tick as i32, - bid: 1.2345 + (counter as f64 * 0.0001), - ask: 1.2347 + (counter as f64 * 0.0001), - timestamp_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - volume: 1000.0, - order_book_levels: Vec::new(), - }; - - if tx.send(Ok(tick_data)).await.is_err() { - return; // Client disconnected - } - - // Generate mock order book data - let orderbook_data = MarketDataEvent { - symbol: symbol.clone(), - data_type: MarketDataType::OrderBook as i32, - bid: 1.2345, - ask: 1.2347, - timestamp_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - volume: 0.0, - order_book_levels: vec![ - OrderBookLevel { price: 1.2345, size: 10000.0, side: OrderSide::Buy as i32 }, - OrderBookLevel { price: 1.2347, size: 15000.0, side: OrderSide::Sell as i32 }, - ], - }; - - if tx.send(Ok(orderbook_data)).await.is_err() { - return; // Client disconnected - } - } - - counter += 1; - tokio::time::sleep(Duration::from_millis(100)).await; - } - }); - - let stream = ReceiverStream::new(rx); - Ok(Response::new(stream)) - } - - pub async fn health_check(&mut self, request: Request) -> Result, Status> { - let _req = request.into_inner(); - - Ok(Response::new(HealthCheckResponse { - status: ServiceStatus::Healthy as i32, - version: "1.0.0".to_string(), - uptime_seconds: 3600, - message: "Trading service is healthy".to_string(), - })) - } - - pub async fn simulate_failure(&mut self, request: Request) -> Result, Status> { - let req = request.into_inner(); - - Ok(Response::new(SimulateFailureResponse { - success: true, - message: format!("Simulating {:?} failure for {} seconds", req.failure_type(), req.duration_seconds), - })) - } -} - -// Additional mock implementations for other services would follow similar patterns... -// This demonstrates the comprehensive approach needed for 95%+ integration test coverage - -// Mock types and enums -#[derive(Debug, Clone, PartialEq)] -pub enum ServiceType { - Trading, - Backtesting, - MLTraining, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum ServiceStatus { - Healthy, - Unhealthy, - Degraded, -} - -// OrderSide, OrderType, and OrderStatus already imported via trading_engine::prelude::* and common::types::* (lines 17-18) - -// REMOVED: TimeInForce duplicate - use common::TimeInForce -// Note: GTD variant not supported in canonical definition - -#[derive(Debug, Clone, PartialEq)] -pub enum BacktestStatus { - Queued, - Running, - Completed, - Failed, - Cancelled, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum MarketDataType { - Tick, - OrderBook, - Trade, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum MLModelType { - DQN, - PPO, - MAMBA, - TFT, - TLOB, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum TrainingStatus { - Queued, - Running, - Completed, - Failed, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum PredictionType { - SingleSample, - Batch, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum Permission { - Trading, - RiskManagement, - MarketData, - BacktestingRead, - BacktestingWrite, - MLModelRead, - MLModelWrite, - ConfigRead, - ConfigWrite, - AdminAccess, -} - -#[derive(Debug, Clone, PartialEq)] -pub enum FailureType { - NetworkPartition, - ServiceCrash, - DatabaseFailure, - HighLatency, -} - -// Mock request/response types -#[derive(Debug, Clone)] -pub struct PlaceOrderRequest { - pub symbol: String, - pub side: i32, - pub order_type: i32, - pub quantity: f64, - pub price: Option, - pub time_in_force: i32, - pub client_order_id: String, - pub trader_id: String, -} - -#[derive(Debug, Clone)] -pub struct PlaceOrderResponse { - pub success: bool, - pub order_id: String, - pub message: String, - pub error_message: String, - pub estimated_fill_time_ms: u64, -} - -// Additional mock request/response types would be defined here... -// This demonstrates the comprehensive testing framework needed for service integration - -// Continue with remaining mock implementations for complete test coverage -// This establishes the foundation for achieving 95%+ integration test coverage \ No newline at end of file