Run the complete migration process.
(self)
| 60 | ] |
| 61 | |
| 62 | def run_migration(self) -> bool: |
| 63 | """Run the complete migration process.""" |
| 64 | print("Starting Legacy Workflow Migration") |
| 65 | print("=" * 60) |
| 66 | |
| 67 | if self.dry_run: |
| 68 | print("[DRY RUN] MODE - No changes will be made") |
| 69 | print() |
| 70 | |
| 71 | # Load previous migration state if exists |
| 72 | self._load_migration_state() |
| 73 | |
| 74 | all_passed = True |
| 75 | |
| 76 | for step in self.steps: |
| 77 | if step.completed: |
| 78 | print(f"[SKIP] {step.name} (already completed)") |
| 79 | continue |
| 80 | |
| 81 | print(f"\nExecuting: {step.description}") |
| 82 | |
| 83 | try: |
| 84 | success = self._execute_step(step) |
| 85 | |
| 86 | if success: |
| 87 | step.completed = True |
| 88 | print(f"[PASS] {step.name}") |
| 89 | else: |
| 90 | print(f"[FAIL] {step.name}") |
| 91 | if step.error: |
| 92 | print(f" Error: {step.error}") |
| 93 | |
| 94 | if step.required: |
| 95 | all_passed = False |
| 96 | print("[CRITICAL] Required step failed. Migration cannot continue.") |
| 97 | break |
| 98 | |
| 99 | except Exception as e: |
| 100 | step.error = str(e) |
| 101 | print(f"[FAIL] {step.name}") |
| 102 | print(f" Exception: {e}") |
| 103 | |
| 104 | if step.required: |
| 105 | all_passed = False |
| 106 | print("[CRITICAL] Required step failed. Migration cannot continue.") |
| 107 | break |
| 108 | |
| 109 | # Save migration state after each step |
| 110 | self._save_migration_state() |
| 111 | |
| 112 | self._generate_migration_report(all_passed) |
| 113 | return all_passed |
| 114 | |
| 115 | def _execute_step(self, step: MigrationStep) -> bool: |
| 116 | """Execute a single migration step.""" |
no test coverage detected