Coverage for gsb/fastforward.py: 100%

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: # no unsaved changes 

67 head = _git.show(repo_root, "HEAD").hash # type: ignore[union-attr] 

68 

69 try: 

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

71 _git.checkout_branch(repo_root, branch_name, starting_point) 

72 head = starting_point 

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

74 for revision in new_history: 

75 match revision: 

76 case _git.Commit(): 

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

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

79 try: 

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 except ValueError: # nothing to commit 

89 # identical to the last revision, so fuhgeddaboudit 

90 pass 

91 case _git.Tag(): 

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

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

94 try: 

95 new_hash = _git.commit( 

96 repo_root, 

97 message=( 

98 (revision.annotation or revision.name) 

99 + "\n\n" 

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

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

102 ), 

103 timestamp=revision.target.timestamp, 

104 ).hash 

105 except ValueError: # nothing to commit 

106 LOGGER.warning( 

107 "Backup %s is identical to %s", 

108 revision.name, 

109 head, 

110 ) 

111 new_hash = head 

112 tags_to_update.append((revision, new_hash)) 

113 head = new_hash 

114 

115 case _: # pragma: no cover 

116 raise NotImplementedError( 

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

118 ) 

119 

120 for tag, target in tags_to_update: 

121 _git.delete_tag(repo_root, tag.name) 

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

123 try: 

124 _git.delete_branch(repo_root, "gsb") 

125 except ValueError as delete_fail: 

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

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

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

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

130 return head 

131 except Exception as something_went_wrong: # pragma: no cover 

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

133 raise something_went_wrong 

134 finally: 

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

136 _git.delete_branch(repo_root, branch_name) 

137 

138 

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

140 """Delete the specified backups 

141 

142 Parameters 

143 ---------- 

144 repo_root : Path 

145 The directory containing the GSB-managed repo 

146 *revisions : str 

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

148 

149 Returns 

150 ------- 

151 str 

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

153 history 

154 

155 Notes 

156 ----- 

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

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

159 history is rewritten). 

160 - Deleting the initial backup is not currently supported. 

161 

162 Raises 

163 ------ 

164 OSError 

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

166 ValueError 

167 If the specified revision does not exist 

168 """ 

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

170 for revision in revisions: 

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

172 case _git.Commit(): 

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

174 # an error about it later 

175 to_delete[reference] = revision 

176 case _git.Tag(): 

177 to_delete[reference.target] = revision 

178 case _: # pragma: no cover 

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

180 

181 to_keep: list[str] = [] 

182 for commit in _git.log(repo_root): 

183 if commit in to_delete: 

184 del to_delete[commit] 

185 else: 

186 to_keep.insert(0, commit.hash) 

187 if len(to_delete) == 0: 

188 break 

189 else: 

190 if len(to_delete) == 0: 

191 raise NotImplementedError( 

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

193 ) 

194 raise ValueError( 

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

196 " the linear commit history:\n" 

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

198 ) 

199 return rewrite_history(repo_root, to_keep[0], *to_keep[1:])