diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index d2c900023..cf3b46727 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 -""" -Comprehensive CI Pipeline Runner for SAMO Deep Learning +"""Comprehensive CI Pipeline Runner for SAMO Deep Learning. This script runs the complete CI pipeline end-to-end, including: - Environment validation @@ -14,50 +13,87 @@ import logging import os +import subprocess import sys import time -import subprocess from pathlib import Path -from typing import Dict, List, Tuple +from typing import Dict, Optional, Tuple # Use shared truthy parsing try: from src.common.env import is_truthy -except Exception: # Fallback to local helper if import path not available - def is_truthy(value: str | None) -> bool: +except ImportError: # Fallback to local helper if import path not available + + def is_truthy(value: Optional[str]) -> bool: return bool(value) and value.strip().lower() in {"1", "true", "yes"} +# Add src to path for local imports and import BERT classifier +sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) + +try: + from models.emotion_detection.bert_classifier import BERTEmotionClassifier +except ImportError: + from src.models.emotion_detection.bert_classifier import BERTEmotionClassifier + + # Configure logging logging.basicConfig( level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', + format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.StreamHandler(sys.stdout), - logging.FileHandler('ci_pipeline.log') - ] + logging.FileHandler("ci_pipeline.log"), + ], ) logger = logging.getLogger(__name__) class CIPipelineRunner: """Comprehensive CI Pipeline Runner.""" - + def __init__(self): self.results = {} self.start_time = time.time() self.ci_scripts = [ "scripts/ci/api_health_check.py", - "scripts/ci/bert_model_test.py", + "scripts/ci/bert_model_test.py", "scripts/ci/t5_summarization_test.py", "scripts/ci/whisper_transcription_test.py", "scripts/ci/model_calibration_test.py", "scripts/ci/onnx_conversion_test.py", ] + @staticmethod + def _run_subprocess_command( + command: list, timeout: int + ) -> subprocess.CompletedProcess: + """Run a subprocess command with standardized parameters. + + Parameters + ---------- + command : list + The command to run as a list of strings + timeout : int + Timeout in seconds + + Returns + ------- + subprocess.CompletedProcess + The completed process result + """ + return subprocess.run( + command, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + def _get_test_stats(self) -> tuple[dict, int, int]: """Calculate statistics on test results. - - Returns: + + Returns + ------- tuple: (test_results dict, total_tests, passed_tests) """ test_results = { @@ -73,7 +109,7 @@ def _get_test_stats(self) -> tuple[dict, int, int]: def detect_environment(self) -> Dict[str, str]: """Detect the current environment (local vs Colab).""" logger.info("๐Ÿ” Detecting environment...") - + env_info = { "platform": sys.platform, "python_version": sys.version, @@ -81,274 +117,374 @@ def detect_environment(self) -> Dict[str, str]: "gpu_available": False, "conda_env": os.environ.get("CONDA_DEFAULT_ENV", "unknown"), } - + # Check for GPU try: import torch + env_info["gpu_available"] = torch.cuda.is_available() if env_info["gpu_available"]: env_info["gpu_count"] = torch.cuda.device_count() env_info["gpu_name"] = torch.cuda.get_device_name(0) except ImportError: logger.warning("โš ๏ธ PyTorch not available for GPU detection") - + # Check for Colab if env_info["is_colab"]: logger.info("๐ŸŽฏ Running in Google Colab environment") env_info["colab_gpu"] = os.environ.get("COLAB_GPU", "unknown") else: logger.info("๐Ÿ’ป Running in local environment") - - logger.info(f"๐Ÿ“Š Environment: {env_info}") + + logger.info("๐Ÿ“Š Environment: %s", env_info) return env_info - + def validate_dependencies(self) -> bool: """Validate that all required dependencies are available.""" logger.info("๐Ÿ“ฆ Validating dependencies...") - + required_packages = [ - "torch", "transformers", "fastapi", "pydantic", - "datasets", "tokenizers", "numpy", "pandas" + "torch", + "transformers", + "fastapi", + "pydantic", + "datasets", + "tokenizers", + "numpy", + "pandas", ] - + missing_packages = [] for package in required_packages: try: __import__(package) - logger.info(f"โœ… {package} available") + logger.info("โœ… %s available", package) except ImportError: missing_packages.append(package) - logger.error(f"โŒ {package} missing") - + logger.error("โŒ %s missing", package) + if missing_packages: - logger.error(f"โŒ Missing packages: {missing_packages}") + logger.error("โŒ Missing packages: %s", missing_packages) return False - + logger.info("โœ… All dependencies validated") return True - + def run_ci_script(self, script_path: str) -> Tuple[bool, str]: """Run a single CI script and return success status and output.""" - logger.info(f"๐Ÿš€ Running {script_path}...") - + logger.info("๐Ÿš€ Running %s...", script_path) + try: # Use the correct Python interpreter python_executable = sys.executable - - # Run the script - result = subprocess.run( + + # Run the script - script_path is controlled internally by self.ci_scripts + # No command injection risk as paths are static and predefined + result = self._run_subprocess_command( [python_executable, script_path], - capture_output=True, - text=True, - timeout=300 # 5 minute timeout + timeout=300, # 5 minute timeout ) - + if result.returncode == 0: - logger.info(f"โœ… {script_path} PASSED") + logger.info("โœ… %s PASSED", script_path) return True, result.stdout else: - logger.error(f"โŒ {script_path} FAILED") - logger.error(f"Error output: {result.stderr}") + logger.error("โŒ %s FAILED", script_path) + logger.error("Error output: %s", result.stderr) + logger.error("Standard output: %s", result.stdout) return False, result.stderr - + except subprocess.TimeoutExpired: - logger.error(f"โฐ {script_path} TIMEOUT") + logger.error("โฐ %s TIMEOUT", script_path) return False, "Script timed out after 5 minutes" except Exception as e: - logger.error(f"๐Ÿ’ฅ {script_path} ERROR: {e}") + logger.exception("๐Ÿ’ฅ %s ERROR: %s", script_path, e) return False, str(e) - + def run_unit_tests(self) -> bool: """Run unit tests.""" logger.info("๐Ÿงช Running unit tests...") - + try: - result = subprocess.run( + # Static command - no injection risk, all arguments are literals + result = self._run_subprocess_command( [sys.executable, "-m", "pytest", "tests/unit/", "-v"], - capture_output=True, - text=True, - timeout=1200 # 20 minute timeout (increased from 10) + timeout=1200, # 20 minute timeout (increased from 10) ) - + if result.returncode == 0: logger.info("โœ… Unit tests PASSED") return True else: logger.error("โŒ Unit tests FAILED") - logger.error(f"Return code: {result.returncode}") - logger.error(f"Error output: {result.stderr}") - logger.error(f"Standard output: {result.stdout}") + logger.error("Return code: %s", result.returncode) + logger.error("Error output: %s", result.stderr) + logger.error("Standard output: %s", result.stdout) return False - + except subprocess.TimeoutExpired: logger.error("โฐ Unit tests TIMEOUT") return False except Exception as e: - logger.error(f"๐Ÿ’ฅ Unit tests ERROR: {e}") + logger.exception("๐Ÿ’ฅ Unit tests ERROR: %s", e) return False - + def run_e2e_tests(self) -> bool: """Run end-to-end tests.""" logger.info("๐ŸŽฏ Running E2E tests...") - + try: - result = subprocess.run( + # Static command - no injection risk, all arguments are literals + result = self._run_subprocess_command( [sys.executable, "-m", "pytest", "tests/e2e/", "-v"], - capture_output=True, - text=True, - timeout=900 # 15 minute timeout + timeout=900, # 15 minute timeout ) - + if result.returncode == 0: logger.info("โœ… E2E tests PASSED") return True else: logger.error("โŒ E2E tests FAILED") - logger.error(f"Error output: {result.stderr}") + logger.error("Error output: %s", result.stderr) + logger.error("Standard output: %s", result.stdout) return False - + except Exception as e: - logger.error(f"๐Ÿ’ฅ E2E tests ERROR: {e}") + logger.exception("๐Ÿ’ฅ E2E tests ERROR: %s", e) return False - - def test_gpu_compatibility(self) -> bool: - """Test GPU compatibility if available.""" - logger.info("๐Ÿ–ฅ๏ธ Testing GPU compatibility...") - + + @staticmethod + def _test_gpu_model_forward_pass() -> bool: + """Test GPU model forward pass with BERT classifier. + + Returns + ------- + bool + True if GPU forward pass succeeds, False otherwise + """ try: import torch - - if not torch.cuda.is_available(): - logger.info("โ„น๏ธ No GPU available, skipping GPU tests") - return True - - logger.info(f"๐ŸŽฎ GPU detected: {torch.cuda.get_device_name(0)}") - - # Test GPU model loading + device = torch.device("cuda") - - # Add src to path for imports - import sys - from pathlib import Path - sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) - - # Test BERT on GPU - try: - from models.emotion_detection.bert_classifier import BERTEmotionClassifier - except ImportError: - from src.models.emotion_detection.bert_classifier import BERTEmotionClassifier + + # Use the module-level BERT classifier import model = BERTEmotionClassifier().to(device) - + # Test forward pass - import torch dummy_input = torch.randint(0, 1000, (2, 512)).to(device) with torch.no_grad(): output = model(dummy_input, torch.ones_like(dummy_input)) - - logger.info(f"โœ… GPU forward pass successful, output shape: {output.shape}") + + logger.info( + "โœ… GPU forward pass successful, output shape: %s", output.shape + ) return True - + + except Exception as e: + logger.exception("โŒ GPU model forward pass failed: %s", e) + return False + + def test_gpu_compatibility(self) -> bool: + """Test GPU compatibility if available.""" + logger.info("๐Ÿ–ฅ๏ธ Testing GPU compatibility...") + + try: + import torch + + if not torch.cuda.is_available(): + logger.info("โ„น๏ธ No GPU available, skipping GPU tests") + return True + + logger.info("๐ŸŽฎ GPU detected: %s", torch.cuda.get_device_name(0)) + + # Test GPU model loading and forward pass + return self._test_gpu_model_forward_pass() + except Exception as e: - logger.error(f"โŒ GPU compatibility test failed: {e}") + logger.exception("โŒ GPU compatibility test failed: %s", e) return False - + + @staticmethod + def _measure_model_loading_time() -> float: + """Measure BERT model loading time. + + Returns + ------- + float + Loading time in seconds + """ + start_time = time.time() + _ = BERTEmotionClassifier() # Instantiate model to measure loading time + loading_time = time.time() - start_time + + logger.info("โœ… Model loading time: %.2fs", loading_time) + return loading_time + + @staticmethod + def _measure_inference_time(model) -> float: + """Measure model inference time. + + Parameters + ---------- + model : torch.nn.Module + The model to test inference on + + Returns + ------- + float + Inference time in seconds + """ + import torch + + start_time = time.time() + dummy_input = torch.randint(0, 1000, (1, 512)) + with torch.no_grad(): + model(dummy_input, torch.ones_like(dummy_input)) + inference_time = time.time() - start_time + + logger.info("โœ… Inference time: %.2fs", inference_time) + return inference_time + + @staticmethod + def _validate_performance_thresholds( + loading_time: float, inference_time: float + ) -> bool: + """Validate that performance times are within acceptable thresholds. + + Parameters + ---------- + loading_time : float + Model loading time in seconds + inference_time : float + Inference time in seconds + + Returns + ------- + bool + True if performance is acceptable, False otherwise + """ + # Increased threshold for CPU environments + if loading_time < 10.0 and inference_time < 5.0: + logger.info("โœ… Performance benchmarks passed") + return True + + logger.error( + "โŒ Performance too slow - loading: %.2fs, inference: %.2fs", + loading_time, + inference_time, + ) + return False + def run_performance_benchmarks(self) -> bool: """Run performance benchmarks.""" logger.info("โšก Running performance benchmarks...") - + try: - # Simple performance test - model loading speed - import time - import torch - - # Test BERT model loading speed - start_time = time.time() - - # Add src to path - import sys - from pathlib import Path - sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) - - try: - from models.emotion_detection.bert_classifier import BERTEmotionClassifier - except ImportError: - from src.models.emotion_detection.bert_classifier import BERTEmotionClassifier - + # Measure model loading time + loading_time = self._measure_model_loading_time() + + # Import model again for inference test (avoid reusing loaded model) model = BERTEmotionClassifier() - loading_time = time.time() - start_time - - # Test inference speed - start_time = time.time() - dummy_input = torch.randint(0, 1000, (1, 512)) - with torch.no_grad(): - output = model(dummy_input, torch.ones_like(dummy_input)) - inference_time = time.time() - start_time - - logger.info(f"โœ… Model loading time: {loading_time:.2f}s") - logger.info(f"โœ… Inference time: {inference_time:.2f}s") - - # Check if times are reasonable - if loading_time < 10.0 and inference_time < 5.0: # Increased threshold for CPU environments - logger.info("โœ… Performance benchmarks passed") - return True - else: - logger.error(f"โŒ Performance too slow - loading: {loading_time:.2f}s, inference: {inference_time:.2f}s") - return False - + + # Measure inference time + inference_time = self._measure_inference_time(model) + + # Validate performance thresholds + return self._validate_performance_thresholds(loading_time, inference_time) + except Exception as e: - logger.error(f"โŒ Performance benchmark failed: {e}") + logger.exception("โŒ Performance benchmark failed: %s", e) return False - + + def run_pipeline_and_exit(self) -> None: + """Run the CI pipeline and exit with appropriate code. + + This method handles the complete pipeline execution flow including: + - Running all tests + - Generating reports + - Writing CI artifacts + - Exiting with proper status codes + """ + try: + _ = self.run_full_pipeline() + report = self.generate_report() + + print(report) + # Only write report to file in CI so it can be uploaded as an artifact + write_ci_report_if_needed(report) + + # Exit with appropriate code + _, total_tests, passed_tests = self._get_test_stats() + + if total_tests == 0: + logger.error("โŒ CI Pipeline failed - no boolean tests were executed!") + sys.exit(1) + elif passed_tests == total_tests: + logger.info("๐ŸŽ‰ CI Pipeline completed successfully!") + sys.exit(0) + else: + logger.error("โŒ CI Pipeline failed!") + sys.exit(1) + + except KeyboardInterrupt: + logger.info("โน๏ธ CI Pipeline interrupted by user") + sys.exit(1) + except Exception as e: + logger.exception("๐Ÿ’ฅ CI Pipeline crashed: %s", e) + sys.exit(1) + def run_full_pipeline(self) -> Dict[str, bool]: """Run the complete CI pipeline.""" logger.info("๐Ÿš€ Starting Comprehensive CI Pipeline") logger.info("=" * 60) - + # Environment detection env_info = self.detect_environment() self.results["environment"] = env_info - + # Dependency validation self.results["dependencies"] = self.validate_dependencies() - + # Run individual CI scripts for script in self.ci_scripts: script_name = Path(script).stem success, output = self.run_ci_script(script) self.results[script_name] = success - + if not success: - logger.error(f"โŒ {script_name} failed, but continuing...") - + logger.error("โŒ %s failed, but continuing...", script_name) + # Run unit tests self.results["unit_tests"] = self.run_unit_tests() - + # Run E2E tests self.results["e2e_tests"] = self.run_e2e_tests() - + # Test GPU compatibility self.results["gpu_compatibility"] = self.test_gpu_compatibility() - + # Run performance benchmarks self.results["performance"] = self.run_performance_benchmarks() - + return self.results - + def generate_report(self) -> str: """Generate a comprehensive CI report.""" logger.info("๐Ÿ“Š Generating CI Report") logger.info("=" * 60) - + # Only count boolean results as actual tests test_results, total_tests, passed_tests = self._get_test_stats() - - # Guard against division by zero when no boolean tests were collected - safe_total = total_tests if total_tests > 0 else 1 - success_rate = (passed_tests / safe_total) * 100.0 + + # Handle case where no boolean tests were collected + if total_tests == 0: + success_rate = 0.0 + else: + success_rate = (passed_tests / total_tests) * 100.0 report = f""" ๐ŸŽฏ COMPREHENSIVE CI PIPELINE REPORT -{'=' * 60} +{"=" * 60} ๐Ÿ“Š SUMMARY: - Total Tests: {total_tests} @@ -358,28 +494,33 @@ def generate_report(self) -> str: ๐Ÿ” DETAILED RESULTS: """ - + for test_name, result in self.results.items(): if isinstance(result, bool): status = "โœ… PASSED" if result else "โŒ FAILED" report += f"- {test_name}: {status}\n" elif isinstance(result, dict): report += f"- {test_name}: {result}\n" - + report += f""" โฑ๏ธ EXECUTION TIME: {time.time() - self.start_time:.1f}s ๐ŸŽฏ RECOMMENDATIONS: """ - - if passed_tests == total_tests: + + if total_tests == 0: + report += ( + "โš ๏ธ No boolean tests were executed. Treating pipeline as failed.\n" + ) + elif passed_tests == total_tests: report += "๐ŸŽ‰ All tests passed! Pipeline is ready for deployment.\n" else: - failed_test_names = [name for name, result in test_results.items() - if not result] + failed_test_names = [ + name for name, result in test_results.items() if not result + ] report += f"โš ๏ธ Failed tests: {', '.join(failed_test_names)}\n" report += "๐Ÿ”ง Please fix the failed tests before deployment.\n" - + return report @@ -393,32 +534,8 @@ def write_ci_report_if_needed(report: str) -> None: def main(): """Main function to run the CI pipeline.""" runner = CIPipelineRunner() - - try: - _ = runner.run_full_pipeline() - report = runner.generate_report() - - print(report) - # Only write report to file in CI so it can be uploaded as an artifact - write_ci_report_if_needed(report) - - # Exit with appropriate code - _, total_tests, passed_tests = runner._get_test_stats() - - if passed_tests == total_tests: - logger.info("๐ŸŽ‰ CI Pipeline completed successfully!") - sys.exit(0) - else: - logger.error("โŒ CI Pipeline failed!") - sys.exit(1) - - except KeyboardInterrupt: - logger.info("โน๏ธ CI Pipeline interrupted by user") - sys.exit(1) - except Exception as e: - logger.error(f"๐Ÿ’ฅ CI Pipeline crashed: {e}") - sys.exit(1) + runner.run_pipeline_and_exit() if __name__ == "__main__": - main() \ No newline at end of file + main()