Coverage for src / gitq / git_queue.py: 80%

1import re 

2import argparse 

3 

4import yaml 

5 

6from .git import Git 

7from .queue import QueueFile, Baseline, Queue, Loader as QueueLoader 

8from .continuations import Abort, UserError 

9from . import continuations 

10 

11 

12def parse_baseline(ref: str, *, git: Git) -> Baseline: 

13 "create a new baseline from user-provided string" 

14 url = None 

15 sha = git.rev_parse(ref) 

16 full_name = git.symbolic_full_name(ref) 

17 if m := re.match(r"refs/remotes/(\w+)/(.*)", full_name or ""): 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 remote, branch = m.groups() 

19 url = git.cmd(["git", "remote", "get-url", remote], quiet=True).strip() 

20 return Baseline(sha, f"refs/heads/{branch}", url) 

21 elif ref == sha or ref == "HEAD": 

22 return Baseline(sha, None, None) 

23 else: 

24 return Baseline(sha, full_name, None) 

25 

26 

27description_init = """ 

28Initialize a queue on the current branch (or a new branch with -b), 

29with one or more baselines. Each BASELINE is a branch, tag, or commit. 

30""" 

31 

32description_rebase = """ 

33Rebase the queue onto its baselines, incorporating any upstream changes. 

34If a baseline branch is itself a queue managed by this tool, it will be 

35recursively rebased first. 

36 

37Use --add to incorporate an additional baseline, or --remove to drop one, 

38at the same time as rebasing. 

39 

40If conflicts arise during cherry-picking, the operation suspends so the user 

41can resolve them, then resume with `git queue continue`. 

42""" 

43 

44 

45class Main(continuations.Main): 

46 

47 tool = "git-queue" 

48 suspend_message = "Suspended! Resolve conflicts and resume with `git queue continue`" 

49 

50 def main(self) -> None: 

51 parser = argparse.ArgumentParser( 

52 "git-queue", 

53 description="manage a bunch of patches", 

54 formatter_class=argparse.RawDescriptionHelpFormatter, 

55 ) 

56 subs = parser.add_subparsers(dest="command") 

57 

58 init_parser = subs.add_parser( 

59 "init", 

60 help="initialize a queue", 

61 description=description_init, 

62 formatter_class=argparse.RawDescriptionHelpFormatter, 

63 ) 

64 init_parser.add_argument("baselines", action="extend", nargs="+", metavar="BASELINE") 

65 init_parser.add_argument("--title") 

66 init_parser.add_argument("--branch", "-b", help="make a new branch") 

67 

68 add_parser = subs.add_parser( 

69 "add", 

70 help="add a baseline", 

71 description="Add baselines and rebase.", 

72 ) 

73 add_parser.add_argument("add", action="extend", nargs="+", metavar="BASELINE") 

74 

75 remove_parser = subs.add_parser( 

76 "remove", help="remove a baseline", description="Remove baselines and rebase." 

77 ) 

78 remove_parser.add_argument("remove", action="extend", nargs="+", metavar="BASELINE") 

79 

80 rebase_parser = subs.add_parser( 

81 "rebase", 

82 description=description_rebase, 

83 formatter_class=argparse.RawDescriptionHelpFormatter, 

84 help="rebase queue onto baselines", 

85 ) 

86 rebase_parser.add_argument("--add", metavar="BASELINE", action="append", default=[]) 

87 rebase_parser.add_argument("--remove", metavar="BASELINE", action="append", default=[]) 

88 

89 subs.add_parser( 

90 "tidy", help="normalize .git-queue file", description="Normalize .git-queue file." 

91 ) 

92 

93 subs.add_parser( 

94 "status", help="print status", description="Print status of a suspended operation." 

95 ) 

96 subs.add_parser( 

97 "continue", 

98 help="continue suspended operation", 

99 description="Continue a suspended operation.", 

100 ) 

101 subs.add_parser( 

102 "abort", 

103 help="abort suspend operation", 

104 description="Abort a suspended operation and restore previous state.", 

105 ) 

106 

107 args = parser.parse_args() 

108 if args.command is None: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 parser.print_usage() 

110 

111 if args.command == "status": 

112 self.status() 

113 return 

114 

115 if args.command == "continue": 

116 self.resume(None) 

117 return 

118 

119 if args.command == "abort": 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

120 self.resume(Abort()) 

121 

122 queuefile = self.git.directory / Queue.queuefile_name 

123 

124 if args.command == "tidy": 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 if queuefile.exists(): 

126 with open(queuefile, "r") as f: 

127 q = yaml.load(f, Loader=QueueLoader) 

128 with open(queuefile, "w") as f: 

129 q.dump(f) 

130 

131 with self.setup(): 

132 

133 if args.command == "init": 

134 baselines = [parse_baseline(ref, git=self.git) for ref in args.baselines] 

135 q = QueueFile(baselines=list(baselines), title=args.title) 

136 queue = Queue(self.git, qf=q) 

137 if args.branch: 

138 queue.init_new_branch(args.branch) 

139 else: 

140 queue.init() 

141 

142 if args.command in ("rebase", "add", "remove"): 

143 queue = Queue(self.git) 

144 onto = list(queue.qf.baselines) 

145 

146 for baseline in getattr(args, "add", ()): 

147 onto.append(parse_baseline(baseline, git=self.git)) 

148 

149 for baseline in getattr(args, "remove", ()): 

150 if baseline.startswith("refs/"): 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 ref = baseline 

152 else: 

153 ref = self.git.symbolic_full_name(baseline) 

154 for i, baseline in enumerate(onto): 154 ↛ 158line 154 didn't jump to line 158 because the loop on line 154 didn't complete

155 if baseline.ref == ref: 155 ↛ 154line 155 didn't jump to line 154 because the condition on line 155 was always true

156 break 

157 else: 

158 raise UserError(f"{ref} not found in baselines") 

159 del onto[i] 

160 

161 queue.rebase(onto=onto) 

162 

163 

164main = Main() 

165 

166if __name__ == "__main__": 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 main()