use anyhow::Result; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::process::Stdio; use tokio::process::Command; use tokio::io::{AsyncBufReadExt, BufReader}; use tracing::{info, warn, error, debug}; /// Corrode-MCP integration for E2E test execution pub struct CorrodeTestRunner { config: CorrodeConfig, active_sessions: HashMap, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct CorrodeConfig { pub executable_path: String, pub workspace_path: String, pub timeout_seconds: u64, pub max_parallel_sessions: usize, pub log_level: String, } impl Default for CorrodeConfig { fn default() -> Self { Self { executable_path: "corrode".to_string(), workspace_path: "/home/jgrusewski/Work/foxhunt".to_string(), timeout_seconds: 300, max_parallel_sessions: 4, log_level: "info".to_string(), } } } #[derive(Debug)] pub struct CorrodeSession { pub id: String, pub process: tokio::process::Child, pub start_time: tokio::time::Instant, pub test_name: String, } #[derive(Debug, Serialize, Deserialize)] pub struct TestExecutionRequest { pub test_name: String, pub test_command: String, pub environment: HashMap, pub working_directory: Option, pub timeout_seconds: Option, pub capture_output: bool, } #[derive(Debug, Serialize, Deserialize)] pub struct TestExecutionResult { pub test_name: String, pub success: bool, pub exit_code: Option, pub stdout: String, pub stderr: String, pub execution_time: tokio::time::Duration, pub error_message: Option, } impl CorrodeTestRunner { pub fn new(config: CorrodeConfig) -> Self { Self { config, active_sessions: HashMap::new(), } } pub fn with_default_config() -> Self { Self::new(CorrodeConfig::default()) } /// Execute a single test via corrode-mcp pub async fn execute_test(&mut self, request: TestExecutionRequest) -> Result { info!("Executing test: {}", request.test_name); if self.active_sessions.len() >= self.config.max_parallel_sessions { return Err(anyhow::anyhow!("Max parallel sessions reached")); } let session_id = uuid::Uuid::new_v4().to_string(); let start_time = tokio::time::Instant::now(); // Prepare environment variables let mut env_vars = self.prepare_environment(&request)?; // Add Foxhunt-specific environment variables env_vars.insert("FOXHUNT_TEST_MODE".to_string(), "true".to_string()); env_vars.insert("DATABASE_URL".to_string(), "postgresql://localhost/foxhunt_test".to_string()); env_vars.insert("RUST_LOG".to_string(), self.config.log_level.clone()); // Build corrode command let mut command = Command::new(&self.config.executable_path); command .current_dir(request.working_directory.as_ref().unwrap_or(&self.config.workspace_path)) .args(&["execute", "--test", &request.test_command]) .envs(&env_vars); if request.capture_output { command.stdout(Stdio::piped()).stderr(Stdio::piped()); } debug!("Starting corrode process: {:?}", command); let mut process = command.spawn().map_err(|e| { error!("Failed to spawn corrode process: {}", e); anyhow::anyhow!("Failed to spawn corrode process: {}", e) })?; let session = CorrodeSession { id: session_id.clone(), process, start_time, test_name: request.test_name.clone(), }; self.active_sessions.insert(session_id.clone(), session); // Wait for completion with timeout let timeout = tokio::time::Duration::from_secs( request.timeout_seconds.unwrap_or(self.config.timeout_seconds) ); let result = tokio::time::timeout(timeout, self.wait_for_completion(&session_id)).await; // Clean up session self.cleanup_session(&session_id).await?; match result { Ok(execution_result) => Ok(execution_result?), Err(_) => { warn!("Test execution timed out: {}", request.test_name); Ok(TestExecutionResult { test_name: request.test_name, success: false, exit_code: None, stdout: String::new(), stderr: "Test execution timed out".to_string(), execution_time: timeout, error_message: Some("Timeout exceeded".to_string()), }) } } } /// Execute multiple tests in parallel pub async fn execute_test_suite(&mut self, requests: Vec) -> Result> { info!("Executing test suite with {} tests", requests.len()); let mut results = Vec::new(); let mut tasks = Vec::new(); // Split requests into batches to respect max parallel sessions for batch in requests.chunks(self.config.max_parallel_sessions) { let mut batch_tasks = Vec::new(); for request in batch { let mut runner = CorrodeTestRunner::new(self.config.clone()); let request_clone = request.clone(); let task = tokio::spawn(async move { runner.execute_test(request_clone).await }); batch_tasks.push(task); } // Wait for current batch to complete for task in batch_tasks { match task.await { Ok(result) => results.push(result?), Err(e) => { error!("Task execution error: {}", e); return Err(anyhow::anyhow!("Task execution failed: {}", e)); } } } } Ok(results) } /// Execute E2E trading workflow test pub async fn execute_trading_workflow_test(&mut self) -> Result { let request = TestExecutionRequest { test_name: "e2e_trading_workflow".to_string(), test_command: "cargo test --package foxhunt-e2e --test trading_workflow -- --nocapture".to_string(), environment: self.create_trading_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(600), // 10 minutes for full trading workflow capture_output: true, }; self.execute_test(request).await } /// Execute ML pipeline test pub async fn execute_ml_pipeline_test(&mut self) -> Result { let request = TestExecutionRequest { test_name: "e2e_ml_pipeline".to_string(), test_command: "cargo test --package foxhunt-e2e --test ml_pipeline -- --nocapture".to_string(), environment: self.create_ml_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(900), // 15 minutes for ML training/inference capture_output: true, }; self.execute_test(request).await } /// Execute database integration test pub async fn execute_database_test(&mut self) -> Result { let request = TestExecutionRequest { test_name: "e2e_database_integration".to_string(), test_command: "cargo test --package foxhunt-e2e --test database -- --nocapture".to_string(), environment: self.create_database_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(300), // 5 minutes for database tests capture_output: true, }; self.execute_test(request).await } /// Execute complete E2E test suite pub async fn execute_full_e2e_suite(&mut self) -> Result> { info!("Starting complete E2E test suite execution"); let requests = vec![ // Service startup tests TestExecutionRequest { test_name: "service_startup".to_string(), test_command: "cargo test --package foxhunt-e2e test_service_startup".to_string(), environment: self.create_service_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(120), capture_output: true, }, // Database integration TestExecutionRequest { test_name: "database_integration".to_string(), test_command: "cargo test --package foxhunt-e2e test_database_integration".to_string(), environment: self.create_database_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(300), capture_output: true, }, // gRPC client tests TestExecutionRequest { test_name: "grpc_clients".to_string(), test_command: "cargo test --package foxhunt-e2e test_grpc_clients".to_string(), environment: self.create_service_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(180), capture_output: true, }, // ML pipeline tests TestExecutionRequest { test_name: "ml_pipeline".to_string(), test_command: "cargo test --package foxhunt-e2e test_ml_pipeline".to_string(), environment: self.create_ml_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(900), capture_output: true, }, // Trading workflow tests TestExecutionRequest { test_name: "trading_workflows".to_string(), test_command: "cargo test --package foxhunt-e2e test_trading_workflows".to_string(), environment: self.create_trading_test_environment()?, working_directory: Some(self.config.workspace_path.clone()), timeout_seconds: Some(600), capture_output: true, }, ]; self.execute_test_suite(requests).await } /// Generate comprehensive test report pub async fn generate_test_report(&self, results: &[TestExecutionResult]) -> Result { let total_tests = results.len(); let passed_tests = results.iter().filter(|r| r.success).count(); let failed_tests = total_tests - passed_tests; let total_execution_time: tokio::time::Duration = results.iter() .map(|r| r.execution_time) .sum(); let mut report = String::new(); report.push_str("# Foxhunt E2E Test Report\n\n"); report.push_str(&format!("Generated: {}\n", chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC"))); report.push_str(&format!("Total Tests: {}\n", total_tests)); report.push_str(&format!("Passed: {}\n", passed_tests)); report.push_str(&format!("Failed: {}\n", failed_tests)); report.push_str(&format!("Success Rate: {:.1}%\n", (passed_tests as f64 / total_tests as f64) * 100.0)); report.push_str(&format!("Total Execution Time: {:?}\n\n", total_execution_time)); report.push_str("## Test Results\n\n"); for result in results { let status = if result.success { "✅ PASS" } else { "❌ FAIL" }; report.push_str(&format!("### {} - {}\n", status, result.test_name)); report.push_str(&format!("- Execution Time: {:?}\n", result.execution_time)); if let Some(exit_code) = result.exit_code { report.push_str(&format!("- Exit Code: {}\n", exit_code)); } if !result.success { if let Some(error) = &result.error_message { report.push_str(&format!("- Error: {}\n", error)); } if !result.stderr.is_empty() { report.push_str("- Stderr:\n```\n"); report.push_str(&result.stderr); report.push_str("\n```\n"); } } report.push_str("\n"); } Ok(report) } // Private helper methods async fn wait_for_completion(&mut self, session_id: &str) -> Result { let session = self.active_sessions.get_mut(session_id) .ok_or_else(|| anyhow::anyhow!("Session not found: {}", session_id))?; let start_time = session.start_time; let test_name = session.test_name.clone(); let output = session.process.wait_with_output().await?; let execution_time = start_time.elapsed(); let result = TestExecutionResult { test_name, success: output.status.success(), exit_code: output.status.code(), stdout: String::from_utf8_lossy(&output.stdout).to_string(), stderr: String::from_utf8_lossy(&output.stderr).to_string(), execution_time, error_message: if !output.status.success() { Some(format!("Process exited with code: {:?}", output.status.code())) } else { None }, }; Ok(result) } async fn cleanup_session(&mut self, session_id: &str) -> Result<()> { if let Some(mut session) = self.active_sessions.remove(session_id) { // Ensure process is terminated if let Err(e) = session.process.kill().await { warn!("Failed to kill process for session {}: {}", session_id, e); } } Ok(()) } fn prepare_environment(&self, request: &TestExecutionRequest) -> Result> { let mut env = request.environment.clone(); // Add standard test environment variables env.insert("RUST_BACKTRACE".to_string(), "1".to_string()); env.insert("RUST_LOG".to_string(), self.config.log_level.clone()); Ok(env) } fn create_service_test_environment(&self) -> Result> { let mut env = HashMap::new(); env.insert("TRADING_SERVICE_PORT".to_string(), "50051".to_string()); env.insert("BACKTESTING_SERVICE_PORT".to_string(), "50052".to_string()); env.insert("ML_TRAINING_SERVICE_PORT".to_string(), "50053".to_string()); env.insert("DATABASE_URL".to_string(), "postgresql://localhost/foxhunt_test".to_string()); Ok(env) } fn create_database_test_environment(&self) -> Result> { let mut env = HashMap::new(); env.insert("DATABASE_URL".to_string(), "postgresql://localhost/foxhunt_test".to_string()); env.insert("PGUSER".to_string(), "postgres".to_string()); env.insert("PGPASSWORD".to_string(), "postgres".to_string()); env.insert("PGDATABASE".to_string(), "foxhunt_test".to_string()); Ok(env) } fn create_ml_test_environment(&self) -> Result> { let mut env = HashMap::new(); env.insert("ML_MODEL_PATH".to_string(), "/tmp/foxhunt_test_models".to_string()); env.insert("CUDA_VISIBLE_DEVICES".to_string(), "0".to_string()); env.insert("TORCH_DEVICE".to_string(), "cpu".to_string()); // Use CPU for tests Ok(env) } fn create_trading_test_environment(&self) -> Result> { let mut env = HashMap::new(); env.insert("TRADING_MODE".to_string(), "simulation".to_string()); env.insert("BROKER_ENDPOINT".to_string(), "simulation://localhost".to_string()); env.insert("RISK_CHECK_ENABLED".to_string(), "true".to_string()); Ok(env) } } impl Clone for TestExecutionRequest { fn clone(&self) -> Self { Self { test_name: self.test_name.clone(), test_command: self.test_command.clone(), environment: self.environment.clone(), working_directory: self.working_directory.clone(), timeout_seconds: self.timeout_seconds, capture_output: self.capture_output, } } } #[cfg(test)] mod tests { use super::*; #[tokio::test] async fn test_corrode_config() { let config = CorrodeConfig::default(); assert!(!config.executable_path.is_empty()); assert!(config.timeout_seconds > 0); } #[test] fn test_environment_preparation() { let runner = CorrodeTestRunner::with_default_config(); let request = TestExecutionRequest { test_name: "test".to_string(), test_command: "echo test".to_string(), environment: HashMap::new(), working_directory: None, timeout_seconds: None, capture_output: true, }; let env = runner.prepare_environment(&request).unwrap(); assert!(env.contains_key("RUST_BACKTRACE")); assert!(env.contains_key("RUST_LOG")); } }