""" Git operations for release management. This module handles all Git-related operations needed for releases. """ import subprocess from pathlib import Path from typing import Dict, Any, Optional, List class GitManager: """Manages Git operations for releases.""" def __init__(self, project_root: Optional[Path] = None, dry_run: bool = False): """Initialize Git manager. Args: project_root: Root directory of the project dry_run: If True, show what would be done without executing """ self.project_root = project_root or Path.cwd() self.dry_run = dry_run def get_repository_status(self) -> Dict[str, Any]: """Get current git repository status. Returns: Dictionary with repository status information """ try: # Get current branch branch_result = self._run_command(['git', 'branch', '--show-current']) current_branch = branch_result.stdout.strip() # Check for uncommitted changes status_result = self._run_command(['git', 'status', '--porcelain']) has_changes = bool(status_result.stdout.strip()) # Get latest commit commit_result = self._run_command(['git', 'rev-parse', '--short', 'HEAD']) latest_commit = commit_result.stdout.strip() # Get latest tag try: tag_result = self._run_command(['git', 'describe', '--tags', '--abbrev=0']) latest_tag = tag_result.stdout.strip() except subprocess.CalledProcessError: latest_tag = None # Get unpushed tags unpushed_tags = self.get_unpushed_tags() return { 'is_repo': True, 'branch': current_branch, 'has_changes': has_changes, 'latest_commit': latest_commit, 'latest_tag': latest_tag, 'unpushed_tags': unpushed_tags } except subprocess.CalledProcessError: return {'is_repo': False} def create_tag(self, version: str, message: Optional[str] = None, push: bool = True) -> bool: """Create and optionally push git tag. Args: version: Version to tag (e.g., "1.0.0") message: Optional tag message push: Whether to push the tag to origin (default: True) Returns: True if successful, False otherwise """ if not version.startswith('v'): tag_name = f"v{version}" else: tag_name = version tag_message = message or f"Release {version.lstrip('v')}" print(f"🏷️ Creating git tag {tag_name}") try: # Create annotated tag self._run_command(['git', 'tag', '-a', tag_name, '-m', tag_message]) print(f"✅ Tag {tag_name} created") # Push tag to origin if requested if push: try: print(f"📤 Pushing tag to origin...") self._run_command(['git', 'push', 'origin', tag_name]) print(f"✅ Tag pushed to origin") return True except subprocess.CalledProcessError as e: print(f"⚠️ Could not push tag to origin: {e}") print(f"You can push it manually with: git push origin {tag_name}") return True # Tag created successfully, push can be done manually else: return True # Tag created successfully, user chose not to push except subprocess.CalledProcessError as e: print(f"❌ Failed to create tag: {e}") return False def validate_release_state(self, force: bool = False) -> tuple[bool, List[str]]: """Validate that repository is ready for release. Args: force: Skip validation checks if True Returns: Tuple of (is_valid, list_of_issues) """ issues = [] status = self.get_repository_status() if not status['is_repo']: issues.append("Not in a git repository") else: if status['has_changes'] and not force: issues.append("Repository has uncommitted changes") if status['branch'] != 'main' and not force: issues.append(f"Not on main branch (currently on {status['branch']})") return len(issues) == 0, issues def get_commits_since_tag(self, tag_name: Optional[str] = None) -> List[str]: """Get list of commits since specified tag. Args: tag_name: Tag to compare against. If None, uses latest tag. Returns: List of commit messages since tag """ try: if tag_name is None: # Get latest tag tag_result = self._run_command(['git', 'describe', '--tags', '--abbrev=0']) tag_name = tag_result.stdout.strip() # Get commits since tag log_result = self._run_command([ 'git', 'log', f'{tag_name}..HEAD', '--oneline', '--no-merges' ]) commits = [] for line in log_result.stdout.strip().split('\n'): if line: commits.append(line) return commits except subprocess.CalledProcessError: return [] def tag_exists(self, tag_name: str) -> bool: """Check if a git tag exists. Args: tag_name: Tag name to check Returns: True if tag exists, False otherwise """ try: self._run_command(['git', 'rev-parse', f'refs/tags/{tag_name}']) return True except subprocess.CalledProcessError: return False def get_remote_url(self, remote: str = 'origin') -> Optional[str]: """Get the URL of a git remote. Args: remote: Remote name (default: 'origin') Returns: Remote URL or None if not found """ try: result = self._run_command(['git', 'remote', 'get-url', remote]) return result.stdout.strip() except subprocess.CalledProcessError: return None def get_unpushed_tags(self, remote: str = 'origin') -> List[str]: """Get list of tags that exist locally but not on remote. Args: remote: Remote name to compare against (default: 'origin') Returns: List of unpushed tag names """ try: # Get local tags local_result = self._run_command(['git', 'tag', '-l']) local_tags = set(tag.strip() for tag in local_result.stdout.strip().split('\n') if tag.strip()) # Get remote tags try: remote_result = self._run_command(['git', 'ls-remote', '--tags', remote]) remote_lines = remote_result.stdout.strip().split('\n') # Parse remote tags (format: "hash refs/tags/tagname") remote_tags = set() for line in remote_lines: if not line: continue parts = line.split('refs/tags/') if len(parts) > 1: # Remove ^{} suffix for annotated tags tag_name = parts[1].replace('^{}', '') remote_tags.add(tag_name) # Find tags that are local but not remote unpushed = sorted(local_tags - remote_tags) return unpushed except subprocess.CalledProcessError: # Remote not available, assume all tags are unpushed return sorted(local_tags) except subprocess.CalledProcessError: return [] def _run_command(self, cmd: List[str]) -> subprocess.CompletedProcess: """Run a git command. Args: cmd: Command to execute Returns: CompletedProcess result Raises: subprocess.CalledProcessError: If command fails """ if self.dry_run and not any(read_only in cmd for read_only in ['status', 'branch', 'rev-parse', 'describe', 'log', 'remote']): print(f"[DRY RUN] Would run: {' '.join(cmd)}") return subprocess.CompletedProcess(cmd, 0, "", "") return subprocess.run( cmd, capture_output=True, text=True, check=True, cwd=self.project_root )