Coverage for gsb/fastforward.py: 100%

69 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2024-09-08 20:16 +0000

1"""Functionality for removing backups from a repo's history""" 

2 

3import datetime as dt 

4import logging 

5from pathlib import Path 

6 

7from . import _git, backup 

8from .logging import IMPORTANT 

9 

10LOGGER = logging.getLogger(__name__) 

11 

12 

13def rewrite_history(repo_root: Path, starting_point: str, *revisions: str) -> str: 

14 """Rewrite the repo's history by only including the specified backups, 

15 effectively deleting the ones in between 

16 

17 Parameters 

18 ---------- 

19 repo_root : Path 

20 The directory containing the GSB-managed repo 

21 starting_point: str 

22 The commit hash or tag name to start revising from (all prior backups 

23 will be kept) 

24 *revisions: str 

25 The commit hashes / tag names for the backups that should be included 

26 / kept in the new history 

27 

28 Returns 

29 ------- 

30 str 

31 The tag name or commit hash for the most recent backup in the rewritten 

32 history 

33 

34 Notes 

35 ----- 

36 - The current repo state will always be kept (and, in the case that there 

37 are un-backed-up changes, those changes will be backed up before the 

38 history is rewritten). 

39 - The ordering of the provided revisions is not checked in advance, nor is 

40 anything done to check for duplicates. Providing the backups out-of-order 

41 will create a new history that frames the backups in the order provided. 

42 

43 Raises 

44 ------ 

45 OSError 

46 If the specified repo does not exist or is not a gsb-managed repo 

47 ValueError 

48 If any of the specified revisions do not exist 

49 """ 

50 _ = _git.show(repo_root, starting_point) # ensure starting point is valid 

51 tag_lookup = { 

52 tag.target: tag for tag in _git.get_tags(repo_root, annotated_only=False) 

53 } 

54 new_history: list[_git.Tag | _git.Commit] = [] 

55 

56 for reference in revisions: 

57 revision: _git.Tag | _git.Commit = _git.show(repo_root, reference) 

58 if revision in tag_lookup.keys(): 

59 revision = tag_lookup[revision] # type: ignore[index] 

60 new_history.append(revision) 

61 

62 try: 

63 head = backup.create_backup(repo_root) 

64 LOGGER.log(IMPORTANT, "Unsaved changes have been backed up as %s", head[:8]) 

65 new_history.append(_git.show(repo_root, head)) 

66 except ValueError: 

67 # nothing to back up 

68 pass 

69 

70 try: 

71 branch_name = dt.datetime.now().strftime("gsb_rebase_%Y.%m.%d+%H%M%S") 

72 _git.checkout_branch(repo_root, branch_name, starting_point) 

73 head = starting_point 

74 tags_to_update: list[tuple[_git.Tag, str]] = [] 

75 for revision in new_history: 

76 match revision: 

77 case _git.Commit(): 

78 _git.reset(repo_root, revision.hash, hard=True) 

79 _git.reset(repo_root, head, hard=False) 

80 new_hash = _git.commit( 

81 repo_root, 

82 message=( 

83 revision.message + "\n\n" + f"rebase of {revision.hash}" 

84 ), 

85 timestamp=revision.timestamp, 

86 ).hash 

87 head = new_hash 

88 case _git.Tag(): 

89 _git.reset(repo_root, revision.target.hash, hard=True) 

90 _git.reset(repo_root, head, hard=False) 

91 new_hash = _git.commit( 

92 repo_root, 

93 message=( 

94 (revision.annotation or revision.name) 

95 + "\n\n" 

96 + f"rebase of {revision.target.hash}" 

97 + f' ("{revision.target.message.strip()}")' 

98 ), 

99 timestamp=revision.target.timestamp, 

100 ).hash 

101 tags_to_update.append((revision, new_hash)) 

102 head = new_hash 

103 case _: # pragma: no cover 

104 raise NotImplementedError( 

105 f"Don't know how to handle revision of type {type(revision)}" 

106 ) 

107 for tag, target in tags_to_update: 

108 _git.delete_tag(repo_root, tag.name) 

109 _git.tag(repo_root, tag.name, tag.annotation, target=target) 

110 try: 

111 _git.delete_branch(repo_root, "gsb") 

112 except ValueError as delete_fail: 

113 # this can happen if you onboarded an existing repo to gsb, in 

114 # which case the active branch won't necessarily be gsb 

115 LOGGER.warning("Could not delete branch %s:\n %s", "gsb", delete_fail) 

116 _git.checkout_branch(repo_root, "gsb", head) 

117 return head 

118 except Exception as something_went_wrong: # pragma: no cover 

119 _git.reset(repo_root, head, hard=True) 

120 raise something_went_wrong 

121 finally: 

122 _git.checkout_branch(repo_root, "gsb", None) 

123 _git.delete_branch(repo_root, branch_name) 

124 

125 

126def delete_backups(repo_root: Path, *revisions: str) -> str: 

127 """Delete the specified backups 

128 

129 Parameters 

130 ---------- 

131 repo_root : Path 

132 The directory containing the GSB-managed repo 

133 *revisions : str 

134 The commit hashes and tag names of the backups to delete 

135 

136 Returns 

137 ------- 

138 str 

139 The tag name or commit hash for the most recent backup in the rewritten 

140 history 

141 

142 Notes 

143 ----- 

144 - The current repo state will always be kept (and, in the case that there 

145 are un-backed-up changes, those changes will be backed up before the 

146 history is rewritten). 

147 - Deleting the initial backup is not currently supported. 

148 

149 Raises 

150 ------ 

151 OSError 

152 If the specified repo does not exist or is not a gsb-managed repo 

153 ValueError 

154 If the specified revision does not exist 

155 """ 

156 to_delete: dict[_git.Commit, str] = {} 

157 for revision in revisions: 

158 match reference := _git.show(repo_root, revision): 

159 case _git.Commit(): 

160 # keep the link back to the revision in case we need to raise 

161 # an error about it later 

162 to_delete[reference] = revision 

163 case _git.Tag(): 

164 to_delete[reference.target] = revision 

165 case _: # pragma: no cover 

166 raise NotImplementedError(f"Don't know how to handle {type(reference)}") 

167 

168 to_keep: list[str] = [] 

169 for commit in _git.log(repo_root): 

170 if commit in to_delete: 

171 del to_delete[commit] 

172 else: 

173 to_keep.insert(0, commit.hash) 

174 if len(to_delete) == 0: 

175 break 

176 else: 

177 if len(to_delete) == 0: 

178 raise NotImplementedError( 

179 "Deleting the initial backup is not currently supported." 

180 ) 

181 raise ValueError( 

182 "The following revisions exist, but they are not within" 

183 " the linear commit history:\n" 

184 + "\n".join((f" - {revision}" for revision in to_delete.values())) 

185 ) 

186 return rewrite_history(repo_root, *to_keep)