Coverage for gsb/fastforward.py: 100%

69 statements  

« prev     ^ index     » next       coverage.py v7.2.6, created at 2024-09-08 16:23 -0400

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

2import datetime as dt 

3import logging 

4from pathlib import Path 

5 

6from . import _git, backup 

7from .logging import IMPORTANT 

8 

9LOGGER = logging.getLogger(__name__) 

10 

11 

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

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

14 effectively deleting the ones in between 

15 

16 Parameters 

17 ---------- 

18 repo_root : Path 

19 The directory containing the GSB-managed repo 

20 starting_point: str 

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

22 will be kept) 

23 *revisions: str 

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

25 / kept in the new history 

26 

27 Returns 

28 ------- 

29 str 

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

31 history 

32 

33 Notes 

34 ----- 

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

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

37 history is rewritten). 

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

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

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

41 

42 Raises 

43 ------ 

44 OSError 

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

46 ValueError 

47 If any of the specified revisions do not exist 

48 """ 

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

50 tag_lookup = { 

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

52 } 

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

54 

55 for reference in revisions: 

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

57 if revision in tag_lookup.keys(): 

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

59 new_history.append(revision) 

60 

61 try: 

62 head = backup.create_backup(repo_root) 

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

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

65 except ValueError: 

66 # nothing to back up 

67 pass 

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 new_hash = _git.commit( 

80 repo_root, 

81 message=( 

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

83 ), 

84 timestamp=revision.timestamp, 

85 ).hash 

86 head = new_hash 

87 case _git.Tag(): 

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

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

90 new_hash = _git.commit( 

91 repo_root, 

92 message=( 

93 (revision.annotation or revision.name) 

94 + "\n\n" 

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

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

97 ), 

98 timestamp=revision.target.timestamp, 

99 ).hash 

100 tags_to_update.append((revision, new_hash)) 

101 head = new_hash 

102 case _: # pragma: no cover 

103 raise NotImplementedError( 

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

105 ) 

106 for tag, target in tags_to_update: 

107 _git.delete_tag(repo_root, tag.name) 

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

109 try: 

110 _git.delete_branch(repo_root, "gsb") 

111 except ValueError as delete_fail: 

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

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

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

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

116 return head 

117 except Exception as something_went_wrong: # pragma: no cover 

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

119 raise something_went_wrong 

120 finally: 

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

122 _git.delete_branch(repo_root, branch_name) 

123 

124 

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

126 """Delete the specified backups 

127 

128 Parameters 

129 ---------- 

130 repo_root : Path 

131 The directory containing the GSB-managed repo 

132 *revisions : str 

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

134 

135 Returns 

136 ------- 

137 str 

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

139 history 

140 

141 Notes 

142 ----- 

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

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

145 history is rewritten). 

146 - Deleting the initial backup is not currently supported. 

147 

148 Raises 

149 ------ 

150 OSError 

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

152 ValueError 

153 If the specified revision does not exist 

154 """ 

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

156 for revision in revisions: 

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

158 case _git.Commit(): 

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

160 # an error about it later 

161 to_delete[reference] = revision 

162 case _git.Tag(): 

163 to_delete[reference.target] = revision 

164 case _: # pragma: no cover 

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

166 

167 to_keep: list[str] = [] 

168 for commit in _git.log(repo_root): 

169 if commit in to_delete: 

170 del to_delete[commit] 

171 else: 

172 to_keep.insert(0, commit.hash) 

173 if len(to_delete) == 0: 

174 break 

175 else: 

176 if len(to_delete) == 0: 

177 raise NotImplementedError( 

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

179 ) 

180 raise ValueError( 

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

182 " the linear commit history:\n" 

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

184 ) 

185 return rewrite_history(repo_root, *to_keep)