Coverage for core/src/version_finder/git_executer.py: 75%

1""" 

2git_executor.py 

3==================================== 

4Module for handling git command execution logic. 

5""" 

6from dataclasses import dataclass 

7from pathlib import Path 

8import subprocess 

9import time 

10import os 

11from typing import Optional, Union 

12from version_finder.logger import get_logger 

13from version_finder.common import ( 

14 DEFAULT_GIT_TIMEOUT, 

15 DEFAULT_GIT_MAX_RETRIES, 

16 DEFAULT_GIT_RETRY_DELAY, 

17 ENV_GIT_TIMEOUT, 

18 ENV_GIT_MAX_RETRIES, 

19 ENV_GIT_RETRY_DELAY 

20) 

21 

22 

23logger = get_logger() 

24 

25 

26@dataclass 

27class GitConfig: 

28 """Configuration settings for git operations""" 

29 timeout: int = int(os.environ.get(ENV_GIT_TIMEOUT, str(DEFAULT_GIT_TIMEOUT))) 

30 max_retries: int = int(os.environ.get(ENV_GIT_MAX_RETRIES, str(DEFAULT_GIT_MAX_RETRIES))) 

31 retry_delay: int = int(os.environ.get(ENV_GIT_RETRY_DELAY, str(DEFAULT_GIT_RETRY_DELAY))) 

32 

33 def __post_init__(self): 

34 if self.timeout <= 0: 

35 raise ValueError("timeout must be positive") 

36 if self.max_retries < 0: 

37 raise ValueError("max_retries cannot be negative") 

38 if self.retry_delay <= 0: 

39 raise ValueError("retry_delay must be positive") 

40 

41 

42class GitCommandError(Exception): 

43 """Base exception for git command failures""" 

44 

45 

46class GitNetworkError(GitCommandError): 

47 """Raised when git operations fail due to network issues""" 

48 

49 

50class GitTimeoutError(GitCommandError): 

51 """Raised when git operations timeout""" 

52 

53 

54class GitPermissionError(GitCommandError): 

55 """Raised when git operations fail due to permission issues""" 

56 

57 

58class GitCommandExecutor: 

59 def __init__(self, 

60 repository_path: Path, 

61 config: Optional[GitConfig] = None): 

62 self.repository_path = repository_path 

63 self.config = config or GitConfig() 

64 

65 # Check Git is installed 

66 try: 

67 subprocess.check_output(["git", "--version"]) 

68 except FileNotFoundError: 

69 raise GitCommandError("Git is not installed") 

70 

71 def execute(self, command: list[str], retries: int = 0, 

72 check: bool = True) -> Union[bytes, subprocess.CompletedProcess]: 

73 """ 

74 Execute a git command with retry logic and timeout. 

75 

76 Args: 

77 command: Git command and arguments as list 

78 retries: Number of retries attempted so far 

79 check: Whether to check return code and raise on error 

80 

81 Returns: 

82 Command output as bytes or CompletedProcess if check=False 

83 

84 Raises: 

85 GitCommandError: Base exception for command failures 

86 GitNetworkError: When network-related errors occur 

87 GitTimeoutError: When command execution times out 

88 GitPermissionError: When permission issues occur 

89 """ 

90 try: 

91 logger.debug(f"Executing git command: {' '.join(command)}") 

92 output = subprocess.check_output( 

93 ["git"] + command, 

94 cwd=self.repository_path, 

95 stderr=subprocess.PIPE, 

96 timeout=self.config.timeout 

97 ) 

98 return output 

99 except subprocess.TimeoutExpired as e: 

100 if not check: 

101 return subprocess.CompletedProcess(args=["git"] + 

102 command, returncode=1, stdout=b"", stderr=str(e).encode()) 

103 

104 if retries < self.config.max_retries: 

105 logger.warning(f"Git command timed out, retrying in {self.config.retry_delay}s: {e}") 

106 time.sleep(self.config.retry_delay) 

107 return self.execute(command, retries + 1) 

108 

109 raise GitTimeoutError(f"Git command timed out after {self.config.timeout}s: {' '.join(command)}") from e 

110 

111 except subprocess.CalledProcessError as e: 

112 if not check: 

113 return e 

114 

115 error_msg = e.stderr.decode('utf-8', errors='replace') 

116 

117 # Handle specific error types 

118 if any( 

119 net_err in error_msg for net_err in [ 

120 'could not resolve host', 

121 'Connection refused', 

122 'Connection timed out']): 

123 raise GitNetworkError(f"Network error during git operation: {error_msg}") from e 

124 

125 if any(perm_err in error_msg for perm_err in ['Permission denied', 'authentication failed']): 

126 raise GitPermissionError(f"Permission error during git operation: {error_msg}") from e 

127 

128 if retries < self.config.max_retries: 

129 logger.warning(f"Git command failed, retrying in {self.config.retry_delay}s: {error_msg}") 

130 time.sleep(self.config.retry_delay) 

131 return self.execute(command, retries + 1) 

132 

133 raise GitCommandError(f"Git command failed: {error_msg}") from e