Coverage for rdak/commands/policy_queue.py: 35%

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# © 2026, Ansgar 🙀 <ansgar@debian.org> 

3 

4""" 

5`rdak policy-queue ...` - manage uploads in policy queues. 

6 

7Mirrors the RPCs of `dak.policyqueue.v1.PolicyQueueService`. 

8""" 

9 

10import argparse 

11import sys 

12from typing import IO 

13 

14from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError 

15from tabulate import tabulate 

16 

17from dak.policyqueue.v1 import policyqueue_pb2, policyqueue_pb2_grpc 

18from rdak import config, connection, output 

19from rdak.errors import CliError, rpc_errors 

20 

21 

22class TextOutput: 

23 """Render PolicyQueueService responses as human-readable text.""" 

24 

25 def __init__(self, stream: IO[str] | None = None) -> None: 

26 self._stream = stream if stream is not None else sys.stdout 

27 

28 def print_uploads(self, response: policyqueue_pb2.ListUploadsResponse) -> None: 

29 rows = sorted((u.name, u.target_suite) for u in response.uploads) 

30 if not rows: 

31 print("no uploads", file=self._stream) 

32 return 

33 print(tabulate(rows, headers=["upload", "target suite"]), file=self._stream) 

34 

35 def print_upload(self, upload: policyqueue_pb2.Upload) -> None: 

36 print(f"upload: {upload.name}", file=self._stream) 

37 print(f"policy queue: {upload.policy_queue}", file=self._stream) 

38 print(f"target suite: {upload.target_suite}", file=self._stream) 

39 if upload.HasField("create_time"): 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 created = upload.create_time.ToDatetime().isoformat() 

41 print(f"created: {created}", file=self._stream) 

42 print(file=self._stream) 

43 if not upload.missing_overrides: 

44 print("no missing overrides", file=self._stream) 

45 return 

46 print("missing overrides:", file=self._stream) 

47 rows = sorted( 

48 (o.package, o.type, o.component, o.priority, o.section) 

49 for o in upload.missing_overrides 

50 ) 

51 print( 

52 tabulate( 

53 rows, 

54 headers=["package", "type", "component", "priority", "section"], 

55 ), 

56 file=self._stream, 

57 ) 

58 

59 

60class JsonOutput: 

61 """Render PolicyQueueService responses as JSON.""" 

62 

63 def __init__(self, stream: IO[str] | None = None) -> None: 

64 self._stream = stream if stream is not None else sys.stdout 

65 

66 def print_uploads(self, response: policyqueue_pb2.ListUploadsResponse) -> None: 

67 output.dump_json(output.to_dict(response), self._stream) 

68 

69 def print_upload(self, upload: policyqueue_pb2.Upload) -> None: 

70 output.dump_json(output.to_dict(upload), self._stream) 

71 

72 

73type Output = TextOutput | JsonOutput 

74 

75 

76def _output(args: argparse.Namespace) -> Output: 

77 return JsonOutput() if args.json else TextOutput() 

78 

79 

80def register(subparsers: "argparse._SubParsersAction") -> None: 

81 parser = subparsers.add_parser( 

82 "policy-queue", help="manage uploads in policy queues" 

83 ) 

84 verbs = parser.add_subparsers(dest="command", required=True) 

85 

86 p = verbs.add_parser("list", help="list uploads in a queue") 

87 p.add_argument("queue") 

88 p.set_defaults(func=cmd_list) 

89 

90 p = verbs.add_parser("show", help="show details for one upload") 

91 p.add_argument("queue") 

92 p.add_argument("name") 

93 p.set_defaults(func=cmd_show) 

94 

95 p = verbs.add_parser("add-overrides", help="add overrides for an upload") 

96 p.add_argument("queue") 

97 p.add_argument("name") 

98 p.add_argument( 

99 "--all-missing", 

100 action="store_true", 

101 help="add every override suggested by the server", 

102 ) 

103 p.add_argument( 

104 "--from-json", 

105 metavar="FILE", 

106 help="read overrides from a JSON file ('-' for stdin)", 

107 ) 

108 p.add_argument( 

109 "--override", 

110 action="append", 

111 default=[], 

112 metavar="PACKAGE:TYPE:COMPONENT:PRIORITY:SECTION", 

113 help="a single override (repeatable)", 

114 ) 

115 p.set_defaults(func=cmd_add_overrides) 

116 

117 p = verbs.add_parser("accept", help="mark an upload for acceptance") 

118 p.add_argument("queue") 

119 p.add_argument("name") 

120 p.set_defaults(func=cmd_accept) 

121 

122 p = verbs.add_parser("reject", help="reject an upload") 

123 p.add_argument("queue") 

124 p.add_argument("name") 

125 p.add_argument("--reason", required=True, help="reason for the rejection") 

126 p.add_argument( 

127 "--rejected-by", 

128 default="", 

129 metavar='"Name <mail@example.com>"', 

130 help="entity that rejected the upload (used as the mail From address)", 

131 ) 

132 p.set_defaults(func=cmd_reject) 

133 

134 

135def _stub(args: argparse.Namespace) -> policyqueue_pb2_grpc.PolicyQueueServiceStub: 

136 cfg = config.load_config(args.config) 

137 _name, instance = cfg.select(args.instance) 

138 channel = connection.connect(instance) 

139 return policyqueue_pb2_grpc.PolicyQueueServiceStub(channel) 

140 

141 

142def cmd_list(args: argparse.Namespace) -> int: 

143 stub = _stub(args) 

144 with rpc_errors(): 

145 response = stub.ListUploads( 

146 policyqueue_pb2.ListUploadsRequest(policy_queue=args.queue) 

147 ) 

148 _output(args).print_uploads(response) 

149 return 0 

150 

151 

152def cmd_show(args: argparse.Namespace) -> int: 

153 stub = _stub(args) 

154 with rpc_errors(): 

155 upload = stub.GetUpload( 

156 policyqueue_pb2.GetUploadRequest(policy_queue=args.queue, name=args.name) 

157 ) 

158 _output(args).print_upload(upload) 

159 return 0 

160 

161 

162def _parse_override(spec: str) -> "policyqueue_pb2.Override": 

163 parts = spec.split(":") 

164 if len(parts) != 5: 

165 raise CliError( 

166 f"invalid --override {spec!r}; expected " 

167 "PACKAGE:TYPE:COMPONENT:PRIORITY:SECTION", 

168 exit_code=2, 

169 ) 

170 package, type_, component, priority, section = parts 

171 return policyqueue_pb2.Override( 

172 component=component, 

173 type=type_, 

174 package=package, 

175 priority=priority, 

176 section=section, 

177 ) 

178 

179 

180class _OverrideInput(BaseModel): 

181 """One override entry read from `--from-json`""" 

182 

183 model_config = ConfigDict(extra="forbid") 

184 

185 component: str 

186 type: str 

187 package: str 

188 priority: str 

189 section: str 

190 

191 

192_OVERRIDES_ADAPTER = TypeAdapter(list[_OverrideInput]) 

193 

194 

195def _overrides_from_json(source: str) -> list["policyqueue_pb2.Override"]: 

196 if source == "-": 

197 text = sys.stdin.read() 

198 else: 

199 try: 

200 with open(source, encoding="utf-8") as f: 

201 text = f.read() 

202 except OSError as e: 

203 raise CliError(f"cannot read {source}: {e}", exit_code=2) from e 

204 try: 

205 entries = _OVERRIDES_ADAPTER.validate_json(text) 

206 except ValidationError as e: 

207 raise CliError(f"invalid overrides in {source}: {e}", exit_code=2) from e 

208 return [ 

209 policyqueue_pb2.Override( 

210 component=o.component, 

211 type=o.type, 

212 package=o.package, 

213 priority=o.priority, 

214 section=o.section, 

215 ) 

216 for o in entries 

217 ] 

218 

219 

220def cmd_add_overrides(args: argparse.Namespace) -> int: 

221 stub = _stub(args) 

222 if args.all_missing: 

223 if args.from_json or args.override: 

224 raise CliError( 

225 "--all-missing cannot be combined with --from-json/--override", 

226 exit_code=2, 

227 ) 

228 with rpc_errors(): 

229 upload = stub.GetUpload( 

230 policyqueue_pb2.GetUploadRequest( 

231 policy_queue=args.queue, name=args.name 

232 ) 

233 ) 

234 overrides = list(upload.missing_overrides) 

235 if not overrides: 

236 print("no missing overrides to add") 

237 return 0 

238 else: 

239 overrides = [] 

240 if args.from_json: 

241 overrides.extend(_overrides_from_json(args.from_json)) 

242 overrides.extend(_parse_override(spec) for spec in args.override) 

243 if not overrides: 

244 raise CliError( 

245 "specify --all-missing, --from-json or --override", exit_code=2 

246 ) 

247 with rpc_errors(): 

248 stub.AddOverrides( 

249 policyqueue_pb2.AddOverridesRequest( 

250 policy_queue=args.queue, name=args.name, overrides=overrides 

251 ) 

252 ) 

253 print(f"added {len(overrides)} override(s)") 

254 return 0 

255 

256 

257def cmd_accept(args: argparse.Namespace) -> int: 

258 stub = _stub(args) 

259 with rpc_errors(): 

260 stub.AcceptUpload( 

261 policyqueue_pb2.AcceptUploadRequest(policy_queue=args.queue, name=args.name) 

262 ) 

263 print(f"accepted {args.name}") 

264 return 0 

265 

266 

267def cmd_reject(args: argparse.Namespace) -> int: 

268 stub = _stub(args) 

269 with rpc_errors(): 

270 stub.RejectUpload( 

271 policyqueue_pb2.RejectUploadRequest( 

272 policy_queue=args.queue, 

273 name=args.name, 

274 rejected_by=args.rejected_by, 

275 reason=args.reason, 

276 ) 

277 ) 

278 print(f"rejected {args.name}") 

279 return 0