Coverage for /home/benjarobin/Bootlin/projects/Schneider-Electric-Senux/sbom-cve-check/src/sbom_cve_check/utils/git.py: 42%

1# -*- coding: utf-8 -*- 

2# SPDX-License-Identifier: GPL-2.0-only 

3 

4import pathlib 

5import subprocess 

6from datetime import UTC, datetime 

7 

8 

9class GitRepo: 

10 def __init__(self, git_dir: pathlib.Path, remote_name: str = "origin") -> None: 

11 self._remote = remote_name 

12 self._git_dir = git_dir 

13 self._git_exe: str = "git" 

14 

15 @property 

16 def path(self) -> pathlib.Path: 

17 return self._git_dir 

18 

19 def _exec_git_cmd(self, args: list[str]) -> subprocess.CompletedProcess[str]: 

20 cmd = [self._git_exe] 

21 cmd.extend(args) 

22 return subprocess.run( 

23 cmd, 

24 input="", 

25 capture_output=True, 

26 check=True, 

27 cwd=self._git_dir, 

28 encoding="utf-8", 

29 ) 

30 

31 def is_git_repo(self) -> bool: 

32 return self._git_dir.joinpath(".git", "HEAD").is_file() 

33 

34 def get_fetch_url(self) -> str | None: 

35 r = self._exec_git_cmd(["config", "--get", f"remote.{self._remote}.url"]) 

36 return r.stdout.strip() 

37 

38 def is_shallow_repository(self) -> bool: 

39 r = self._exec_git_cmd(["rev-parse", "--is-shallow-repository"]) 

40 return r.stdout.strip() == "true" 

41 

42 def resolve_sha_ref(self, ref: str) -> str | None: 

43 try: 

44 r = self._exec_git_cmd(["rev-parse", "--verify", ref + "^{object}"]) 

45 return r.stdout.strip() 

46 except subprocess.CalledProcessError: 

47 return None 

48 

49 def is_valid_object(self, ref: str) -> bool: 

50 try: 

51 self._exec_git_cmd(["cat-file", "-e", ref + "^{object}"]) 

52 except subprocess.CalledProcessError: 

53 return False 

54 else: 

55 return True 

56 

57 def get_current_branch(self) -> str | None: 

58 try: 

59 r = self._exec_git_cmd(["rev-parse", "--abbrev-ref", "HEAD"]) 

60 return r.stdout.strip() 

61 except subprocess.CalledProcessError: 

62 return None 

63 

64 def get_default_remote_branch(self) -> str | None: 

65 try: 

66 r = self._exec_git_cmd( 

67 ["symbolic-ref", "--short", f"refs/remotes/{self._remote}/HEAD"] 

68 ) 

69 branch = r.stdout.strip() 

70 remote_prefix = f"{self._remote}/" 

71 if branch.startswith(remote_prefix): 

72 len_prefix = len(remote_prefix) 

73 return branch[len_prefix:] 

74 except subprocess.CalledProcessError: 

75 pass 

76 return None 

77 

78 def update_default_remote_branch(self) -> str: 

79 self._exec_git_cmd(["remote", "set-head", self._remote, "--auto"]) 

80 b = self.get_default_remote_branch() 

81 if not b: 

82 raise RuntimeError(f"Default branch not found for {self._git_dir}") 

83 return b 

84 

85 def get_date_last_commit(self) -> datetime: 

86 """ 

87 Get the commiter date time of last commit (HEAD) from the git repository. 

88 """ 

89 r = self._exec_git_cmd(["show", "-s", "--format=%ci", "HEAD"]) 

90 lastest_commit_date = r.stdout.strip() 

91 return datetime.fromisoformat(lastest_commit_date).astimezone(UTC) 

92 

93 def get_date_last_update(self) -> datetime | None: 

94 """ 

95 Get the date time of last database update. 

96 

97 Get the modification date time of .git/ORIG_HEAD. 

98 Return None if the database does not exist yet. 

99 """ 

100 orig_head = self._git_dir.joinpath(".git", "ORIG_HEAD") 

101 if not orig_head.is_file(): 

102 return None 

103 return datetime.fromtimestamp(orig_head.stat().st_mtime, UTC) 

104 

105 def clone(self, url: str, branch: str | None, depth: int = 0) -> None: 

106 cmd = ["clone"] 

107 if depth > 0: 

108 cmd.extend(["--depth", str(depth)]) 

109 if branch: 

110 cmd.extend(["--single-branch", "-b", branch]) 

111 cmd.append(url) 

112 cmd.append(".") 

113 

114 self._git_dir.mkdir(parents=True, exist_ok=True) 

115 self._exec_git_cmd(cmd) 

116 

117 def fetch(self, branch: str, unshallow: bool = False, depth: int = 0) -> None: 

118 cmd = ["fetch", "-f"] 

119 if depth > 0: 

120 cmd.extend(["--depth", str(depth)]) 

121 if unshallow: 

122 cmd.append("--unshallow") 

123 cmd.append(self._remote) 

124 cmd.append(f"refs/heads/{branch}:refs/remotes/{self._remote}/{branch}") 

125 self._exec_git_cmd(cmd) 

126 

127 def switch_force_create_branch(self, branch: str, discard_changes: bool) -> None: 

128 cmd = ["switch"] 

129 if discard_changes: 

130 cmd.append("-f") 

131 cmd.extend(["-C", branch, f"{self._remote}/{branch}"]) 

132 self._exec_git_cmd(cmd) 

133 

134 def switch_detach_head(self, ref: str, discard_changes: bool) -> None: 

135 cmd = ["switch"] 

136 if discard_changes: 

137 cmd.append("-f") 

138 cmd.extend(["--detach", ref]) 

139 self._exec_git_cmd(cmd)