From b763c9cbe68f3da6334f0a531210df30e609c4a8 Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 16:11:29 +0200 Subject: [PATCH 01/12] feat: add comprehensive CI pipeline orchestration system MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extracted from monster PR #171 as part of Phase 3A strategic decomposition. Comprehensive CI pipeline runner with: - Multi-environment detection (local/Colab/CI) - GPU compatibility testing with PyTorch - Dependency validation for all required packages - Individual CI script orchestration with timeouts - Unit and E2E test execution (pytest integration) - Performance benchmarking with thresholds - Detailed reporting with success metrics - CI artifact generation (conditional report writing) - Graceful error handling and logging Supports both local development and production CI workflows. ๐Ÿค– Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- scripts/ci/run_full_ci_pipeline.py | 259 ++++++++++++++++------------- 1 file changed, 141 insertions(+), 118 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index d2c900023..420a729c5 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,40 +13,42 @@ 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, 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: return bool(value) and value.strip().lower() in {"1", "true", "yes"} + # 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", @@ -56,8 +57,9 @@ def __init__(self): 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 +75,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,36 +83,43 @@ 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}") 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: @@ -118,237 +127,250 @@ def validate_dependencies(self) -> bool: logger.info(f"โœ… {package} available") except ImportError: missing_packages.append(package) - logger.error(f"โŒ {package} missing") - + logger.exception(f"โŒ {package} missing") + if missing_packages: logger.error(f"โŒ Missing packages: {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}...") - + try: # Use the correct Python interpreter python_executable = sys.executable - + # Run the script result = subprocess.run( [python_executable, script_path], + check=False, 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") return True, result.stdout - else: - logger.error(f"โŒ {script_path} FAILED") - logger.error(f"Error output: {result.stderr}") - return False, result.stderr - + logger.error(f"โŒ {script_path} FAILED") + logger.error(f"Error output: {result.stderr}") + return False, result.stderr + except subprocess.TimeoutExpired: - logger.error(f"โฐ {script_path} TIMEOUT") + logger.exception(f"โฐ {script_path} TIMEOUT") return False, "Script timed out after 5 minutes" except Exception as e: - logger.error(f"๐Ÿ’ฅ {script_path} ERROR: {e}") + logger.exception(f"๐Ÿ’ฅ {script_path} ERROR: {e}") return False, str(e) - + def run_unit_tests(self) -> bool: """Run unit tests.""" logger.info("๐Ÿงช Running unit tests...") - + try: result = subprocess.run( [sys.executable, "-m", "pytest", "tests/unit/", "-v"], + check=False, 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}") - return False - + 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}") + return False + except subprocess.TimeoutExpired: - logger.error("โฐ Unit tests TIMEOUT") + logger.exception("โฐ Unit tests TIMEOUT") return False except Exception as e: - logger.error(f"๐Ÿ’ฅ Unit tests ERROR: {e}") + logger.exception(f"๐Ÿ’ฅ Unit tests ERROR: {e}") return False - + def run_e2e_tests(self) -> bool: """Run end-to-end tests.""" logger.info("๐ŸŽฏ Running E2E tests...") - + try: result = subprocess.run( [sys.executable, "-m", "pytest", "tests/e2e/", "-v"], + check=False, 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}") - return False - + logger.error("โŒ E2E tests FAILED") + logger.error(f"Error output: {result.stderr}") + return False + except Exception as e: - logger.error(f"๐Ÿ’ฅ E2E tests ERROR: {e}") + logger.exception(f"๐Ÿ’ฅ E2E tests ERROR: {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(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 + from models.emotion_detection.bert_classifier import ( + BERTEmotionClassifier, + ) except ImportError: - from src.models.emotion_detection.bert_classifier import BERTEmotionClassifier + from src.models.emotion_detection.bert_classifier import ( + BERTEmotionClassifier, + ) 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}") return True - + except Exception as e: - logger.error(f"โŒ GPU compatibility test failed: {e}") + logger.exception(f"โŒ GPU compatibility test failed: {e}") 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 + from models.emotion_detection.bert_classifier import ( + BERTEmotionClassifier, + ) except ImportError: - from src.models.emotion_detection.bert_classifier import BERTEmotionClassifier - + from src.models.emotion_detection.bert_classifier import ( + BERTEmotionClassifier, + ) + 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)) + 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 + 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 - + logger.error( + f"โŒ Performance too slow - loading: {loading_time:.2f}s, inference: {inference_time:.2f}s", + ) + return False + except Exception as e: - logger.error(f"โŒ Performance benchmark failed: {e}") + logger.exception(f"โŒ Performance benchmark failed: {e}") return False - + 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...") - + # 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 report = f""" ๐ŸŽฏ COMPREHENSIVE CI PIPELINE REPORT -{'=' * 60} +{"=" * 60} ๐Ÿ“Š SUMMARY: - Total Tests: {total_tests} @@ -358,28 +380,29 @@ 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: 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 +416,32 @@ 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}") + logger.exception(f"๐Ÿ’ฅ CI Pipeline crashed: {e}") sys.exit(1) if __name__ == "__main__": - main() \ No newline at end of file + main() From 75f6734bbb49eb4a6331c7f31e8b3b88a3d1c618 Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 16:33:00 +0200 Subject: [PATCH 02/12] fix: address code review feedback for CI pipeline orchestration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace logger.exception with logger.error for expected errors (TimeoutExpired, ImportError) - Add security comments clarifying no command injection risk in subprocess calls - All arguments are static literals or controlled internally Addresses Sourcery AI review feedback to improve code quality and security clarity. ๐Ÿค– Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- scripts/ci/run_full_ci_pipeline.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 420a729c5..b9a8f6be0 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -127,7 +127,7 @@ def validate_dependencies(self) -> bool: logger.info(f"โœ… {package} available") except ImportError: missing_packages.append(package) - logger.exception(f"โŒ {package} missing") + logger.error(f"โŒ {package} missing") if missing_packages: logger.error(f"โŒ Missing packages: {missing_packages}") @@ -144,7 +144,8 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: # Use the correct Python interpreter python_executable = sys.executable - # Run the script + # Run the script - script_path is controlled internally by self.ci_scripts + # No command injection risk as paths are static and predefined result = subprocess.run( [python_executable, script_path], check=False, @@ -161,7 +162,7 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: return False, result.stderr except subprocess.TimeoutExpired: - logger.exception(f"โฐ {script_path} TIMEOUT") + logger.error(f"โฐ {script_path} TIMEOUT") return False, "Script timed out after 5 minutes" except Exception as e: logger.exception(f"๐Ÿ’ฅ {script_path} ERROR: {e}") @@ -172,6 +173,7 @@ def run_unit_tests(self) -> bool: logger.info("๐Ÿงช Running unit tests...") try: + # Static command - no injection risk, all arguments are literals result = subprocess.run( [sys.executable, "-m", "pytest", "tests/unit/", "-v"], check=False, @@ -190,7 +192,7 @@ def run_unit_tests(self) -> bool: return False except subprocess.TimeoutExpired: - logger.exception("โฐ Unit tests TIMEOUT") + logger.error("โฐ Unit tests TIMEOUT") return False except Exception as e: logger.exception(f"๐Ÿ’ฅ Unit tests ERROR: {e}") @@ -201,6 +203,7 @@ def run_e2e_tests(self) -> bool: logger.info("๐ŸŽฏ Running E2E tests...") try: + # Static command - no injection risk, all arguments are literals result = subprocess.run( [sys.executable, "-m", "pytest", "tests/e2e/", "-v"], check=False, From 8f4c836a04bc60650033d1563484633432f54c6d Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:29:42 +0200 Subject: [PATCH 03/12] Address gemini-code-assist review comments: refactor imports, improve logging consistency - Change broad Exception to specific ImportError for is_truthy import - Move BERT classifier import logic to module level to eliminate duplication - Add stdout logging to run_ci_script and run_e2e_tests methods for consistency - Remove duplicated _import_bert_classifier method and sys.path.insert calls --- scripts/ci/run_full_ci_pipeline.py | 269 ++++++++++++++++++----------- 1 file changed, 169 insertions(+), 100 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index b9a8f6be0..49c12a05a 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -22,11 +22,19 @@ # Use shared truthy parsing try: from src.common.env import is_truthy -except Exception: # Fallback to local helper if import path not available +except ImportError: # Fallback to local helper if import path not available def is_truthy(value: str | None) -> 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( @@ -55,6 +63,29 @@ def __init__(self): "scripts/ci/onnx_conversion_test.py", ] + def _run_subprocess_command(self, 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. @@ -146,11 +177,8 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: # Run the script - script_path is controlled internally by self.ci_scripts # No command injection risk as paths are static and predefined - result = subprocess.run( + result = self._run_subprocess_command( [python_executable, script_path], - check=False, - capture_output=True, - text=True, timeout=300, # 5 minute timeout ) @@ -159,6 +187,7 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: return True, result.stdout logger.error(f"โŒ {script_path} FAILED") logger.error(f"Error output: {result.stderr}") + logger.error(f"Standard output: {result.stdout}") return False, result.stderr except subprocess.TimeoutExpired: @@ -174,11 +203,8 @@ def run_unit_tests(self) -> bool: try: # Static command - no injection risk, all arguments are literals - result = subprocess.run( + result = self._run_subprocess_command( [sys.executable, "-m", "pytest", "tests/unit/", "-v"], - check=False, - capture_output=True, - text=True, timeout=1200, # 20 minute timeout (increased from 10) ) @@ -204,11 +230,8 @@ def run_e2e_tests(self) -> bool: try: # Static command - no injection risk, all arguments are literals - result = subprocess.run( + result = self._run_subprocess_command( [sys.executable, "-m", "pytest", "tests/e2e/", "-v"], - check=False, - capture_output=True, - text=True, timeout=900, # 15 minute timeout ) @@ -217,12 +240,41 @@ def run_e2e_tests(self) -> bool: return True logger.error("โŒ E2E tests FAILED") logger.error(f"Error output: {result.stderr}") + logger.error(f"Standard output: {result.stdout}") return False except Exception as e: logger.exception(f"๐Ÿ’ฅ E2E tests ERROR: {e}") return False + def _test_gpu_model_forward_pass(self) -> bool: + """Test GPU model forward pass with BERT classifier. + + Returns + ------- + bool + True if GPU forward pass succeeds, False otherwise + """ + try: + import torch + + device = torch.device("cuda") + + # Use the module-level BERT classifier import + model = BERTEmotionClassifier().to(device) + + # Test forward pass + 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}") + return True + + except Exception as e: + logger.exception(f"โŒ GPU model forward pass failed: {e}") + return False + def test_gpu_compatibility(self) -> bool: """Test GPU compatibility if available.""" logger.info("๐Ÿ–ฅ๏ธ Testing GPU compatibility...") @@ -236,93 +288,134 @@ def test_gpu_compatibility(self) -> bool: logger.info(f"๐ŸŽฎ GPU detected: {torch.cuda.get_device_name(0)}") - # Test GPU model loading - device = torch.device("cuda") + # Test GPU model loading and forward pass + return self._test_gpu_model_forward_pass() - # Add src to path for imports - import sys + except Exception as e: + logger.exception(f"โŒ GPU compatibility test failed: {e}") + return False - sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) + def _measure_model_loading_time(self) -> float: + """Measure BERT model loading time. - # Test BERT on GPU - try: - from models.emotion_detection.bert_classifier import ( - BERTEmotionClassifier, - ) - except ImportError: - from src.models.emotion_detection.bert_classifier import ( - BERTEmotionClassifier, - ) - model = BERTEmotionClassifier().to(device) + Returns + ------- + float + Loading time in seconds + """ + import time - # Test forward pass - import torch + start_time = time.time() + model = BERTEmotionClassifier() + loading_time = time.time() - start_time - 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"โœ… Model loading time: {loading_time:.2f}s") + return loading_time - logger.info(f"โœ… GPU forward pass successful, output shape: {output.shape}") + def _measure_inference_time(self, model) -> float: + """Measure model inference time. + + Parameters + ---------- + model : torch.nn.Module + The model to test inference on + + Returns + ------- + float + Inference time in seconds + """ + import time + 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(f"โœ… Inference time: {inference_time:.2f}s") + return inference_time + + def _validate_performance_thresholds(self, 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 - except Exception as e: - logger.exception(f"โŒ GPU compatibility test failed: {e}") - return False + logger.error( + f"โŒ Performance too slow - loading: {loading_time:.2f}s, inference: {inference_time:.2f}s", + ) + 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 + # Measure model loading time + loading_time = self._measure_model_loading_time() - import torch + # Import model again for inference test (avoid reusing loaded model) + model = BERTEmotionClassifier() - # Test BERT model loading speed - start_time = time.time() + # Measure inference time + inference_time = self._measure_inference_time(model) - # Add src to path - import sys + # Validate performance thresholds + return self._validate_performance_thresholds(loading_time, inference_time) - sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) + except Exception as e: + logger.exception(f"โŒ Performance benchmark failed: {e}") + return False - try: - from models.emotion_detection.bert_classifier import ( - BERTEmotionClassifier, - ) - except ImportError: - from src.models.emotion_detection.bert_classifier import ( - BERTEmotionClassifier, - ) + def run_pipeline_and_exit(self) -> None: + """Run the CI pipeline and exit with appropriate code. - model = BERTEmotionClassifier() - loading_time = time.time() - start_time + 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() - # Test inference speed - 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 + print(report) + # Only write report to file in CI so it can be uploaded as an artifact + write_ci_report_if_needed(report) - logger.info(f"โœ… Model loading time: {loading_time:.2f}s") - logger.info(f"โœ… Inference time: {inference_time:.2f}s") + # Exit with appropriate code + _, total_tests, passed_tests = self._get_test_stats() - # 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 - logger.error( - f"โŒ Performance too slow - loading: {loading_time:.2f}s, inference: {inference_time:.2f}s", - ) - return False + 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.exception(f"โŒ Performance benchmark failed: {e}") - return False + logger.exception(f"๐Ÿ’ฅ CI Pipeline crashed: {e}") + sys.exit(1) def run_full_pipeline(self) -> Dict[str, bool]: """Run the complete CI pipeline.""" @@ -419,31 +512,7 @@ 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.exception(f"๐Ÿ’ฅ CI Pipeline crashed: {e}") - sys.exit(1) + runner.run_pipeline_and_exit() if __name__ == "__main__": From 2ead216909cf24865532184ca5ed68e8d18e6c2b Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:31:45 +0200 Subject: [PATCH 04/12] Fix zero-test case reporting logic in CI pipeline - Detect when no boolean tests are executed (total_tests == 0) - Set success_rate = 0.0 explicitly for zero-test scenarios - Display explicit failure message: 'No boolean tests were executed. Treating pipeline as failed.' - Update exit logic to fail when no tests are executed - Prevent false 'All tests passed' reports when actually no tests ran --- scripts/ci/run_full_ci_pipeline.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 49c12a05a..41058620c 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -403,7 +403,10 @@ def run_pipeline_and_exit(self) -> None: # Exit with appropriate code _, total_tests, passed_tests = self._get_test_stats() - if passed_tests == total_tests: + 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: @@ -460,9 +463,11 @@ def generate_report(self) -> str: # 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 @@ -490,7 +495,9 @@ def generate_report(self) -> str: ๐ŸŽฏ 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 = [ From 9cbdf854d726831aed8d4fe155e3cb6833e27df0 Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:33:34 +0200 Subject: [PATCH 05/12] Add explicit else clauses for test result handling - Add else: clauses to run_ci_script, run_unit_tests, and run_e2e_tests methods - Make control flow more explicit and clear for developers and AI tools - Prevent any confusion about when error logging occurs (only on failure) - Maintain existing functionality while improving code readability --- scripts/ci/run_full_ci_pipeline.py | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 41058620c..35b554ec9 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -185,10 +185,11 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: if result.returncode == 0: logger.info(f"โœ… {script_path} PASSED") return True, result.stdout - logger.error(f"โŒ {script_path} FAILED") - logger.error(f"Error output: {result.stderr}") - logger.error(f"Standard output: {result.stdout}") - return False, result.stderr + else: + logger.error(f"โŒ {script_path} FAILED") + logger.error(f"Error output: {result.stderr}") + logger.error(f"Standard output: {result.stdout}") + return False, result.stderr except subprocess.TimeoutExpired: logger.error(f"โฐ {script_path} TIMEOUT") @@ -211,11 +212,12 @@ def run_unit_tests(self) -> bool: if result.returncode == 0: logger.info("โœ… Unit tests PASSED") return True - 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}") - return False + 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}") + return False except subprocess.TimeoutExpired: logger.error("โฐ Unit tests TIMEOUT") @@ -238,10 +240,11 @@ def run_e2e_tests(self) -> bool: if result.returncode == 0: logger.info("โœ… E2E tests PASSED") return True - logger.error("โŒ E2E tests FAILED") - logger.error(f"Error output: {result.stderr}") - logger.error(f"Standard output: {result.stdout}") - return False + else: + logger.error("โŒ E2E tests FAILED") + logger.error(f"Error output: {result.stderr}") + logger.error(f"Standard output: {result.stdout}") + return False except Exception as e: logger.exception(f"๐Ÿ’ฅ E2E tests ERROR: {e}") From 85e657eeed37d95ce64f6177940e51a79bf6e89e Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:37:12 +0200 Subject: [PATCH 06/12] Fix unused variable warning in CI pipeline - Rename unused 'model' variable to '_' in _measure_model_loading_time method - Add comment explaining why model instantiation is needed (for timing measurement) - Resolve PYL-W0612 linter warning for unused variable --- scripts/ci/run_full_ci_pipeline.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 35b554ec9..478037a42 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -309,7 +309,7 @@ def _measure_model_loading_time(self) -> float: import time start_time = time.time() - model = BERTEmotionClassifier() + _ = BERTEmotionClassifier() # Instantiate model to measure loading time loading_time = time.time() - start_time logger.info(f"โœ… Model loading time: {loading_time:.2f}s") From 54a5982d7e564682e19c3744a2afcef60fa14de3 Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:38:09 +0200 Subject: [PATCH 07/12] Fix variable shadowing warnings for 'time' import - Remove redundant local 'import time' statements in _measure_model_loading_time and _measure_inference_time methods - Use module-level 'time' import instead to avoid PYL-W0621 shadowing warnings - Maintain functionality while following Python best practices for imports --- scripts/ci/run_full_ci_pipeline.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 478037a42..dd7a2f8c6 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -306,8 +306,6 @@ def _measure_model_loading_time(self) -> float: float Loading time in seconds """ - import time - start_time = time.time() _ = BERTEmotionClassifier() # Instantiate model to measure loading time loading_time = time.time() - start_time @@ -328,7 +326,6 @@ def _measure_inference_time(self, model) -> float: float Inference time in seconds """ - import time import torch start_time = time.time() From a486a57f69159aa7f0f19f2c5d0e5c94666d0596 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Thu, 25 Sep 2025 20:40:38 +0000 Subject: [PATCH 08/12] feat: add comprehensive CI pipeline orchestration system Resolved issues in scripts/ci/run_full_ci_pipeline.py with DeepSource Autofix --- scripts/ci/run_full_ci_pipeline.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index dd7a2f8c6..a223b9c9e 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -63,7 +63,8 @@ def __init__(self): "scripts/ci/onnx_conversion_test.py", ] - def _run_subprocess_command(self, command: list, timeout: int) -> subprocess.CompletedProcess: + @staticmethod + def _run_subprocess_command(command: list, timeout: int) -> subprocess.CompletedProcess: """Run a subprocess command with standardized parameters. Parameters @@ -250,7 +251,8 @@ def run_e2e_tests(self) -> bool: logger.exception(f"๐Ÿ’ฅ E2E tests ERROR: {e}") return False - def _test_gpu_model_forward_pass(self) -> bool: + @staticmethod + def _test_gpu_model_forward_pass() -> bool: """Test GPU model forward pass with BERT classifier. Returns @@ -298,7 +300,8 @@ def test_gpu_compatibility(self) -> bool: logger.exception(f"โŒ GPU compatibility test failed: {e}") return False - def _measure_model_loading_time(self) -> float: + @staticmethod + def _measure_model_loading_time() -> float: """Measure BERT model loading time. Returns @@ -313,7 +316,8 @@ def _measure_model_loading_time(self) -> float: logger.info(f"โœ… Model loading time: {loading_time:.2f}s") return loading_time - def _measure_inference_time(self, model) -> float: + @staticmethod + def _measure_inference_time(model) -> float: """Measure model inference time. Parameters @@ -337,7 +341,8 @@ def _measure_inference_time(self, model) -> float: logger.info(f"โœ… Inference time: {inference_time:.2f}s") return inference_time - def _validate_performance_thresholds(self, loading_time: float, inference_time: float) -> bool: + @staticmethod + def _validate_performance_thresholds(loading_time: float, inference_time: float) -> bool: """Validate that performance times are within acceptable thresholds. Parameters From 70b7d693c6c7cf36281eaecb1e64703df5972a4d Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:41:25 +0200 Subject: [PATCH 09/12] Fix PYL-W1203 logging format warnings - Replace f-string logging with lazy % formatting for better performance - Convert all logger calls to use % formatting instead of f-strings - Fixes 8 occurrences of formatted string passed to logging module - Maintains all logging functionality while improving performance when logging is disabled --- scripts/ci/run_full_ci_pipeline.py | 57 +++++++++++++++--------------- 1 file changed, 29 insertions(+), 28 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index a223b9c9e..4b7ab5919 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -134,7 +134,7 @@ def detect_environment(self) -> Dict[str, str]: 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: @@ -156,13 +156,13 @@ def validate_dependencies(self) -> bool: 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") @@ -170,7 +170,7 @@ def validate_dependencies(self) -> bool: 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 @@ -184,19 +184,19 @@ def run_ci_script(self, script_path: str) -> Tuple[bool, str]: ) 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(f"Standard output: {result.stdout}") + 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.exception(f"๐Ÿ’ฅ {script_path} ERROR: {e}") + logger.exception("๐Ÿ’ฅ %s ERROR: %s", script_path, e) return False, str(e) def run_unit_tests(self) -> bool: @@ -215,16 +215,16 @@ def run_unit_tests(self) -> bool: 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.exception(f"๐Ÿ’ฅ Unit tests ERROR: {e}") + logger.exception("๐Ÿ’ฅ Unit tests ERROR: %s", e) return False def run_e2e_tests(self) -> bool: @@ -243,12 +243,12 @@ def run_e2e_tests(self) -> bool: return True else: logger.error("โŒ E2E tests FAILED") - logger.error(f"Error output: {result.stderr}") - logger.error(f"Standard output: {result.stdout}") + logger.error("Error output: %s", result.stderr) + logger.error("Standard output: %s", result.stdout) return False except Exception as e: - logger.exception(f"๐Ÿ’ฅ E2E tests ERROR: {e}") + logger.exception("๐Ÿ’ฅ E2E tests ERROR: %s", e) return False @staticmethod @@ -273,11 +273,11 @@ def _test_gpu_model_forward_pass() -> bool: 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(f"โŒ GPU model forward pass failed: {e}") + logger.exception("โŒ GPU model forward pass failed: %s", e) return False def test_gpu_compatibility(self) -> bool: @@ -291,13 +291,13 @@ def test_gpu_compatibility(self) -> bool: logger.info("โ„น๏ธ No GPU available, skipping GPU tests") return True - logger.info(f"๐ŸŽฎ GPU detected: {torch.cuda.get_device_name(0)}") + 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.exception(f"โŒ GPU compatibility test failed: {e}") + logger.exception("โŒ GPU compatibility test failed: %s", e) return False @staticmethod @@ -313,7 +313,7 @@ def _measure_model_loading_time() -> float: _ = BERTEmotionClassifier() # Instantiate model to measure loading time loading_time = time.time() - start_time - logger.info(f"โœ… Model loading time: {loading_time:.2f}s") + logger.info("โœ… Model loading time: %.2fs", loading_time) return loading_time @staticmethod @@ -338,7 +338,7 @@ def _measure_inference_time(model) -> float: model(dummy_input, torch.ones_like(dummy_input)) inference_time = time.time() - start_time - logger.info(f"โœ… Inference time: {inference_time:.2f}s") + logger.info("โœ… Inference time: %.2fs", inference_time) return inference_time @staticmethod @@ -363,7 +363,8 @@ def _validate_performance_thresholds(loading_time: float, inference_time: float) return True logger.error( - f"โŒ Performance too slow - loading: {loading_time:.2f}s, inference: {inference_time:.2f}s", + "โŒ Performance too slow - loading: %.2fs, inference: %.2fs", + loading_time, inference_time ) return False @@ -385,7 +386,7 @@ def run_performance_benchmarks(self) -> bool: return self._validate_performance_thresholds(loading_time, inference_time) except Exception as e: - logger.exception(f"โŒ Performance benchmark failed: {e}") + logger.exception("โŒ Performance benchmark failed: %s", e) return False def run_pipeline_and_exit(self) -> None: @@ -422,7 +423,7 @@ def run_pipeline_and_exit(self) -> None: logger.info("โน๏ธ CI Pipeline interrupted by user") sys.exit(1) except Exception as e: - logger.exception(f"๐Ÿ’ฅ CI Pipeline crashed: {e}") + logger.exception("๐Ÿ’ฅ CI Pipeline crashed: %s", e) sys.exit(1) def run_full_pipeline(self) -> Dict[str, bool]: @@ -444,7 +445,7 @@ def run_full_pipeline(self) -> Dict[str, bool]: 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() From 3826f0209cf84ba5c9200ddd19effb280d7c457a Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:43:58 +0200 Subject: [PATCH 10/12] Fix line length violations (FLK-E501) - Break long function signatures across multiple lines - Split long string literals in report generation - Reformat multi-line logger.error calls to stay within 88 char limit - Ensure all lines comply with configured maximum line length --- scripts/ci/run_full_ci_pipeline.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 4b7ab5919..0f109036d 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -64,7 +64,9 @@ def __init__(self): ] @staticmethod - def _run_subprocess_command(command: list, timeout: int) -> subprocess.CompletedProcess: + def _run_subprocess_command( + command: list, timeout: int + ) -> subprocess.CompletedProcess: """Run a subprocess command with standardized parameters. Parameters @@ -342,7 +344,9 @@ def _measure_inference_time(model) -> float: return inference_time @staticmethod - def _validate_performance_thresholds(loading_time: float, inference_time: float) -> bool: + def _validate_performance_thresholds( + loading_time: float, inference_time: float + ) -> bool: """Validate that performance times are within acceptable thresholds. Parameters @@ -364,7 +368,8 @@ def _validate_performance_thresholds(loading_time: float, inference_time: float) logger.error( "โŒ Performance too slow - loading: %.2fs, inference: %.2fs", - loading_time, inference_time + loading_time, + inference_time, ) return False @@ -502,7 +507,9 @@ def generate_report(self) -> str: """ if total_tests == 0: - report += "โš ๏ธ No boolean tests were executed. Treating pipeline as failed.\n" + 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: From 3c06a2cd731d94a6df9f44e65425e2870b1ee02b Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:46:11 +0200 Subject: [PATCH 11/12] Fix remaining line length violation (FLK-E501) - Break long logger.info call in GPU forward pass test to stay within 88 char limit - Ensure all lines comply with maximum line length configuration --- scripts/ci/run_full_ci_pipeline.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 0f109036d..6af25823f 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -275,7 +275,9 @@ def _test_gpu_model_forward_pass() -> bool: with torch.no_grad(): output = model(dummy_input, torch.ones_like(dummy_input)) - logger.info("โœ… GPU forward pass successful, output shape: %s", output.shape) + logger.info( + "โœ… GPU forward pass successful, output shape: %s", output.shape + ) return True except Exception as e: From da6c0159aa0597edd149abf0f1e86e32e6b99320 Mon Sep 17 00:00:00 2001 From: Deniz Ulker <156104354+uelkerd@users.noreply.github.com> Date: Thu, 25 Sep 2025 22:50:10 +0200 Subject: [PATCH 12/12] Fix Python 3.9 compatibility in is_truthy function - Replace PEP 604 union syntax (str | None) with Optional[str] for Python 3.9 compatibility - Add Optional import from typing module - Maintain function behavior unchanged --- scripts/ci/run_full_ci_pipeline.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/ci/run_full_ci_pipeline.py b/scripts/ci/run_full_ci_pipeline.py index 6af25823f..cf3b46727 100644 --- a/scripts/ci/run_full_ci_pipeline.py +++ b/scripts/ci/run_full_ci_pipeline.py @@ -17,14 +17,14 @@ import sys import time from pathlib import Path -from typing import Dict, Tuple +from typing import Dict, Optional, Tuple # Use shared truthy parsing try: from src.common.env import is_truthy except ImportError: # Fallback to local helper if import path not available - def is_truthy(value: str | None) -> bool: + 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