Print current pipeline status.
(self)
| 252 | print("Pipeline reset successfully") |
| 253 | |
| 254 | def print_status(self): |
| 255 | """Print current pipeline status.""" |
| 256 | status = self.get_pipeline_status() |
| 257 | |
| 258 | print("=" * 60) |
| 259 | print("WORKFLOW PIPELINE STATUS") |
| 260 | print("=" * 60) |
| 261 | print(f"Overall Status: {status['overall_status'].upper()}") |
| 262 | print(f"Progress: {status['completed_phases']}/{status['total_phases']} phases completed") |
| 263 | |
| 264 | if status["current_phase"]: |
| 265 | print(f"Current Phase: {status['current_phase']}") |
| 266 | |
| 267 | if status["next_phase"]: |
| 268 | print(f"Next Phase: {status['next_phase']}") |
| 269 | |
| 270 | print("\nPhase Details:") |
| 271 | for phase_name, phase_info in status["phases"].items(): |
| 272 | status_icon = {"success": "✅", "failure": "❌", "running": "🔄", "pending": "⏳", "not_started": "⚪"}.get(phase_info["status"], "❓") |
| 273 | |
| 274 | duration_text = f" ({phase_info['duration']:.1f}s)" if phase_info["duration"] else "" |
| 275 | artifacts_text = f" [{phase_info['artifacts_count']} artifacts]" if phase_info["artifacts_count"] > 0 else "" |
| 276 | |
| 277 | print(f" {status_icon} {phase_name}: {phase_info['status']}{duration_text}{artifacts_text}") |
| 278 | |
| 279 | |
| 280 | def main(): |
no test coverage detected