//! E2E Test Framework Core //! //! Provides the main E2ETestFramework struct that orchestrates all testing components //! including service management, gRPC clients, database connections, and ML pipeline testing. use anyhow::{Context, Result}; use std::time::Duration; use tokio::time::sleep; use tracing::{debug, info, warn}; use crate::proto::backtesting::backtesting_service_client::BacktestingServiceClient; use crate::proto::config::config_service_client::ConfigServiceClient; use crate::proto::trading::trading_service_client::TradingServiceClient; use crate::{ database::TestDatabase, ml_pipeline::MLPipelineTestHarness, performance::PerformanceTracker, services::ServiceManager, }; use tonic::metadata::AsciiMetadataValue; use tonic::service::{interceptor::InterceptedService, Interceptor}; use tonic::transport::Channel; /// JWT Authentication Interceptor for E2E Tests /// /// Automatically injects JWT token into all gRPC requests #[derive(Clone)] pub struct AuthInterceptor { token: AsciiMetadataValue, } impl AuthInterceptor { fn new(token: &str) -> Result { let bearer_token = format!("Bearer {}", token); let token = AsciiMetadataValue::try_from(bearer_token) .context("Failed to create metadata value from token")?; Ok(Self { token }) } } impl Interceptor for AuthInterceptor { fn call( &mut self, mut request: tonic::Request<()>, ) -> Result, tonic::Status> { request .metadata_mut() .insert("authorization", self.token.clone()); Ok(request) } } /// Main E2E Test Framework /// /// Provides centralized orchestration for all testing components: /// - Service lifecycle management /// /// - gRPC client connections (via API Gateway with JWT auth) /// - Database testing harness /// /// - ML pipeline testing /// - Performance monitoring /// /// - Test data management #[derive(Debug)] pub struct E2ETestFramework { pub service_manager: ServiceManager, pub database_harness: TestDatabase, pub ml_pipeline: MLPipelineTestHarness, pub performance_tracker: PerformanceTracker, // gRPC clients (initialized on demand, connect via API Gateway with auth interceptor) pub trading_client: Option>>, pub backtesting_client: Option>>, pub config_client: Option>>, // Authentication token for E2E tests pub auth_token: String, // Framework state pub services_started: bool, pub test_session_id: String, } impl E2ETestFramework { /// Generate a test JWT token for E2E authentication /// /// Creates a valid JWT token with test user credentials. /// /// This token is accepted by API Gateway for test scenarios. fn generate_test_jwt_token() -> Result { // Load .env file if present (development mode) // Silent failure allows CI/CD to override with environment variables let _ = dotenvy::dotenv(); use chrono::Utc; use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; use serde::{Deserialize, Serialize}; #[derive(Debug, Serialize, Deserialize)] struct Claims { sub: String, // user_id exp: usize, // expiration iat: usize, // issued at iss: String, // issuer aud: String, // audience jti: String, // JWT ID roles: Vec, permissions: Vec, } let now = Utc::now().timestamp() as usize; let claims = Claims { sub: "e2e_test_user".to_string(), exp: now + 3600, // 1 hour expiration iat: now, iss: "foxhunt-api".to_string(), aud: "foxhunt-services".to_string(), jti: uuid::Uuid::new_v4().to_string(), roles: vec!["trader".to_string(), "admin".to_string()], permissions: vec![ "api.access".to_string(), "trading.submit".to_string(), "trading.cancel".to_string(), "backtesting.run".to_string(), "ml.train".to_string(), ], }; // Load JWT secret from environment (loaded from .env or CI/CD) // CRITICAL: Must match API Gateway JWT_SECRET configuration let secret = std::env::var("JWT_SECRET").context( "JWT_SECRET not configured. Options:\n \ 1. Create .env file with JWT_SECRET (development) - AUTOMATIC\n \ 2. Export JWT_SECRET environment variable (CI/CD)\n \ 3. Verify .env file exists in project root", )?; // Validate secret length (security requirement) if secret.len() < 64 { anyhow::bail!( "JWT_SECRET must be at least 64 characters (current: {}). \n\ Generate a secure secret: openssl rand -base64 64", secret.len() ); } let token = encode( &Header::new(Algorithm::HS256), &claims, &EncodingKey::from_secret(secret.as_bytes()), ) .context("Failed to encode JWT token")?; Ok(token) } /// Create a new E2E test framework instance /// /// This initializes all components but does not start services. /// /// Call `start_services()` to begin service orchestration. pub async fn new() -> Result { info!("🔧 Initializing E2E Test Framework..."); let test_session_id = format!( "test_session_{}", chrono::Utc::now().format("%Y%m%d_%H%M%S") ); debug!("Generated test session ID: {}", test_session_id); // Generate JWT token for E2E testing let auth_token = Self::generate_test_jwt_token().context("Failed to generate test JWT token")?; debug!("Generated E2E test JWT token"); // Initialize database harness let database_harness = TestDatabase::new("postgresql://localhost/foxhunt_test".to_string()); // Initialize ML pipeline testing let ml_pipeline = MLPipelineTestHarness::new() .await .context("Failed to initialize ML pipeline test harness")?; // Initialize service manager let service_manager = ServiceManager::new(); // Initialize performance tracker let performance_tracker = PerformanceTracker::new(&test_session_id) .context("Failed to initialize performance tracker")?; info!("✅ E2E Test Framework initialized successfully"); Ok(Self { service_manager, database_harness, ml_pipeline, performance_tracker, trading_client: None, backtesting_client: None, config_client: None, auth_token, services_started: false, test_session_id, }) } /// Start all services needed for testing pub async fn start_services(&mut self) -> Result<()> { if self.services_started { debug!("Services already started, skipping startup"); return Ok(()); } info!("🚀 Starting services for E2E testing..."); // Start services in proper order self.service_manager .start_all_services() .await .context("Failed to start services")?; // Wait for services to be ready info!("⏳ Waiting for services to be ready..."); self.wait_for_services_ready() .await .context("Services failed to become ready")?; self.services_started = true; info!("✅ All services started and ready"); Ok(()) } /// Stop all services and cleanup pub async fn stop_services(&mut self) -> Result<()> { if !self.services_started { debug!("Services not started, skipping shutdown"); return Ok(()); } info!("🛑 Stopping services..."); // Close client connections first self.trading_client = None; self.backtesting_client = None; self.config_client = None; // Stop services self.service_manager .stop_all_services() .await .context("Failed to stop services")?; self.services_started = false; info!("✅ All services stopped"); Ok(()) } /// Get Trading Service gRPC client (via API Gateway with JWT auth) pub async fn get_trading_client( &mut self, ) -> Result<&mut TradingServiceClient>> { if self.trading_client.is_none() { info!("🔌 Connecting to Trading Service via API Gateway (port 50051)..."); // Create authenticated channel via interceptor let channel = Channel::from_static("http://[::1]:50051") .connect() .await .context("Failed to connect to API Gateway")?; let interceptor = AuthInterceptor::new(&self.auth_token) .context("Failed to create auth interceptor")?; let client = TradingServiceClient::with_interceptor(channel, interceptor); self.trading_client = Some(client); info!("✅ Connected to Trading Service via API Gateway with JWT auth"); } Ok(self.trading_client.as_mut().unwrap()) } /// Get Backtesting Service gRPC client (via API Gateway with JWT auth) pub async fn get_backtesting_client( &mut self, ) -> Result<&mut BacktestingServiceClient>> { if self.backtesting_client.is_none() { info!("🔌 Connecting to Backtesting Service via API Gateway (port 50051)..."); // Create authenticated channel via interceptor let channel = Channel::from_static("http://[::1]:50051") .connect() .await .context("Failed to connect to API Gateway")?; let interceptor = AuthInterceptor::new(&self.auth_token) .context("Failed to create auth interceptor")?; let client = BacktestingServiceClient::with_interceptor(channel, interceptor); self.backtesting_client = Some(client); info!("✅ Connected to Backtesting Service via API Gateway with JWT auth"); } Ok(self.backtesting_client.as_mut().unwrap()) } /// Get Configuration Service client (via API Gateway with JWT auth) pub async fn get_config_client( &mut self, ) -> Result<&mut ConfigServiceClient>> { if self.config_client.is_none() { info!("🔌 Connecting to Configuration Service via API Gateway (port 50051)..."); // Create authenticated channel via interceptor let channel = Channel::from_static("http://[::1]:50051") .connect() .await .context("Failed to connect to API Gateway")?; let interceptor = AuthInterceptor::new(&self.auth_token) .context("Failed to create auth interceptor")?; let client = ConfigServiceClient::with_interceptor(channel, interceptor); self.config_client = Some(client); info!("✅ Connected to Configuration Service via API Gateway with JWT auth"); } Ok(self.config_client.as_mut().unwrap()) } /// Check health status of all services pub async fn check_services_health(&self) -> Result { info!("🩺 Checking services health..."); let mut status = ServicesHealthStatus { trading_service: ServiceHealth::Unknown, config_service: ServiceHealth::Unknown, database: ServiceHealth::Unknown, all_healthy: false, }; // Check Trading Service status.trading_service = match self.check_trading_service_health().await { Ok(_) => ServiceHealth::Healthy, Err(e) => { warn!("Trading Service health check failed: {}", e); ServiceHealth::Unhealthy }, }; // Note: Backtesting Service health check removed - service not available yet // Check Database (simplified for now) status.database = ServiceHealth::Healthy; // Configuration service is database-backed, so use database status status.config_service = status.database.clone(); // All services are healthy if no service is unhealthy status.all_healthy = ![ &status.trading_service, &status.config_service, &status.database, ] .iter() .any(|s| matches!(s, ServiceHealth::Unhealthy)); if status.all_healthy { info!("✅ All services are healthy"); } else { warn!("⚠️ Some services are not healthy: {:#?}", status); } Ok(status) } /// Wait for all services to be ready with retries async fn wait_for_services_ready(&self) -> Result<()> { let max_retries = 30; // 30 attempts let retry_delay = Duration::from_secs(2); // 2 seconds between attempts for attempt in 1..=max_retries { debug!("Health check attempt {}/{}", attempt, max_retries); let health = self .check_services_health() .await .context("Failed to check services health")?; if health.all_healthy { info!("✅ All services are ready after {} attempts", attempt); return Ok(()); } if attempt < max_retries { debug!("Some services not ready, retrying in {:?}...", retry_delay); sleep(retry_delay).await; } } Err(anyhow::anyhow!( "Services failed to become ready after {} attempts", max_retries )) } /// Check Trading Service health (via API Gateway) async fn check_trading_service_health(&self) -> Result<()> { // Simple TCP connection check to API Gateway use tokio::net::TcpStream; let _stream = TcpStream::connect("127.0.0.1:50051") .await .context("Could not connect to API Gateway port 50051")?; Ok(()) } // Note: Backtesting Service health check removed - service not available yet /// Get the test session ID pub fn get_test_session_id(&self) -> &str { &self.test_session_id } /// Setup test database pub async fn setup_test_database(&self) -> Result<()> { self.database_harness.setup().await } /// Teardown test database pub async fn teardown_test_database(&self) -> Result<()> { self.database_harness.teardown().await } } /// Service health status enumeration #[derive(Debug, Clone, PartialEq)] pub enum ServiceHealth { Healthy, Unhealthy, Unknown, } /// Overall services health status #[derive(Debug, Clone)] pub struct ServicesHealthStatus { pub trading_service: ServiceHealth, pub config_service: ServiceHealth, pub database: ServiceHealth, pub all_healthy: bool, } impl ServicesHealthStatus { /// Get a summary of health status as a string pub fn summary(&self) -> String { let services = [ ("Trading", &self.trading_service), ("Config", &self.config_service), ("Database", &self.database), ]; let healthy_count = services .iter() .filter(|(_, health)| matches!(health, ServiceHealth::Healthy)) .count(); let total_count = services.len(); format!("{}/{} services healthy", healthy_count, total_count) } } #[cfg(test)] mod tests { use super::*; #[tokio::test] #[ignore] // Requires running services (database, JWT, ML pipeline) async fn test_framework_creation() { let framework = E2ETestFramework::new().await; assert!(framework.is_ok()); let framework = framework.unwrap(); assert!(!framework.services_started); assert!(!framework.test_session_id.is_empty()); } #[test] fn test_service_health_summary() { let status = ServicesHealthStatus { trading_service: ServiceHealth::Healthy, config_service: ServiceHealth::Unhealthy, database: ServiceHealth::Healthy, all_healthy: false, }; let summary = status.summary(); assert_eq!(summary, "2/3 services healthy"); } }