|
#!/usr/bin/env python3 |
|
|
|
import os |
|
import gitlab |
|
import sys |
|
import argparse |
|
from typing import Dict, Any, List |
|
import warnings |
|
|
|
def get_gitlab_client() -> gitlab.Gitlab: |
|
"""Initialize and return a GitLab client.""" |
|
if 'GITLAB_TOKEN' not in os.environ: |
|
print("Error: GITLAB_TOKEN environment variable not set") |
|
sys.exit(1) |
|
|
|
if 'GITLAB_URL' not in os.environ: |
|
print("Error: GITLAB_URL environment variable not set") |
|
sys.exit(1) |
|
|
|
gl = gitlab.Gitlab(os.environ['GITLAB_URL'], private_token=os.environ['GITLAB_TOKEN']) |
|
# Enable debug mode for troubleshooting |
|
# gl.enable_debug() |
|
return gl |
|
|
|
def get_project_slack_integration(project: Any) -> Any: |
|
"""Get the current Slack integration for a project.""" |
|
try: |
|
return project.integrations.get('slack') |
|
except gitlab.exceptions.GitlabGetError: |
|
return None |
|
except AttributeError: |
|
return None |
|
|
|
def get_project_gitlab_slack_app_integration(project: Any) -> Any: |
|
"""Get the GitLab for Slack app integration for a project.""" |
|
try: |
|
return project.integrations.get('gitlab-slack-application') |
|
except gitlab.exceptions.GitlabGetError: |
|
return None |
|
except AttributeError: |
|
return None |
|
|
|
def update_project_slack_integration(project: Any, old_integration: Any, verbose: bool = False) -> None: |
|
"""Update project to use GitLab for Slack app instead of old Slack integration.""" |
|
if not old_integration: |
|
return |
|
|
|
if verbose: |
|
# Print all attributes of the old integration |
|
print(f"\nAll attributes of old Slack integration for {project.name}:") |
|
for key, value in old_integration.attributes.items(): |
|
print(f" {key}: {value}") |
|
|
|
# Prepare new integration for GitLab for Slack app |
|
new_integration = get_project_gitlab_slack_app_integration(project) |
|
if not new_integration: |
|
print(f"✗ Could not get GitLab for Slack app integration for project {project.name}") |
|
return |
|
|
|
if verbose: |
|
# Print all attributes of the new integration before update |
|
print(f"\nAll attributes of new GitLab for Slack app integration for {project.name} (before update):") |
|
for key, value in new_integration.attributes.items(): |
|
print(f" {key}: {value}") |
|
|
|
new_integration.inherited = False |
|
new_integration.use_inherited_settings = False |
|
new_integration.active = True |
|
|
|
# Copy selected attributes from old to new integration |
|
attributes_to_copy = [ |
|
'commit_events', |
|
'push_events', |
|
'issues_events', |
|
'incident_events', |
|
'alert_events', |
|
'confidential_issues_events', |
|
'merge_requests_events', |
|
'tag_push_events', |
|
'deployment_events', |
|
'note_events', |
|
'confidential_note_events', |
|
'pipeline_events', |
|
'wiki_page_events', |
|
'job_events', |
|
'comment_on_event_enabled', |
|
'vulnerability_events', |
|
] |
|
|
|
for prop in attributes_to_copy: |
|
setattr(new_integration, prop, old_integration.attributes[prop]) |
|
|
|
# Copy selected properties from old to new integration |
|
properties_to_copy = [ |
|
'channel', |
|
'notify_only_broken_pipelines', |
|
'branches_to_be_notified', |
|
'labels_to_be_notified', |
|
'labels_to_be_notified_behavior', |
|
'push_channel', |
|
'issue_channel', |
|
'confidential_issue_channel', |
|
'merge_request_channel', |
|
'note_channel', |
|
'confidential_note_channel', |
|
'tag_push_channel', |
|
'pipeline_channel', |
|
'wiki_page_channel', |
|
'deployment_channel', |
|
'incident_channel', |
|
'vulnerability_channel', |
|
'alert_channel', |
|
] |
|
|
|
for prop in properties_to_copy: |
|
setattr(new_integration, prop, old_integration.properties[prop]) |
|
|
|
if verbose: |
|
print(f"\nAttempting to save the following attributes for {project.name} (new integration):") |
|
for key, value in new_integration.attributes.items(): |
|
if value: |
|
print(f" {key}: {value}") |
|
try: |
|
new_integration.save() |
|
print(f"✓ Updated settings for project: {project.name} (new integration)") |
|
if verbose: |
|
print(f"Attributes after save (API response) for {project.name}:") |
|
for key, value in new_integration.attributes.items(): |
|
print(f" {key}: {value}") |
|
except Exception as e: |
|
print(f"✗ Failed to update project {project.name} (new integration): {str(e)}") |
|
if verbose: |
|
print("Attributes attempted:") |
|
print(new_integration.attributes) |
|
|
|
# Now delete the old Slack integration |
|
try: |
|
project.integrations.delete('slack') |
|
print(f"✓ Deleted old Slack integration for project: {project.name}") |
|
except Exception as e: |
|
print(f"✗ Failed to delete old Slack integration for project {project.name}: {str(e)}") |
|
|
|
def get_projects(gl: gitlab.Gitlab, project_id: str = None) -> List[Any]: |
|
"""Get either a single project or all projects.""" |
|
if project_id: |
|
try: |
|
return [gl.projects.get(project_id)] |
|
except gitlab.exceptions.GitlabGetError as e: |
|
print(f"Error: Could not find project with ID {project_id}") |
|
sys.exit(1) |
|
return gl.projects.list(get_all=True) |
|
|
|
def get_all_project_integrations(project: Any, verbose: bool = False) -> None: |
|
"""List all available integrations for a project.""" |
|
if not verbose: |
|
return |
|
try: |
|
integrations = project.integrations.list(get_all=True) |
|
print(f"\nAvailable integrations for {project.name}:") |
|
for integration in integrations: |
|
active_status = "✓" if integration.active else "✗" |
|
print(f" {active_status} {integration.title} ({integration.slug})") |
|
except Exception as e: |
|
print(f"Error listing integrations: {str(e)}") |
|
|
|
|
|
def main(): |
|
parser = argparse.ArgumentParser(description='Migrate GitLab Slack notification settings') |
|
parser.add_argument('--project-id', help='Optional: Specific project ID to migrate (for testing)') |
|
parser.add_argument('--verbose', action='store_true', help='Show detailed debug output') |
|
args = parser.parse_args() |
|
|
|
gl = get_gitlab_client() |
|
|
|
# Get projects (either one or all) |
|
projects = get_projects(gl, args.project_id) |
|
total_projects = len(projects) |
|
updated_count = 0 |
|
|
|
if args.project_id: |
|
print(f"Running in test mode on project ID: {args.project_id}") |
|
else: |
|
print(f"Found {total_projects} projects. Starting migration...") |
|
|
|
for i, project in enumerate(projects, 1): |
|
print(f"\nProcessing project {i}/{total_projects}: {project.name}") |
|
|
|
# List all available integrations |
|
get_all_project_integrations(project, verbose=args.verbose) |
|
|
|
# Get current Slack integration |
|
old_integration = get_project_slack_integration(project) |
|
|
|
if old_integration and getattr(old_integration, 'active', False): |
|
print(f"Found Slack integration for {project.name}") |
|
if args.verbose: |
|
print("Current settings:") |
|
for key, value in old_integration.attributes.items(): |
|
if value: # Only show non-empty settings |
|
print(f" {key}: {value}") |
|
|
|
if args.project_id: |
|
response = input("\nDo you want to proceed with the migration? (y/N): ") |
|
if response.lower() != 'y': |
|
print("Migration cancelled.") |
|
sys.exit(0) |
|
|
|
update_project_slack_integration(project, old_integration, verbose=args.verbose) |
|
updated_count += 1 |
|
else: |
|
print(f"No active Slack integration found for {project.name}") |
|
|
|
print(f"\nMigration complete! Updated {updated_count} out of {total_projects} projects.") |
|
|
|
if __name__ == "__main__": |
|
main() |