Coverage for app/services/git_service.py: 90.77%

1"""Git service for repository management with HTTPS token authentication.""" 

2 

3import logging 

4import re 

5import shutil 

6from pathlib import Path 

7 

8import git 

9 

10from ..config import settings 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class GitService: 

16 """Service for Git operations with HTTPS token authentication.""" 

17 

18 def __init__(self) -> None: 

19 """Initialize Git service with settings.""" 

20 self.base_path = Path(settings.TEMP_REPO_BASE_PATH) 

21 self.token = settings.GIT_ACCESS_TOKEN 

22 

23 def _parse_git_url(self, git_url: str) -> dict | None: 

24 """Parse Git URL and return components.""" 

25 # SSH format: git@github.com:owner/repo.git 

26 ssh_pattern = r"^git@([^:]+):([^/]+)/(.+?)(?:\.git)?$" 

27 # HTTPS format: https://github.com/owner/repo.git 

28 https_pattern = r"^https?://([^/]+)/([^/]+)/(.+?)(?:\.git)?$" 

29 

30 ssh_match = re.match(ssh_pattern, git_url) 

31 if ssh_match: 

32 host, owner, repo = ssh_match.groups() 

33 return {"host": host, "owner": owner, "repo": repo} 

34 

35 https_match = re.match(https_pattern, git_url) 

36 if https_match: 

37 host, owner, repo = https_match.groups() 

38 return {"host": host, "owner": owner, "repo": repo} 

39 

40 return None 

41 

42 def _get_authenticated_url(self, git_url: str) -> str: 

43 """Convert Git URL to HTTPS format with token authentication.""" 

44 url = git_url 

45 if url.startswith("git@"): 

46 # Convert SSH to HTTPS: git@github.com:owner/repo.git -> https://token@github.com/owner/repo.git 

47 url = url.replace(":", "/").replace("git@", f"https://{self.token}@") 

48 elif url.startswith("https://"): 

49 # Add token to HTTPS URL 

50 parsed = self._parse_git_url(url) 

51 if parsed: 

52 url = f"https://{self.token}@{parsed['host']}/{parsed['owner']}/{parsed['repo']}.git" 

53 

54 return url 

55 

56 def clone_release(self, git_url: str, tag: str, deployment_id: str) -> str: 

57 """ 

58 Clone a specific release/tag using shallow clone with HTTPS token authentication. 

59 

60 Args: 

61 git_url: URL of the Git repository 

62 tag: Tag/Release name 

63 deployment_id: Unique ID for the target folder 

64 

65 Returns: 

66 Path to the cloned repository 

67 

68 Raises: 

69 Exception with detailed error message 

70 """ 

71 repo_path = self.base_path / f"deploy_{deployment_id}" 

72 

73 if repo_path.exists(): 

74 logger.info(f"Removing existing repo at {repo_path}") 

75 shutil.rmtree(repo_path) 

76 

77 try: 

78 # Convert to authenticated HTTPS URL 

79 auth_url = self._get_authenticated_url(git_url) 

80 

81 logger.info(f"Cloning repository {git_url}") 

82 logger.info(f"Target: {repo_path}") 

83 logger.info(f"Branch/Tag: {tag}") 

84 logger.info("Mode: Shallow clone (depth=1, single branch)") 

85 

86 # Clone with --branch <tag> --depth 1 (no history, only the tag) 

87 repo = git.Repo.clone_from(auth_url, str(repo_path), branch=tag, depth=1, single_branch=True) 

88 

89 logger.info(f"✓ Repository cloned successfully to {repo_path}") 

90 logger.info(f"Current HEAD: {repo.head.commit.hexsha}") 

91 

92 return str(repo_path) 

93 

94 except git.exc.GitCommandError as e: 

95 error_msg = f"Git command failed: {e.stderr if hasattr(e, 'stderr') else str(e)}" 

96 logger.error(error_msg) 

97 if repo_path.exists(): 

98 shutil.rmtree(repo_path) 

99 raise Exception(error_msg) 

100 except Exception as e: 

101 error_msg = f"Failed to clone release {tag} from {git_url}: {str(e)}" 

102 logger.error(error_msg) 

103 if repo_path.exists(): 

104 shutil.rmtree(repo_path) 

105 raise Exception(error_msg) 

106 

107 def cleanup_repository(self, repo_path: str) -> None: 

108 """Delete the cloned repository.""" 

109 path = Path(repo_path) 

110 if path.exists(): 

111 logger.info(f"Cleaning up {repo_path}") 

112 shutil.rmtree(path) 

113 

114 

115# Singleton 

116git_service = GitService()