""" Enterprise-Grade Testing Infrastructure This module provides a comprehensive, production-ready testing framework with: - Advanced test orchestration - Real-time monitoring and alerting - Intelligent test data management - Performance benchmarking - Security testing - Compliance validation - Automated reporting """ import asyncio import json import logging import time from datetime import datetime, timezone, timedelta from typing import Dict, List, Any, Optional, Union, Callable, Tuple from dataclasses import dataclass, field, asdict from enum import Enum from pathlib import Path import uuid import statistics from concurrent.futures import ThreadPoolExecutor, as_completed import threading from ..utils.logging import get_logger logger = get_logger(__name__) class TestStatus(Enum): """Comprehensive test status enumeration.""" PENDING = "pending" RUNNING = "running" PASSED = "passed" FAILED = "failed" SKIPPED = "skipped" TIMEOUT = "timeout" ERROR = "error" BLOCKED = "blocked" FLAKY = "flaky" class TestPriority(Enum): """Test priority levels.""" CRITICAL = 1 HIGH = 2 MEDIUM = 3 LOW = 4 OPTIONAL = 5 class TestCategory(Enum): """Test categories for organization.""" UNIT = "unit" INTEGRATION = "integration" SYSTEM = "system" PERFORMANCE = "performance" SECURITY = "security" COMPLIANCE = "compliance" REGRESSION = "regression" SMOKE = "smoke" E2E = "end_to_end" @dataclass class TestConfiguration: """Comprehensive test configuration.""" test_id: str name: str description: str category: TestCategory priority: TestPriority timeout_seconds: int = 300 retry_count: int = 3 retry_delay: float = 1.0 parallel_execution: bool = True dependencies: List[str] = field(default_factory=list) tags: List[str] = field(default_factory=list) environment_requirements: Dict[str, Any] = field(default_factory=dict) data_requirements: Dict[str, Any] = field(default_factory=dict) cleanup_required: bool = True notifications: Dict[str, List[str]] = field(default_factory=dict) @dataclass class TestResult: """Detailed test result with comprehensive metrics.""" test_id: str test_name: str status: TestStatus start_time: datetime end_time: Optional[datetime] = None duration_seconds: Optional[float] = None # Execution details attempt_number: int = 1 max_attempts: int = 1 error_message: Optional[str] = None stack_trace: Optional[str] = None # Performance metrics cpu_usage: Optional[float] = None memory_usage: Optional[float] = None network_requests: int = 0 database_queries: int = 0 # Test-specific metrics custom_metrics: Dict[str, Any] = field(default_factory=dict) # Context and environment environment: str = "default" test_data_hash: Optional[str] = None configuration_hash: Optional[str] = None # Quality metrics flakiness_score: float = 0.0 reliability_score: float = 1.0 def __post_init__(self): if self.end_time and self.start_time: self.duration_seconds = (self.end_time - self.start_time).total_seconds() @dataclass class TestSuite: """Advanced test suite with orchestration capabilities.""" suite_id: str name: str description: str tests: List[TestConfiguration] suite_config: Dict[str, Any] = field(default_factory=dict) execution_order: str = "priority" # priority, dependency, parallel max_concurrent_tests: int = 10 timeout_seconds: int = 3600 retry_policy: Dict[str, Any] = field(default_factory=dict) reporting_config: Dict[str, Any] = field(default_factory=dict) @dataclass class TestExecution: """Test execution context and state.""" execution_id: str suite_id: str start_time: datetime status: TestStatus = TestStatus.PENDING results: List[TestResult] = field(default_factory=list) environment: str = "default" triggered_by: str = "system" metadata: Dict[str, Any] = field(default_factory=dict) class TestDataManager: """Advanced test data management with versioning and isolation.""" def __init__(self, data_path: str = "test_data"): self.data_path = Path(data_path) self.data_path.mkdir(exist_ok=True) self.data_cache: Dict[str, Any] = {} self.data_versions: Dict[str, List[str]] = {} async def generate_test_data(self, test_id: str, data_spec: Dict[str, Any], version: Optional[str] = None) -> Dict[str, Any]: """Generate test data based on specification.""" if not version: version = datetime.now().strftime("%Y%m%d_%H%M%S") data = await self._generate_data_from_spec(data_spec) # Store versioned data data_file = self.data_path / f"{test_id}_{version}.json" with open(data_file, 'w') as f: json.dump(data, f, indent=2, default=str) # Update version tracking if test_id not in self.data_versions: self.data_versions[test_id] = [] self.data_versions[test_id].append(version) return data async def _generate_data_from_spec(self, spec: Dict[str, Any]) -> Dict[str, Any]: """Generate data from specification.""" data = {} for key, value in spec.items(): if isinstance(value, dict): if value.get("type") == "user_profile": data[key] = await self._generate_user_profile(value) elif value.get("type") == "trip_request": data[key] = await self._generate_trip_request(value) elif value.get("type") == "api_response": data[key] = await self._generate_api_response(value) else: data[key] = value.get("default", None) else: data[key] = value return data async def _generate_user_profile(self, spec: Dict[str, Any]) -> Dict[str, Any]: """Generate realistic user profile data.""" import random user_types = ["budget_traveler", "luxury_seeker", "family_traveler", "business_traveler"] return { "user_id": f"test_user_{uuid.uuid4().hex[:8]}", "user_type": random.choice(user_types), "budget": random.randint(500, 5000), "preferences": { "airline": random.choice(["AA", "DL", "UA", "SW"]), "hotel_chain": random.choice(["Marriott", "Hilton", "Hyatt", "IHG"]), "travel_style": random.choice(["adventure", "relaxation", "business", "family"]) }, "travel_history": random.randint(0, 20) } async def _generate_trip_request(self, spec: Dict[str, Any]) -> Dict[str, Any]: """Generate realistic trip request data.""" import random origins = ["NYC", "LAX", "CHI", "DFW", "SEA"] destinations = ["LON", "PAR", "ROM", "BCN", "AMS"] return { "origin": random.choice(origins), "destination": random.choice(destinations), "departure_date": (datetime.now() + timedelta(days=random.randint(7, 90))).strftime("%Y-%m-%d"), "return_date": (datetime.now() + timedelta(days=random.randint(14, 120))).strftime("%Y-%m-%d"), "passengers": random.randint(1, 6), "budget": random.randint(800, 4000) } async def _generate_api_response(self, spec: Dict[str, Any]) -> Dict[str, Any]: """Generate mock API response data.""" return { "status": "success", "data": spec.get("mock_data", {}), "timestamp": datetime.now().isoformat(), "request_id": uuid.uuid4().hex } class PerformanceMonitor: """Advanced performance monitoring and benchmarking.""" def __init__(self): self.metrics_history: List[Dict[str, Any]] = [] self.baseline_metrics: Dict[str, float] = {} self.performance_thresholds: Dict[str, Tuple[float, float]] = {} # (warning, critical) async def start_monitoring(self, test_id: str) -> str: """Start performance monitoring for a test.""" monitor_id = f"monitor_{uuid.uuid4().hex[:8]}" # Start monitoring in background asyncio.create_task(self._monitor_performance(monitor_id, test_id)) return monitor_id async def _monitor_performance(self, monitor_id: str, test_id: str): """Monitor performance metrics during test execution.""" start_time = time.time() while True: try: # Collect system metrics metrics = await self._collect_system_metrics() metrics.update({ "monitor_id": monitor_id, "test_id": test_id, "timestamp": datetime.now(timezone.utc), "elapsed_time": time.time() - start_time }) self.metrics_history.append(metrics) # Check thresholds await self._check_performance_thresholds(metrics) await asyncio.sleep(1) # Monitor every second except Exception as e: logger.error(f"Error in performance monitoring: {e}") break async def _collect_system_metrics(self) -> Dict[str, float]: """Collect system performance metrics.""" try: import psutil return { "cpu_percent": psutil.cpu_percent(), "memory_percent": psutil.virtual_memory().percent, "disk_io_read": psutil.disk_io_counters().read_bytes if psutil.disk_io_counters() else 0, "disk_io_write": psutil.disk_io_counters().write_bytes if psutil.disk_io_counters() else 0, "network_bytes_sent": psutil.net_io_counters().bytes_sent if psutil.net_io_counters() else 0, "network_bytes_recv": psutil.net_io_counters().bytes_recv if psutil.net_io_counters() else 0 } except ImportError: # Fallback if psutil not available return { "cpu_percent": 0.0, "memory_percent": 0.0, "disk_io_read": 0, "disk_io_write": 0, "network_bytes_sent": 0, "network_bytes_recv": 0 } async def _check_performance_thresholds(self, metrics: Dict[str, Any]): """Check if metrics exceed performance thresholds.""" for metric_name, value in metrics.items(): if metric_name in self.performance_thresholds: warning_threshold, critical_threshold = self.performance_thresholds[metric_name] if value > critical_threshold: logger.critical(f"Critical performance threshold exceeded: {metric_name}={value}") elif value > warning_threshold: logger.warning(f"Performance threshold warning: {metric_name}={value}") class EnterpriseTestExecutor: """Enterprise-grade test executor with advanced features.""" def __init__(self, max_workers: int = 10, test_timeout: int = 300, retry_policy: Dict[str, Any] = None): self.max_workers = max_workers self.test_timeout = test_timeout self.retry_policy = retry_policy or {"max_retries": 3, "backoff_factor": 2.0} self.executor = ThreadPoolExecutor(max_workers=max_workers) self.test_data_manager = TestDataManager() self.performance_monitor = PerformanceMonitor() # Execution tracking self.active_executions: Dict[str, TestExecution] = {} self.execution_history: List[TestExecution] = [] # Test registry self.test_registry: Dict[str, Callable] = {} self.test_suites: Dict[str, TestSuite] = {} def register_test(self, test_id: str, test_function: Callable): """Register a test function.""" self.test_registry[test_id] = test_function logger.info(f"Registered test: {test_id}") def register_test_suite(self, suite: TestSuite): """Register a test suite.""" self.test_suites[suite.suite_id] = suite logger.info(f"Registered test suite: {suite.suite_id}") async def execute_test(self, test_config: TestConfiguration, test_data: Optional[Dict[str, Any]] = None) -> TestResult: """Execute a single test with comprehensive monitoring.""" test_result = TestResult( test_id=test_config.test_id, test_name=test_config.name, status=TestStatus.RUNNING, start_time=datetime.now(timezone.utc), max_attempts=test_config.retry_count ) # Start performance monitoring monitor_id = await self.performance_monitor.start_monitoring(test_config.test_id) try: # Get test data if not provided if not test_data: test_data = await self.test_data_manager.generate_test_data( test_config.test_id, test_config.data_requirements ) # Execute test with timeout test_function = self.test_registry.get(test_config.test_id) if not test_function: raise ValueError(f"Test function not found: {test_config.test_id}") # Run test with timeout result = await asyncio.wait_for( self._run_test_with_retry(test_function, test_data, test_config), timeout=test_config.timeout_seconds ) test_result.status = TestStatus.PASSED test_result.custom_metrics = result.get("metrics", {}) except asyncio.TimeoutError: test_result.status = TestStatus.TIMEOUT test_result.error_message = f"Test timed out after {test_config.timeout_seconds} seconds" except Exception as e: test_result.status = TestStatus.FAILED test_result.error_message = str(e) test_result.stack_trace = self._get_stack_trace(e) finally: test_result.end_time = datetime.now(timezone.utc) # Stop performance monitoring and collect final metrics final_metrics = await self._collect_final_metrics(monitor_id) test_result.cpu_usage = final_metrics.get("avg_cpu_percent", 0) test_result.memory_usage = final_metrics.get("avg_memory_percent", 0) logger.info(f"Test {test_config.test_id} completed with status: {test_result.status.value}") return test_result async def _run_test_with_retry(self, test_function: Callable, test_data: Dict[str, Any], test_config: TestConfiguration) -> Dict[str, Any]: """Run test with retry logic.""" last_error = None for attempt in range(test_config.retry_count + 1): try: if attempt > 0: delay = test_config.retry_delay * (self.retry_policy["backoff_factor"] ** (attempt - 1)) await asyncio.sleep(delay) logger.info(f"Retrying test {test_config.test_id}, attempt {attempt + 1}") result = await test_function(test_data) return result except Exception as e: last_error = e logger.warning(f"Test {test_config.test_id} failed on attempt {attempt + 1}: {e}") # All retries failed raise last_error async def _collect_final_metrics(self, monitor_id: str) -> Dict[str, float]: """Collect final performance metrics.""" # Find metrics for this monitor monitor_metrics = [ m for m in self.performance_monitor.metrics_history if m.get("monitor_id") == monitor_id ] if not monitor_metrics: return {} # Calculate averages cpu_values = [m.get("cpu_percent", 0) for m in monitor_metrics] memory_values = [m.get("memory_percent", 0) for m in monitor_metrics] return { "avg_cpu_percent": statistics.mean(cpu_values) if cpu_values else 0, "max_cpu_percent": max(cpu_values) if cpu_values else 0, "avg_memory_percent": statistics.mean(memory_values) if memory_values else 0, "max_memory_percent": max(memory_values) if memory_values else 0, "monitoring_duration": len(monitor_metrics) } def _get_stack_trace(self, exception: Exception) -> str: """Get formatted stack trace.""" import traceback return traceback.format_exc() async def execute_suite(self, suite_id: str, environment: str = "default") -> TestExecution: """Execute a complete test suite with orchestration.""" if suite_id not in self.test_suites: raise ValueError(f"Test suite not found: {suite_id}") suite = self.test_suites[suite_id] execution = TestExecution( execution_id=f"exec_{uuid.uuid4().hex[:8]}", suite_id=suite_id, start_time=datetime.now(timezone.utc), environment=environment ) self.active_executions[execution.execution_id] = execution try: # Sort tests by execution order tests_to_run = self._sort_tests_by_order(suite.tests, suite.execution_order) if suite.execution_order == "parallel": # Execute tests in parallel tasks = [] for test_config in tests_to_run[:suite.max_concurrent_tests]: task = asyncio.create_task(self.execute_test(test_config)) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) for i, result in enumerate(results): if isinstance(result, Exception): # Create failed result test_config = tests_to_run[i] failed_result = TestResult( test_id=test_config.test_id, test_name=test_config.name, status=TestStatus.ERROR, start_time=datetime.now(timezone.utc), end_time=datetime.now(timezone.utc), error_message=str(result) ) execution.results.append(failed_result) else: execution.results.append(result) else: # Execute tests sequentially for test_config in tests_to_run: result = await self.execute_test(test_config) execution.results.append(result) # Check if we should stop on failure if result.status in [TestStatus.FAILED, TestStatus.ERROR] and suite.suite_config.get("stop_on_failure", False): logger.warning(f"Stopping suite execution due to test failure: {test_config.test_id}") break # Determine overall execution status if all(r.status == TestStatus.PASSED for r in execution.results): execution.status = TestStatus.PASSED elif any(r.status == TestStatus.FAILED for r in execution.results): execution.status = TestStatus.FAILED else: execution.status = TestStatus.ERROR except Exception as e: execution.status = TestStatus.ERROR logger.error(f"Suite execution failed: {e}") finally: execution.end_time = datetime.now(timezone.utc) self.execution_history.append(execution) # Remove from active executions if execution.execution_id in self.active_executions: del self.active_executions[execution.execution_id] return execution def _sort_tests_by_order(self, tests: List[TestConfiguration], order: str) -> List[TestConfiguration]: """Sort tests by execution order.""" if order == "priority": return sorted(tests, key=lambda t: t.priority.value) elif order == "dependency": # Simple dependency resolution (in production, use proper topological sort) return tests else: return tests def get_execution_report(self, execution_id: str) -> Dict[str, Any]: """Generate comprehensive execution report.""" execution = next( (e for e in self.execution_history if e.execution_id == execution_id), None ) if not execution: return {"error": "Execution not found"} # Calculate statistics total_tests = len(execution.results) passed_tests = len([r for r in execution.results if r.status == TestStatus.PASSED]) failed_tests = len([r for r in execution.results if r.status == TestStatus.FAILED]) # Performance statistics durations = [r.duration_seconds for r in execution.results if r.duration_seconds] avg_duration = statistics.mean(durations) if durations else 0 return { "execution_id": execution_id, "suite_id": execution.suite_id, "status": execution.status.value, "start_time": execution.start_time.isoformat(), "end_time": execution.end_time.isoformat() if execution.end_time else None, "duration_seconds": (execution.end_time - execution.start_time).total_seconds() if execution.end_time else None, "statistics": { "total_tests": total_tests, "passed_tests": passed_tests, "failed_tests": failed_tests, "success_rate": passed_tests / total_tests if total_tests > 0 else 0, "average_duration": avg_duration }, "results": [asdict(result) for result in execution.results], "performance_summary": self._generate_performance_summary(execution.results) } def _generate_performance_summary(self, results: List[TestResult]) -> Dict[str, Any]: """Generate performance summary from test results.""" cpu_values = [r.cpu_usage for r in results if r.cpu_usage is not None] memory_values = [r.memory_usage for r in results if r.memory_usage is not None] return { "cpu_usage": { "average": statistics.mean(cpu_values) if cpu_values else 0, "maximum": max(cpu_values) if cpu_values else 0, "minimum": min(cpu_values) if cpu_values else 0 }, "memory_usage": { "average": statistics.mean(memory_values) if memory_values else 0, "maximum": max(memory_values) if memory_values else 0, "minimum": min(memory_values) if memory_values else 0 } }