Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/git_utils/branch_tracker.py: 36%

1from functools import partial 

2from typing import Optional, Tuple 

3 

4from git import GitCommandError, Head, Remote, Repo 

5from git.objects.commit import Commit 

6from opal_common.git_utils.env import provide_git_ssh_environment 

7from opal_common.git_utils.exceptions import GitFailed 

8from opal_common.logger import logger 

9from tenacity import retry, stop_after_attempt, wait_fixed 

10 

11 

12class BranchTracker: 

13 """Tracks the state of a git branch (hash at branch HEAD). 

14 

15 can also perform git pull and detect if the hash changed. 

16 """ 

17 

18 DEFAULT_RETRY_CONFIG = { 

19 "wait": wait_fixed(3), 

20 "stop": stop_after_attempt(2), 

21 "reraise": True, 

22 } 

23 

24 def __init__( 

25 self, 

26 repo: Repo, 

27 branch_name: str = "master", 

28 remote_name: str = "origin", 

29 retry_config=None, 

30 ssh_key: Optional[str] = None, 

31 ): 

32 """[summary] 

33 

34 Args: 

35 repo (Repo): a git repo in which we want to track the latest commit of a branch 

36 branch_name (str): the branch we want to track 

37 remote_name (str): the remote in which the branch upstream is located 

38 retry_config (dict): Tenacity.retry config (@see https://tenacity.readthedocs.io/en/latest/api.html#retry-main-api) 

39 """ 

40 self._repo = repo 

41 self._branch_name = branch_name 

42 self._remote_name = remote_name 

43 self._ssh_key = ssh_key 

44 self._retry_config = ( 

45 retry_config if retry_config is not None else self.DEFAULT_RETRY_CONFIG 

46 ) 

47 

48 self.checkout() 

49 self._save_latest_commit_as_prev_commit() 

50 

51 @property 

52 def repo(self) -> Repo: 

53 """The repo we are tracking.""" 

54 return self._repo 

55 

56 def pull(self) -> Tuple[bool, Commit, Commit]: 

57 """Git pulls from tracked remote. 

58 

59 Returns: 

60 pull_result (bool, Commit, Commit): a tuple consisting of: 

61 has_changes (bool): whether the remote had new commits on our tracked branch 

62 prev (Commit): the previous (before the pull) top-most commit on the tracked branch 

63 latest (Commit): the new top-most (latest) commit on the tracked branch 

64 """ 

65 self._pull() 

66 

67 if self.prev_commit.hexsha == self.latest_commit.hexsha: 

68 return False, self.prev_commit, self.prev_commit 

69 else: 

70 prev = self._prev_commit 

71 self._save_latest_commit_as_prev_commit() 

72 return True, prev, self.latest_commit 

73 

74 def _pull(self): 

75 """Runs git pull with retries.""" 

76 

77 def _inner_pull(*args, **kwargs): 

78 env = provide_git_ssh_environment(self.tracked_remote.url, self._ssh_key) 

79 with self.tracked_remote.repo.git.custom_environment(**env): 

80 self.tracked_remote.pull(*args, **kwargs) 

81 

82 attempt_pull = retry(**self._retry_config)(_inner_pull) 

83 return attempt_pull() 

84 

85 def checkout(self): 

86 """Checkouts the desired branch.""" 

87 checkout_func = partial(self._repo.git.checkout, self._branch_name) 

88 attempt_checkout = retry(**self._retry_config)(checkout_func) 

89 try: 

90 return attempt_checkout() 

91 except GitCommandError as e: 

92 branches = [ 

93 {"name": head.name, "path": head.path} for head in self._repo.heads 

94 ] 

95 logger.error( 

96 "did not find main branch: {branch_name}, instead found: {branches_found}, got error: {error}", 

97 branch_name=self._branch_name, 

98 branches_found=branches, 

99 error=str(e), 

100 ) 

101 raise GitFailed(e) 

102 

103 def _save_latest_commit_as_prev_commit(self): 

104 """Saves the top of the branch as a last known commit (HEAD). 

105 

106 in the next pull, we can then compare the new branch HEAD to the 

107 previous _prev_commit. 

108 """ 

109 self._prev_commit = self.latest_commit 

110 

111 @property 

112 def latest_commit(self) -> Commit: 

113 """The top commit (HEAD) of the tracked branch.""" 

114 return self.tracked_branch.commit 

115 

116 @property 

117 def prev_commit(self) -> Commit: 

118 """The last previously known HEAD of the tracked branch.""" 

119 return self._prev_commit 

120 

121 @property 

122 def tracked_branch(self) -> Head: 

123 """Returns the tracked branch object (of type git.HEAD) or throws if 

124 such branch does not exist on the repo.""" 

125 try: 

126 return getattr(self._repo.heads, self._branch_name) 

127 except AttributeError as e: 

128 branches = [ 

129 {"name": head.name, "path": head.path} for head in self._repo.heads 

130 ] 

131 logger.exception( 

132 "did not find main branch: {error}, instead found: {branches_found}", 

133 error=e, 

134 branches_found=branches, 

135 ) 

136 raise GitFailed(e) 

137 

138 @property 

139 def tracked_remote(self) -> Remote: 

140 """Returns the tracked remote object (of type git.Remote) or throws if 

141 such remote does not exist on the repo.""" 

142 try: 

143 return getattr(self._repo.remotes, self._remote_name) 

144 except AttributeError as e: 

145 remotes = [remote.name for remote in self._repo.remotes] 

146 logger.exception( 

147 "did not find main branch: {error}, instead found: {remotes_found}", 

148 error=e, 

149 remotes_found=remotes, 

150 ) 

151 raise GitFailed(e)