CI/CD Pipeline Development with Python

1from jenkinsapi.jenkins import Jenkins
2from jenkins import Jenkins as PythonJenkins
3import requests
4from typing import Dict, List, Optional
5import xml.etree.ElementTree as ET
6
7# jenkinsapi - Object-oriented interface
8jenkins = Jenkins('http://jenkins:8080', username='admin', password='token')
9
10# python-jenkins - Lower-level API access
11server = PythonJenkins('http://jenkins:8080', username='admin', password='token')

1class JenkinsJobManager:
2    def __init__(self, url: str, username: str, password: str):
3        self.jenkins = Jenkins(url, username=username, password=password)
4        self.server = PythonJenkins(url, username=username, password=password)
5
6    def create_pipeline_job(self, name: str,
7                           script: str,
8                           parameters: List[Dict] = None) -> bool:
9        """Create Jenkins pipeline job"""
10
11        job_config = f"""<?xml version='1.1' encoding='UTF-8'?>
12<flow-definition plugin="[email protected]">
13  <description>Automated pipeline created via Python</description>
14  <keepDependencies>false</keepDependencies>
15  <properties>
16    <hudson.model.ParametersDefinitionProperty>
17      <parameterDefinitions>
18        {self._generate_parameters_xml(parameters or [])}
19      </parameterDefinitions>
20    </hudson.model.ParametersDefinitionProperty>
21  </properties>
22  <definition class="org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition">
23    <script>{script}</script>
24    <sandbox>true</sandbox>
25  </definition>
26  <triggers/>
27  <disabled>false</disabled>
28</flow-definition>"""
29
30        try:
31            self.server.create_job(name, job_config)
32            return True
33        except Exception as e:
34            print(f"Failed to create job: {e}")
35            return False
36
37    def _generate_parameters_xml(self, parameters: List[Dict]) -> str:
38        """Generate parameter definitions XML"""
39        xml_parts = []
40        for param in parameters:
41            param_type = param.get('type', 'string')
42            if param_type == 'string':
43                xml_parts.append(f"""
44        <hudson.model.StringParameterDefinition>
45          <name>{param['name']}</name>
46          <description>{param.get('description', '')}</description>
47          <defaultValue>{param.get('default', '')}</defaultValue>
48        </hudson.model.StringParameterDefinition>""")
49            elif param_type == 'choice':
50                choices = '\n'.join(param.get('choices', []))
51                xml_parts.append(f"""
52        <hudson.model.ChoiceParameterDefinition>
53          <name>{param['name']}</name>
54          <description>{param.get('description', '')}</description>
55          <choices class="java.util.Arrays$ArrayList">
56            <a class="string-array">
57              <string>{choices}</string>
58            </a>
59          </choices>
60        </hudson.model.ChoiceParameterDefinition>""")
61
62        return '\n'.join(xml_parts)
63
64    def trigger_build(self, job_name: str,
65                     parameters: Dict = None,
66                     wait: bool = True) -> Optional[int]:
67        """Trigger Jenkins build"""
68        try:
69            job = self.jenkins[job_name]
70
71            if parameters:
72                queue_item = job.invoke(build_params=parameters)
73            else:
74                queue_item = job.invoke()
75
76            if wait:
77                queue_item.block_until_complete()
78                build = queue_item.get_build()
79                return build.get_number()
80
81            return None
82
83        except Exception as e:
84            print(f"Failed to trigger build: {e}")
85            return None
86
87    def get_build_status(self, job_name: str, build_number: int) -> Dict:
88        """Get build status and details"""
89        try:
90            job = self.jenkins[job_name]
91            build = job.get_build(build_number)
92
93            return {
94                'number': build.get_number(),
95                'status': build.get_status(),
96                'building': build.is_running(),
97                'result': build.get_result(),
98                'duration': build.get_duration().total_seconds(),
99                'timestamp': build.get_timestamp(),
100                'url': build.baseurl,
101                'console_output': build.get_console()
102            }
103
104        except Exception as e:
105            print(f"Failed to get build status: {e}")
106            return {}
107
108    def get_build_artifacts(self, job_name: str,
109                           build_number: int) -> List[str]:
110        """Get build artifacts"""
111        try:
112            job = self.jenkins[job_name]
113            build = job.get_build(build_number)
114
115            artifacts = []
116            for artifact in build.get_artifact_dict():
117                artifacts.append({
118                    'filename': artifact['fileName'],
119                    'relativePath': artifact['relativePath'],
120                    'url': f"{build.baseurl}artifact/{artifact['relativePath']}"
121                })
122
123            return artifacts
124
125        except Exception as e:
126            print(f"Failed to get artifacts: {e}")
127            return []
128
129    def get_pipeline_stages(self, job_name: str,
130                           build_number: int) -> List[Dict]:
131        """Get pipeline stage information"""
132        try:
133            # Use workflow API to get stage details
134            url = f"{self.jenkins.baseurl}/job/{job_name}/{build_number}/wfapi/describe"
135            response = requests.get(url, auth=(self.jenkins.username,
136                                              self.jenkins.password))
137
138            if response.status_code == 200:
139                data = response.json()
140                stages = []
141
142                for stage in data.get('stages', []):
143                    stages.append({
144                        'name': stage['name'],
145                        'status': stage['status'],
146                        'duration': stage['durationMillis'],
147                        'start_time': stage['startTimeMillis']
148                    })
149
150                return stages
151
152            return []
153
154        except Exception as e:
155            print(f"Failed to get pipeline stages: {e}")
156            return []

1def manage_build_queue(jenkins: Jenkins) -> Dict:
2    """Monitor and manage Jenkins build queue"""
3    queue = jenkins.get_queue()
4
5    queue_info = {
6        'length': len(queue),
7        'items': []
8    }
9
10    for item in queue.get_queue_items():
11        queue_info['items'].append({
12            'id': item.id,
13            'task_name': item.get_job_name(),
14            'why': item.why,
15            'blocked': item.blocked,
16            'stuck': item.stuck,
17            'in_queue_since': item.get_time_in_queue()
18        })
19
20    return queue_info
21
22def cancel_queued_builds(jenkins: Jenkins, job_name: str):
23    """Cancel all queued builds for a job"""
24    queue = jenkins.get_queue()
25
26    for item in queue.get_queue_items():
27        if item.get_job_name() == job_name:
28            item.cancel()
29            print(f"Cancelled queued build: {item.id}")

1def get_node_status(jenkins: Jenkins) -> Dict:
2    """Get Jenkins node/executor status"""
3    nodes_info = {
4        'total_nodes': 0,
5        'online_nodes': 0,
6        'offline_nodes': 0,
7        'nodes': []
8    }
9
10    nodes = jenkins.get_nodes()
11    nodes_info['total_nodes'] = len(nodes)
12
13    for node_name, node in nodes.items():
14        is_online = node.is_online()
15
16        node_data = {
17            'name': node_name,
18            'online': is_online,
19            'executors': node.get_num_executors(),
20            'labels': list(node.get_labels()),
21            'architecture': node.get_architecture()
22        }
23
24        nodes_info['nodes'].append(node_data)
25
26        if is_online:
27            nodes_info['online_nodes'] += 1
28        else:
29            nodes_info['offline_nodes'] += 1
30
31    return nodes_info

1from github import Github, GithubException
2from github.Repository import Repository
3from github.Workflow import Workflow
4from github.WorkflowRun import WorkflowRun
5import yaml
6from typing import Dict, List, Optional
7from datetime import datetime, timedelta
8
9class GitHubActionsManager:
10    def __init__(self, token: str):
11        self.github = Github(token)
12        self.user = self.github.get_user()
13
14    def create_workflow_file(self, repo_name: str,
15                            workflow_name: str,
16                            workflow_content: Dict) -> bool:
17        """Create GitHub Actions workflow file"""
18        try:
19            repo = self.github.get_repo(repo_name)
20
21            # Convert workflow dict to YAML
22            workflow_yaml = yaml.dump(workflow_content,
23                                     default_flow_style=False)
24
25            # Create workflow file
26            file_path = f".github/workflows/{workflow_name}.yml"
27
28            try:
29                # Check if file exists
30                contents = repo.get_contents(file_path)
31                # Update existing file
32                repo.update_file(
33                    path=file_path,
34                    message=f"Update {workflow_name} workflow",
35                    content=workflow_yaml,
36                    sha=contents.sha
37                )
38            except GithubException:
39                # Create new file
40                repo.create_file(
41                    path=file_path,
42                    message=f"Add {workflow_name} workflow",
43                    content=workflow_yaml
44                )
45
46            return True
47
48        except Exception as e:
49            print(f"Failed to create workflow: {e}")
50            return False
51
52    def trigger_workflow(self, repo_name: str,
53                        workflow_id: str,
54                        ref: str = 'main',
55                        inputs: Dict = None) -> bool:
56        """Trigger GitHub Actions workflow"""
57        try:
58            repo = self.github.get_repo(repo_name)
59            workflow = repo.get_workflow(workflow_id)
60
61            if inputs:
62                success = workflow.create_dispatch(ref=ref, inputs=inputs)
63            else:
64                success = workflow.create_dispatch(ref=ref)
65
66            return success
67
68        except Exception as e:
69            print(f"Failed to trigger workflow: {e}")
70            return False
71
72    def get_workflow_runs(self, repo_name: str,
73                         workflow_id: str,
74                         limit: int = 10) -> List[Dict]:
75        """Get recent workflow runs"""
76        try:
77            repo = self.github.get_repo(repo_name)
78            workflow = repo.get_workflow(workflow_id)
79
80            runs = []
81            for run in workflow.get_runs()[:limit]:
82                runs.append({
83                    'id': run.id,
84                    'name': run.name,
85                    'status': run.status,
86                    'conclusion': run.conclusion,
87                    'branch': run.head_branch,
88                    'commit_sha': run.head_sha[:7],
89                    'created_at': run.created_at,
90                    'updated_at': run.updated_at,
91                    'duration': self._calculate_duration(run),
92                    'url': run.html_url
93                })
94
95            return runs
96
97        except Exception as e:
98            print(f"Failed to get workflow runs: {e}")
99            return []
100
101    def _calculate_duration(self, run: WorkflowRun) -> Optional[int]:
102        """Calculate workflow run duration in seconds"""
103        if run.status == 'completed' and run.created_at and run.updated_at:
104            duration = (run.updated_at - run.created_at).total_seconds()
105            return int(duration)
106        return None
107
108    def get_workflow_run_jobs(self, repo_name: str,
109                             run_id: int) -> List[Dict]:
110        """Get jobs for a workflow run"""
111        try:
112            repo = self.github.get_repo(repo_name)
113            run = repo.get_workflow_run(run_id)
114
115            jobs = []
116            for job in run.jobs():
117                jobs.append({
118                    'id': job.id,
119                    'name': job.name,
120                    'status': job.status,
121                    'conclusion': job.conclusion,
122                    'started_at': job.started_at,
123                    'completed_at': job.completed_at,
124                    'steps': [
125                        {
126                            'name': step.name,
127                            'status': step.status,
128                            'conclusion': step.conclusion,
129                            'number': step.number
130                        }
131                        for step in job.steps
132                    ]
133                })
134
135            return jobs
136
137        except Exception as e:
138            print(f"Failed to get workflow jobs: {e}")
139            return []
140
141    def cancel_workflow_run(self, repo_name: str, run_id: int) -> bool:
142        """Cancel a workflow run"""
143        try:
144            repo = self.github.get_repo(repo_name)
145            run = repo.get_workflow_run(run_id)
146
147            if run.status != 'completed':
148                run.cancel()
149                return True
150
151            return False
152
153        except Exception as e:
154            print(f"Failed to cancel workflow: {e}")
155            return False
156
157    def get_workflow_metrics(self, repo_name: str,
158                           workflow_id: str,
159                           days: int = 30) -> Dict:
160        """Calculate workflow metrics"""
161        try:
162            repo = self.github.get_repo(repo_name)
163            workflow = repo.get_workflow(workflow_id)
164
165            since = datetime.now() - timedelta(days=days)
166            runs = [r for r in workflow.get_runs()
167                   if r.created_at > since]
168
169            total_runs = len(runs)
170            successful_runs = len([r for r in runs
171                                  if r.conclusion == 'success'])
172            failed_runs = len([r for r in runs
173                              if r.conclusion == 'failure'])
174
175            durations = [
176                (r.updated_at - r.created_at).total_seconds()
177                for r in runs
178                if r.status == 'completed' and r.created_at and r.updated_at
179            ]
180
181            metrics = {
182                'total_runs': total_runs,
183                'successful_runs': successful_runs,
184                'failed_runs': failed_runs,
185                'success_rate': (successful_runs / total_runs * 100)
186                               if total_runs > 0 else 0,
187                'avg_duration': sum(durations) / len(durations)
188                               if durations else 0,
189                'min_duration': min(durations) if durations else 0,
190                'max_duration': max(durations) if durations else 0
191            }
192
193            return metrics
194
195        except Exception as e:
196            print(f"Failed to calculate metrics: {e}")
197            return {}

1def setup_repository_automation(github: Github,
2                               repo_name: str) -> bool:
3    """Setup comprehensive repository automation"""
4    try:
5        repo = github.get_repo(repo_name)
6
7        # Enable branch protection
8        branch = repo.get_branch('main')
9        branch.edit_protection(
10            required_approving_review_count=2,
11            dismiss_stale_reviews=True,
12            require_code_owner_reviews=True,
13            required_status_checks={
14                'strict': True,
15                'contexts': ['ci/build', 'ci/test']
16            },
17            enforce_admins=True
18        )
19
20        # Setup environment secrets (requires appropriate permissions)
21        # Note: Actual secret values should come from secure sources
22
23        # Create issue labels
24        labels = [
25            {'name': 'bug', 'color': 'ee0701'},
26            {'name': 'enhancement', 'color': '0052cc'},
27            {'name': 'documentation', 'color': '5319e7'}
28        ]
29
30        for label in labels:
31            try:
32                repo.create_label(**label)
33            except GithubException:
34                pass  # Label already exists
35
36        return True
37
38    except Exception as e:
39        print(f"Failed to setup automation: {e}")
40        return False

1def automate_pr_checks(repo: Repository, pr_number: int) -> Dict:
2    """Perform automated PR checks"""
3    pr = repo.get_pull(pr_number)
4
5    checks = {
6        'has_description': len(pr.body or '') > 50,
7        'has_reviewers': pr.requested_reviewers.totalCount > 0,
8        'has_labels': pr.labels.totalCount > 0,
9        'files_changed': pr.changed_files,
10        'additions': pr.additions,
11        'deletions': pr.deletions,
12        'mergeable': pr.mergeable,
13        'ci_passing': all(
14            status.state == 'success'
15            for status in pr.get_commits()[pr.commits - 1].get_statuses()
16        )
17    }
18
19    # Auto-add labels based on changes
20    if checks['files_changed'] > 100:
21        pr.add_to_labels('large-pr')
22
23    if checks['additions'] > checks['deletions'] * 3:
24        pr.add_to_labels('significant-change')
25
26    return checks

1from abc import ABC, abstractmethod
2from enum import Enum
3from typing import Dict, List, Optional, Any
4from dataclasses import dataclass
5import time
6
7class PipelineStatus(Enum):
8    """Pipeline execution status"""
9    PENDING = "pending"
10    RUNNING = "running"
11    SUCCESS = "success"
12    FAILURE = "failure"
13    CANCELLED = "cancelled"
14
15
16@dataclass
17class PipelineResult:
18    """Pipeline execution result"""
19    pipeline_id: str
20    platform: str
21    status: PipelineStatus
22    duration: float
23    build_number: Optional[int] = None
24    url: Optional[str] = None
25    artifacts: List[str] = None
26    error: Optional[str] = None
27
28
29class CIPlatform(ABC):
30    """Abstract base class for CI platforms"""
31
32    @abstractmethod
33    def trigger_pipeline(self, config: Dict) -> str:
34        """Trigger pipeline execution"""
35        pass
36
37    @abstractmethod
38    def get_pipeline_status(self, pipeline_id: str) -> PipelineStatus:
39        """Get pipeline status"""
40        pass
41
42    @abstractmethod
43    def get_pipeline_result(self, pipeline_id: str) -> PipelineResult:
44        """Get pipeline execution result"""
45        pass
46
47    @abstractmethod
48    def cancel_pipeline(self, pipeline_id: str) -> bool:
49        """Cancel running pipeline"""
50        pass
51
52
53class JenkinsPlatform(CIPlatform):
54    """Jenkins CI platform implementation"""
55
56    def __init__(self, jenkins_manager: 'JenkinsJobManager'):
57        self.jenkins = jenkins_manager
58
59    def trigger_pipeline(self, config: Dict) -> str:
60        """Trigger Jenkins pipeline"""
61        job_name = config['job_name']
62        parameters = config.get('parameters', {})
63
64        build_number = self.jenkins.trigger_build(
65            job_name,
66            parameters,
67            wait=False
68        )
69
70        return f"{job_name}:{build_number}"
71
72    def get_pipeline_status(self, pipeline_id: str) -> PipelineStatus:
73        """Get Jenkins pipeline status"""
74        job_name, build_number = pipeline_id.split(':')
75        status = self.jenkins.get_build_status(job_name, int(build_number))
76
77        if status['building']:
78            return PipelineStatus.RUNNING
79        elif status['result'] == 'SUCCESS':
80            return PipelineStatus.SUCCESS
81        elif status['result'] == 'FAILURE':
82            return PipelineStatus.FAILURE
83        else:
84            return PipelineStatus.PENDING
85
86    def get_pipeline_result(self, pipeline_id: str) -> PipelineResult:
87        """Get Jenkins pipeline result"""
88        job_name, build_number = pipeline_id.split(':')
89        status = self.jenkins.get_build_status(job_name, int(build_number))
90        artifacts = self.jenkins.get_build_artifacts(job_name, int(build_number))
91
92        return PipelineResult(
93            pipeline_id=pipeline_id,
94            platform='jenkins',
95            status=self.get_pipeline_status(pipeline_id),
96            duration=status.get('duration', 0),
97            build_number=int(build_number),
98            url=status.get('url'),
99            artifacts=[a['url'] for a in artifacts]
100        )
101
102    def cancel_pipeline(self, pipeline_id: str) -> bool:
103        """Cancel Jenkins pipeline"""
104        job_name, build_number = pipeline_id.split(':')
105        # Implementation depends on Jenkins API capabilities
106        return True
107
108
109class GitHubActionsPlatform(CIPlatform):
110    """GitHub Actions platform implementation"""
111
112    def __init__(self, github_manager: 'GitHubActionsManager'):
113        self.github = github_manager
114
115    def trigger_pipeline(self, config: Dict) -> str:
116        """Trigger GitHub Actions workflow"""
117        repo_name = config['repo_name']
118        workflow_id = config['workflow_id']
119        inputs = config.get('inputs', {})
120        ref = config.get('ref', 'main')
121
122        success = self.github.trigger_workflow(
123            repo_name,
124            workflow_id,
125            ref,
126            inputs
127        )
128
129        if success:
130            # Get the latest run ID
131            runs = self.github.get_workflow_runs(repo_name, workflow_id, limit=1)
132            if runs:
133                return f"{repo_name}:{runs[0]['id']}"
134
135        raise Exception("Failed to trigger workflow")
136
137    def get_pipeline_status(self, pipeline_id: str) -> PipelineStatus:
138        """Get GitHub Actions status"""
139        repo_name, run_id = pipeline_id.rsplit(':', 1)
140        # Fetch run status from GitHub API
141        # Implementation would use github_manager methods
142        return PipelineStatus.RUNNING  # Placeholder
143
144    def get_pipeline_result(self, pipeline_id: str) -> PipelineResult:
145        """Get GitHub Actions result"""
146        repo_name, run_id = pipeline_id.rsplit(':', 1)
147        # Fetch complete result
148        return PipelineResult(
149            pipeline_id=pipeline_id,
150            platform='github-actions',
151            status=PipelineStatus.SUCCESS,
152            duration=0
153        )
154
155    def cancel_pipeline(self, pipeline_id: str) -> bool:
156        """Cancel GitHub Actions workflow"""
157        repo_name, run_id = pipeline_id.rsplit(':', 1)
158        return self.github.cancel_workflow_run(repo_name, int(run_id))
159
160
161class PipelineOrchestrator:
162    """Unified pipeline orchestrator across platforms"""
163
164    def __init__(self):
165        self.platforms: Dict[str, CIPlatform] = {}
166        self.active_pipelines: Dict[str, PipelineResult] = {}
167
168    def register_platform(self, name: str, platform: CIPlatform):
169        """Register CI platform"""
170        self.platforms[name] = platform
171
172    def execute_pipeline(self, platform: str, config: Dict,
173                        wait: bool = True) -> PipelineResult:
174        """Execute pipeline on specified platform"""
175        if platform not in self.platforms:
176            raise ValueError(f"Unknown platform: {platform}")
177
178        ci_platform = self.platforms[platform]
179
180        # Trigger pipeline
181        pipeline_id = ci_platform.trigger_pipeline(config)
182
183        # Track active pipeline
184        result = PipelineResult(
185            pipeline_id=pipeline_id,
186            platform=platform,
187            status=PipelineStatus.PENDING,
188            duration=0
189        )
190        self.active_pipelines[pipeline_id] = result
191
192        # Wait for completion if requested
193        if wait:
194            result = self._wait_for_completion(
195                platform,
196                pipeline_id,
197                timeout=3600
198            )
199
200        return result
201
202    def _wait_for_completion(self, platform: str,
203                            pipeline_id: str,
204                            timeout: int = 3600) -> PipelineResult:
205        """Wait for pipeline completion"""
206        ci_platform = self.platforms[platform]
207        start_time = time.time()
208
209        while time.time() - start_time < timeout:
210            status = ci_platform.get_pipeline_status(pipeline_id)
211
212            if status in [PipelineStatus.SUCCESS,
213                         PipelineStatus.FAILURE,
214                         PipelineStatus.CANCELLED]:
215                return ci_platform.get_pipeline_result(pipeline_id)
216
217            time.sleep(10)
218
219        raise TimeoutError(f"Pipeline {pipeline_id} timeout after {timeout}s")
220
221    def execute_multi_platform(self, configs: List[Dict]) -> List[PipelineResult]:
222        """Execute pipelines across multiple platforms"""
223        results = []
224
225        for config in configs:
226            platform = config.pop('platform')
227            result = self.execute_pipeline(platform, config, wait=False)
228            results.append(result)
229
230        # Wait for all to complete
231        final_results = []
232        for result in results:
233            final_result = self._wait_for_completion(
234                result.platform,
235                result.pipeline_id
236            )
237            final_results.append(final_result)
238
239        return final_results

1class DeploymentPipeline:
2    """Advanced deployment pipeline with gates"""
3
4    def __init__(self, orchestrator: PipelineOrchestrator):
5        self.orchestrator = orchestrator
6        self.deployment_history = []
7
8    def deploy_with_gates(self, environment: str,
9                         version: str) -> PipelineResult:
10        """Deploy with automated quality gates"""
11
12        # Stage 1: Build and test
13        build_result = self.orchestrator.execute_pipeline('jenkins', {
14            'job_name': f'build-{environment}',
15            'parameters': {'VERSION': version}
16        })
17
18        if build_result.status != PipelineStatus.SUCCESS:
19            raise Exception("Build failed")
20
21        # Stage 2: Integration tests
22        test_result = self.orchestrator.execute_pipeline('github-actions', {
23            'repo_name': 'org/app',
24            'workflow_id': 'integration-tests.yml',
25            'inputs': {'version': version}
26        })
27
28        if test_result.status != PipelineStatus.SUCCESS:
29            raise Exception("Integration tests failed")
30
31        # Stage 3: Deploy to staging
32        staging_result = self._deploy_to_environment('staging', version)
33
34        # Stage 4: Smoke tests
35        smoke_passed = self._run_smoke_tests('staging')
36
37        if not smoke_passed:
38            # Rollback staging
39            self._rollback_environment('staging')
40            raise Exception("Smoke tests failed")
41
42        # Stage 5: Manual approval for production
43        if environment == 'production':
44            approved = self._wait_for_approval(version)
45            if not approved:
46                raise Exception("Deployment not approved")
47
48        # Stage 6: Deploy to production
49        prod_result = self._deploy_to_environment('production', version)
50
51        # Record deployment
52        self.deployment_history.append({
53            'environment': environment,
54            'version': version,
55            'timestamp': time.time(),
56            'result': prod_result
57        })
58
59        return prod_result
60
61    def _deploy_to_environment(self, env: str, version: str) -> PipelineResult:
62        """Deploy to specific environment"""
63        # Trigger deployment pipeline
64        return self.orchestrator.execute_pipeline('jenkins', {
65            'job_name': f'deploy-{env}',
66            'parameters': {
67                'VERSION': version,
68                'ENVIRONMENT': env
69            }
70        })
71
72    def _run_smoke_tests(self, env: str) -> bool:
73        """Run smoke tests"""
74        # Trigger smoke test suite
75        result = self.orchestrator.execute_pipeline('github-actions', {
76            'repo_name': 'org/tests',
77            'workflow_id': 'smoke-tests.yml',
78            'inputs': {'environment': env}
79        })
80        return result.status == PipelineStatus.SUCCESS
81
82    def _wait_for_approval(self, version: str, timeout: int = 86400) -> bool:
83        """Wait for manual approval"""
84        # In production, this would integrate with approval systems
85        # like Slack, PagerDuty, or custom approval workflows
86        print(f"Waiting for approval for version {version}")
87        # Placeholder - would check approval status
88        return True
89
90    def _rollback_environment(self, env: str):
91        """Rollback to previous version"""
92        if not self.deployment_history:
93            raise Exception("No previous deployment to rollback to")
94
95        # Get previous successful deployment
96        previous = self.deployment_history[-2] if len(self.deployment_history) > 1 else None
97
98        if previous:
99            self._deploy_to_environment(env, previous['version'])