Coverage for britney2/excusefinder.py: 92%

346 statements  

« prev     ^ index     » next       coverage.py v7.6.0, created at 2026-08-18 12:43 +0000

1import logging 

2import optparse 

3from collections.abc import Iterable 

4from itertools import chain 

5from typing import TYPE_CHECKING, TypeVar 

6from urllib.parse import quote 

7 

8import apt_pkg 

9 

10from britney2 import BinaryPackage, BinaryPackageId, PackageId, Suites 

11from britney2.excuse import Excuse 

12from britney2.migrationitem import MigrationItem, MigrationItemFactory 

13from britney2.policies import PolicyVerdict 

14from britney2.utils import ( 

15 filter_out_faux_gen, 

16 find_smooth_updateable_binaries, 

17 invalidate_excuses, 

18) 

19 

20if TYPE_CHECKING: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 from .hints import HintCollection 

22 from .installability.universe import BinaryPackageUniverse 

23 from .policies.policy import PolicyEngine 

24 

25 

26class ExcuseFinder: 

27 

28 def __init__( 

29 self, 

30 options: optparse.Values, 

31 suite_info: Suites, 

32 all_binaries: dict[BinaryPackageId, BinaryPackage], 

33 pkg_universe: "BinaryPackageUniverse", 

34 policy_engine: "PolicyEngine", 

35 mi_factory: MigrationItemFactory, 

36 hints: "HintCollection", 

37 ) -> None: 

38 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__)) 

39 self.logger = logging.getLogger(logger_name) 

40 self.options = options 

41 self.suite_info = suite_info 

42 self.all_binaries = all_binaries 

43 self.pkg_universe = pkg_universe 

44 self._policy_engine = policy_engine 

45 self._migration_item_factory = mi_factory 

46 self.hints = hints 

47 self.excuses: dict[str, Excuse] = {} 

48 

49 def _get_build_link( 

50 self, arch: str, src: str, ver: str, label: str | None = None 

51 ) -> str: 

52 """Return a link to the build logs, labelled 'arch' per default""" 

53 if label is None: 

54 label = arch 

55 if self.options.build_url: 

56 url = self.options.build_url.format( 

57 arch=arch, source=quote(src), version=quote(ver) 

58 ) 

59 return f'<a href="{url}" target="_blank">{label}</a>' 

60 else: 

61 return label 

62 

63 def _should_remove_source(self, item: MigrationItem) -> bool: 

64 """Check if a source package should be removed from testing 

65 

66 This method checks if a source package should be removed from the 

67 target suite; this happens if the source package is not 

68 present in the primary source suite anymore. 

69 

70 It returns True if the package can be removed, False otherwise. 

71 In the former case, a new excuse is appended to the object 

72 attribute excuses. 

73 """ 

74 if hasattr(self.options, "partial_source"): 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 return False 

76 # if the source package is available in unstable, then do nothing 

77 source_suite = self.suite_info.primary_source_suite 

78 pkg = item.package 

79 if pkg in source_suite.sources: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 return False 

81 # otherwise, add a new excuse for its removal 

82 src = item.suite.sources[pkg] 

83 excuse = Excuse(item) 

84 excuse.addinfo(f"Package not in {source_suite.name}, will try to remove") 

85 excuse.set_vers(src.version, None) 

86 if src.maintainer: 

87 excuse.set_maint(src.maintainer) 

88 if src.section: 88 ↛ 92line 88 didn't jump to line 92 because the condition on line 88 was always true

89 excuse.set_section(src.section) 

90 

91 # if the package is blocked, skip it 

92 if ( 

93 hint := self.hints.search_first("block", package=pkg, removal=True) 

94 ) is not None: 

95 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

96 excuse.add_verdict_info( 

97 excuse.policy_verdict, 

98 f"Not touching package, as requested by {hint.user} " 

99 f"(contact {self.options.distribution}-release if update is needed)", 

100 ) 

101 excuse.addreason("block") 

102 self.excuses[excuse.name] = excuse 

103 return False 

104 

105 excuse.policy_verdict = PolicyVerdict.PASS 

106 self.excuses[excuse.name] = excuse 

107 return True 

108 

109 def _should_upgrade_srcarch(self, item: MigrationItem) -> bool: 

110 """Check if a set of binary packages should be upgraded 

111 

112 This method checks if the binary packages produced by the source 

113 package on the given architecture should be upgraded; this can 

114 happen also if the migration is a binary-NMU for the given arch. 

115 

116 It returns False if the given packages don't need to be upgraded, 

117 True otherwise. In the former case, a new excuse is appended to 

118 the object attribute excuses. 

119 """ 

120 # retrieve the source packages for testing and suite 

121 

122 target_suite = self.suite_info.target_suite 

123 source_suite = item.suite 

124 src = item.package 

125 arch = item.architecture 

126 source_t = target_suite.sources[src] 

127 source_u = source_suite.sources[src] 

128 

129 excuse = Excuse(item) 

130 excuse.set_vers(source_t.version, source_t.version) 

131 if source_u.maintainer: 131 ↛ 133line 131 didn't jump to line 133 because the condition on line 131 was always true

132 excuse.set_maint(source_u.maintainer) 

133 if source_u.section: 133 ↛ 140line 133 didn't jump to line 140 because the condition on line 133 was always true

134 excuse.set_section(source_u.section) 

135 

136 # if there is a `remove' hint and the requested version is the same as the 

137 # version in testing, then stop here and return False 

138 # (as a side effect, a removal may generate such excuses for both the source 

139 # package and its binary packages on each architecture) 

140 if ( 

141 hint := self.hints.search_first( 

142 "remove", package=src, version=source_t.version 

143 ) 

144 ) is not None: 

145 excuse.add_hint(hint) 

146 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

147 excuse.add_verdict_info( 

148 excuse.policy_verdict, f"Removal request by {hint.user}" 

149 ) 

150 excuse.add_verdict_info( 

151 excuse.policy_verdict, "Trying to remove package, not update it" 

152 ) 

153 self.excuses[excuse.name] = excuse 

154 return False 

155 

156 # the starting point is that there is nothing wrong and nothing worth doing 

157 anywrongver = False 

158 anyworthdoing = False 

159 

160 packages_t_a = target_suite.binaries[arch] 

161 packages_s_a = source_suite.binaries[arch] 

162 

163 wrong_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

164 

165 # for every binary package produced by this source in unstable for this architecture 

166 for pkg_id in filter_out_faux_gen(source_u.binaries): 

167 if pkg_id.architecture != arch: 

168 continue 

169 

170 pkg_name = pkg_id.package_name 

171 # TODO filter binaries based on checks below? 

172 excuse.add_package(pkg_id) 

173 

174 # retrieve the testing (if present) and unstable corresponding binary packages 

175 binary_t = packages_t_a[pkg_name] if pkg_name in packages_t_a else None 

176 binary_u = packages_s_a[pkg_name] 

177 

178 # this is the source version for the new binary package 

179 pkgsv = binary_u.source_version 

180 

181 # if the new binary package is architecture-independent, then skip it 

182 if binary_u.architecture == "all": 

183 if pkg_id not in source_t.binaries: 

184 # only add a note if the arch:all does not match the expected version 

185 excuse.add_detailed_info( 

186 "Ignoring %s %s (from %s) as it is arch: all" 

187 % (pkg_name, binary_u.version, pkgsv) 

188 ) 

189 continue 

190 

191 # if the new binary package is not from the same source as the testing one, then skip it 

192 # this implies that this binary migration is part of a source migration 

193 if source_u.version == pkgsv and source_t.version != pkgsv: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true

194 anywrongver = True 

195 excuse.add_verdict_info( 

196 wrong_verdict, 

197 "From wrong source: %s %s (%s not %s)" 

198 % (pkg_name, binary_u.version, pkgsv, source_t.version), 

199 ) 

200 continue 

201 

202 # cruft in unstable 

203 if source_u.version != pkgsv and source_t.version != pkgsv: 

204 if self.options.ignore_cruft: 

205 excuse.add_detailed_info( 

206 "Old cruft: %s %s (but ignoring cruft, so nevermind)" 

207 % (pkg_name, pkgsv) 

208 ) 

209 else: 

210 anywrongver = True 

211 excuse.add_verdict_info( 

212 wrong_verdict, f"Old cruft: {pkg_name} {pkgsv}" 

213 ) 

214 continue 

215 

216 # if the source package has been updated in unstable and this is a binary migration, skip it 

217 # (the binaries are now out-of-date) 

218 if source_t.version == pkgsv and source_t.version != source_u.version: 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true

219 anywrongver = True 

220 excuse.add_verdict_info( 

221 wrong_verdict, 

222 "From wrong source: %s %s (%s not %s)" 

223 % (pkg_name, binary_u.version, pkgsv, source_u.version), 

224 ) 

225 continue 

226 

227 # if the binary is not present in testing, then it is a new binary; 

228 # in this case, there is something worth doing 

229 if not binary_t: 

230 excuse.add_detailed_info(f"New binary: {pkg_name} ({binary_u.version})") 

231 anyworthdoing = True 

232 continue 

233 

234 # at this point, the binary package is present in testing, so we can compare 

235 # the versions of the packages ... 

236 vcompare = apt_pkg.version_compare(binary_t.version, binary_u.version) 

237 

238 # ... if updating would mean downgrading, then stop here: there is something wrong 

239 if vcompare > 0: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true

240 anywrongver = True 

241 excuse.add_verdict_info( 

242 wrong_verdict, 

243 "Not downgrading: %s (%s to %s)" 

244 % (pkg_name, binary_t.version, binary_u.version), 

245 ) 

246 break 

247 # ... if updating would mean upgrading, then there is something worth doing 

248 elif vcompare < 0: 

249 excuse.add_detailed_info( 

250 "Updated binary: %s (%s to %s)" 

251 % (pkg_name, binary_t.version, binary_u.version) 

252 ) 

253 anyworthdoing = True 

254 

255 srcv = source_u.version 

256 same_source = source_t.version == srcv 

257 primary_source_suite = self.suite_info.primary_source_suite 

258 is_primary_source = source_suite == primary_source_suite 

259 

260 # if there is nothing wrong and there is something worth doing or the source 

261 # package is not fake, then check what packages should be removed 

262 if not anywrongver and (anyworthdoing or not source_u.is_fakesrc): 

263 # we want to remove binaries that are no longer produced by the 

264 # new source, but there are some special cases: 

265 # - if this is binary-only (same_source) and not from the primary 

266 # source, we don't do any removals: 

267 # binNMUs in *pu on some architectures would otherwise result in 

268 # the removal of binaries on other architectures 

269 # - for the primary source, smooth binaries in the target suite 

270 # are not considered for removal 

271 if not same_source or is_primary_source: 

272 smoothbins = set() 

273 if is_primary_source: 273 ↛ 291line 273 didn't jump to line 291 because the condition on line 273 was always true

274 binaries_t = target_suite.binaries 

275 possible_smooth_updates = [ 

276 p for p in source_t.binaries if p.architecture == arch 

277 ] 

278 smoothbins = find_smooth_updateable_binaries( 

279 possible_smooth_updates, 

280 source_u, 

281 self.pkg_universe, 

282 target_suite, 

283 binaries_t, 

284 source_suite.binaries, 

285 None, 

286 self.options.smooth_updates, 

287 self.hints, 

288 ) 

289 

290 # for every binary package produced by this source in testing for this architecture 

291 for pkg_id in sorted( 

292 x for x in source_t.binaries if x.architecture == arch 

293 ): 

294 pkg = pkg_id.package_name 

295 # if the package is architecture-independent, then ignore it 

296 tpkg_data = packages_t_a[pkg] 

297 if tpkg_data.architecture == "all": 

298 if pkg_id not in source_u.binaries: 

299 # only add a note if the arch:all does not match the expected version 

300 excuse.add_detailed_info( 

301 f"Ignoring removal of {pkg} as it is arch: all" 

302 ) 

303 continue 

304 # if the package is not produced by the new source package, then remove it from testing 

305 if pkg not in packages_s_a: 

306 excuse.add_detailed_info( 

307 f"Removed binary: {pkg} {tpkg_data.version}" 

308 ) 

309 # the removed binary is only interesting if this is a binary-only migration, 

310 # as otherwise the updated source will already cause the binary packages 

311 # to be updated 

312 if same_source and pkg_id not in smoothbins: 

313 # Special-case, if the binary is a candidate for a smooth update, we do not consider 

314 # it "interesting" on its own. This case happens quite often with smooth updatable 

315 # packages, where the old binary "survives" a full run because it still has 

316 # reverse dependencies. 

317 anyworthdoing = True 

318 

319 if not anyworthdoing and not ( 

320 self.options.archall_inconsistency_allowed and excuse.detailed_info 

321 ): 

322 # nothing worth doing, we don't add an excuse to the list, we just return false 

323 return False 

324 

325 if not anyworthdoing: 

326 # This source has binary differences between the target and source 

327 # suite, but we're not going to upgrade them. Part of the purpose 

328 # of options.archall_inconsistency_allowed is to log the excuse 

329 # with a temporary failure such that the administrators can take 

330 # action so they wish. 

331 excuse.policy_verdict = PolicyVerdict.REJECTED_CANNOT_DETERMINE_IF_PERMANENT 

332 excuse.addreason("everything-ignored") 

333 

334 else: 

335 # there is something worth doing 

336 # we assume that this package will be ok, if not invalidated below 

337 excuse.policy_verdict = PolicyVerdict.PASS 

338 

339 # if there is something something wrong, reject this package 

340 if anywrongver: 

341 excuse.policy_verdict = wrong_verdict 

342 

343 self._policy_engine.apply_srcarch_policies(arch, source_t, source_u, excuse) 

344 

345 self.excuses[excuse.name] = excuse 

346 return excuse.is_valid 

347 

348 def _should_upgrade_src(self, item: MigrationItem) -> bool: 

349 """Check if source package should be upgraded 

350 

351 This method checks if a source package should be upgraded. The analysis 

352 is performed for the source package specified by the `src' parameter, 

353 for the distribution `source_suite'. 

354 

355 It returns False if the given package doesn't need to be upgraded, 

356 True otherwise. In the former case, a new excuse is appended to 

357 the object attribute excuses. 

358 """ 

359 

360 src = item.package 

361 source_suite = item.suite 

362 suite_name = source_suite.name 

363 source_u = source_suite.sources[src] 

364 if source_u.is_fakesrc: 364 ↛ 366line 364 didn't jump to line 366 because the condition on line 364 was never true

365 # it is a fake package created to satisfy Britney implementation details; silently ignore it 

366 return False 

367 

368 target_suite = self.suite_info.target_suite 

369 # retrieve the source packages for testing (if available) and suite 

370 source_t = target_suite.sources.get(src) 

371 if source_t is not None: 

372 # if testing and unstable have the same version, then this is a candidate for binary-NMUs only 

373 if apt_pkg.version_compare(source_t.version, source_u.version) == 0: 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true

374 return False 

375 

376 excuse = Excuse(item) 

377 excuse.set_vers(source_t and source_t.version or None, source_u.version) 

378 if source_u.maintainer: 378 ↛ 380line 378 didn't jump to line 380 because the condition on line 378 was always true

379 excuse.set_maint(source_u.maintainer) 

380 if source_u.section: 380 ↛ 382line 380 didn't jump to line 382 because the condition on line 380 was always true

381 excuse.set_section(source_u.section) 

382 excuse.add_package(PackageId(src, source_u.version, "source")) 

383 

384 # if the version in unstable is older, then stop here with a warning in the excuse and return False 

385 if source_t and apt_pkg.version_compare(source_u.version, source_t.version) < 0: 

386 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

387 excuse.add_verdict_info( 

388 excuse.policy_verdict, 

389 "ALERT: %s is newer in the target suite (%s %s)" 

390 % (src, source_t.version, source_u.version), 

391 ) 

392 self.excuses[excuse.name] = excuse 

393 excuse.addreason("newerintesting") 

394 return False 

395 

396 # the starting point is that we will update the candidate 

397 excuse.policy_verdict = PolicyVerdict.PASS 

398 

399 # if there is a `remove' hint and the requested version is the same as the 

400 # version in testing, then stop here and return False 

401 for hint in self.hints.search("remove", package=src): 

402 if ( 402 ↛ 401line 402 didn't jump to line 401

403 source_t 

404 and source_t.version == hint.version 

405 or source_u.version == hint.version 

406 ): 

407 excuse.add_hint(hint) 

408 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

409 excuse.add_verdict_info( 

410 excuse.policy_verdict, "Removal request by %s" % (hint.user) 

411 ) 

412 excuse.add_verdict_info( 

413 excuse.policy_verdict, "Trying to remove package, not update it" 

414 ) 

415 break 

416 

417 all_binaries = self.all_binaries 

418 

419 # at this point, we check the status of the builds on all the supported architectures 

420 # to catch the out-of-date ones 

421 archs_to_consider = list(self.options.architectures) 

422 archs_to_consider.append("all") 

423 for arch in archs_to_consider: 

424 oodbins: dict[str, set[str]] = {} 

425 uptodatebins = False 

426 # for every binary package produced by this source in the suite for this architecture 

427 if arch == "all": 

428 consider_binaries: Iterable[BinaryPackageId] = source_u.binaries 

429 else: 

430 # Will also include arch:all for the given architecture (they are filtered out 

431 # below) 

432 consider_binaries = sorted( 

433 x for x in source_u.binaries if x.architecture == arch 

434 ) 

435 for pkg_id in consider_binaries: 

436 pkg = pkg_id.package_name 

437 

438 # retrieve the binary package and its source version 

439 binary_u = all_binaries[pkg_id] 

440 pkgsv = binary_u.source_version 

441 

442 # arch:all packages are treated separately from arch:arch 

443 if binary_u.architecture != arch: 

444 continue 

445 

446 # TODO filter binaries based on checks below? 

447 excuse.add_package(pkg_id) 

448 

449 if pkg_id.package_name.endswith("-faux-build-depends"): 

450 continue 

451 

452 # if it wasn't built by the same source, it is out-of-date 

453 # if there is at least one binary on this arch which is 

454 # up-to-date, there is a build on this arch 

455 if source_u.version != pkgsv or pkg_id.architecture == "faux": 

456 if pkgsv not in oodbins: 

457 oodbins[pkgsv] = set() 

458 oodbins[pkgsv].add(pkg) 

459 if pkg_id.architecture != "faux": 

460 excuse.add_old_binary(pkg, pkgsv) 

461 continue 

462 else: 

463 uptodatebins = True 

464 

465 # if there are out-of-date packages, warn about them in the excuse and set excuse.is_valid 

466 # to False to block the update; if the architecture where the package is out-of-date is 

467 # in the `outofsync_arches' list, then do not block the update 

468 if oodbins: 

469 oodtxt = "" 

470 for v in sorted(oodbins): 

471 if oodtxt: 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true

472 oodtxt = oodtxt + "; " 

473 oodtxt = oodtxt + "{} (from {})".format( 

474 ", ".join(sorted(oodbins[v])), 

475 self._get_build_link(arch, src, v, label=v), 

476 ) 

477 

478 if uptodatebins: 

479 text = "Old binaries left on {}: {}".format( 

480 self._get_build_link(arch, src, source_u.version), 

481 oodtxt, 

482 ) 

483 else: 

484 text = "Missing build on %s" % ( 

485 self._get_build_link(arch, src, source_u.version) 

486 ) 

487 

488 if arch in self.options.outofsync_arches: 

489 text = f"{text} (but {arch} isn't keeping up, so nevermind)" 

490 if not uptodatebins: 490 ↛ 423line 490 didn't jump to line 423 because the condition on line 490 was always true

491 excuse.missing_build_on_ood_arch(arch) 

492 else: 

493 if uptodatebins: 

494 if self.options.ignore_cruft: 

495 text = f"{text} (but ignoring cruft, so nevermind)" 

496 excuse.add_detailed_info(text) 

497 else: 

498 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

499 excuse.addreason("cruft") 

500 excuse.add_verdict_info(excuse.policy_verdict, text) 

501 else: 

502 excuse.policy_verdict = ( 

503 PolicyVerdict.REJECTED_CANNOT_DETERMINE_IF_PERMANENT 

504 ) 

505 excuse.missing_build_on_arch(arch) 

506 excuse.addreason("missingbuild") 

507 excuse.add_verdict_info(excuse.policy_verdict, text) 

508 if excuse.old_binaries: 

509 excuse.add_detailed_info( 

510 f"old binaries on {arch}: {oodtxt}" 

511 ) 

512 

513 # if the source package has no binaries, set is_valid to False to block the update 

514 if not any( 

515 x 

516 for x in filter_out_faux_gen(source_u.binaries) 

517 if x.architecture != "faux" 

518 ): 

519 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

520 excuse.add_verdict_info( 

521 excuse.policy_verdict, f"{src} has no binaries on any arch" 

522 ) 

523 excuse.addreason("no-binaries") 

524 

525 self._policy_engine.apply_src_policies(source_t, source_u, excuse) 

526 

527 if source_suite.suite_class.is_additional_source and source_t: 

528 # o-o-d(ish) checks for (t-)p-u 

529 # This only makes sense if the package is actually in testing. 

530 for arch in self.options.architectures: 

531 # if the package in testing has no binaries on this 

532 # architecture, it can't be out-of-date 

533 if not any( 

534 x 

535 for x in source_t.binaries 

536 if x.architecture == arch and all_binaries[x].architecture != "all" 

537 ): 

538 continue 

539 

540 # if the (t-)p-u package has produced any binaries on 

541 # this architecture then we assume it's ok. this allows for 

542 # uploads to (t-)p-u which intentionally drop binary 

543 # packages 

544 if any( 

545 x 

546 for x in source_suite.binaries[arch].values() 

547 if x.source == src 

548 and x.source_version == source_u.version 

549 and x.architecture != "all" 

550 ): 

551 continue 

552 

553 # TODO: Find a way to avoid hardcoding pu/stable relation. 

554 if suite_name == "pu": 554 ↛ 555line 554 didn't jump to line 555 because the condition on line 554 was never true

555 base = "stable" 

556 else: 

557 base = target_suite.name 

558 text = "Not yet built on %s (relative to target suite)" % ( 

559 self._get_build_link(arch, src, source_u.version) 

560 ) 

561 

562 if arch in self.options.outofsync_arches: 562 ↛ 563line 562 didn't jump to line 563 because the condition on line 562 was never true

563 text = "{text} (but {arch} isn't keeping up, so never mind)" 

564 excuse.missing_build_on_ood_arch(arch) 

565 excuse.addinfo(text) 

566 else: 

567 excuse.policy_verdict = ( 

568 PolicyVerdict.REJECTED_CANNOT_DETERMINE_IF_PERMANENT 

569 ) 

570 excuse.missing_build_on_arch(arch) 

571 excuse.addreason("missingbuild") 

572 excuse.add_verdict_info(excuse.policy_verdict, text) 

573 

574 # check if there is a `force' hint for this package, which allows it to go in even if it is not updateable 

575 if ( 

576 force_hint := self.hints.search_first( 

577 "force", package=src, version=source_u.version 

578 ) 

579 ) is not None: 

580 # force() updates the final verdict for us 

581 changed_state = excuse.force() 

582 if changed_state: 

583 excuse.addinfo(f"Should ignore, but forced by {force_hint.user}") 

584 

585 self.excuses[excuse.name] = excuse 

586 return excuse.is_valid 

587 

588 def _compute_excuses_and_initial_actionable_items(self) -> set[MigrationItem]: 

589 # list of local methods and variables (for better performance) 

590 excuses = self.excuses 

591 suite_info = self.suite_info 

592 pri_source_suite = suite_info.primary_source_suite 

593 architectures = self.options.architectures 

594 should_remove_source = self._should_remove_source 

595 should_upgrade_srcarch = self._should_upgrade_srcarch 

596 should_upgrade_src = self._should_upgrade_src 

597 

598 sources_ps = pri_source_suite.sources 

599 sources_t = suite_info.target_suite.sources 

600 

601 # this set will contain the packages which are valid candidates; 

602 # if a package is going to be removed, it will have a "-" prefix 

603 actionable_items: set[MigrationItem] = set() 

604 actionable_items_add = actionable_items.add # Every . in a loop slows it down 

605 

606 # for every source package in testing, check if it should be removed 

607 for pkg in sources_t: 

608 if pkg not in sources_ps: 

609 src_t = sources_t[pkg] 

610 item = MigrationItem( 

611 package=pkg, 

612 version=src_t.version, 

613 suite=suite_info.target_suite, 

614 is_removal=True, 

615 ) 

616 if should_remove_source(item): 

617 actionable_items_add(item) 

618 

619 # for every source package in the source suites, check if it should be upgraded 

620 for suite in chain((pri_source_suite, *suite_info.additional_source_suites)): 

621 sources_s = suite.sources 

622 for pkg in sources_s: 

623 src_s_data = sources_s[pkg] 

624 if src_s_data.is_fakesrc: 

625 continue 

626 src_t_data = sources_t.get(pkg) 

627 

628 if ( 

629 src_t_data is None 

630 or apt_pkg.version_compare(src_s_data.version, src_t_data.version) 

631 != 0 

632 ): 

633 item = MigrationItem( 

634 package=pkg, version=src_s_data.version, suite=suite 

635 ) 

636 # check if the source package should be upgraded 

637 if should_upgrade_src(item): 

638 actionable_items_add(item) 

639 else: 

640 # package has same version in source and target suite; check if any of the 

641 # binaries have changed on the various architectures 

642 for arch in architectures: 

643 item = MigrationItem( 

644 package=pkg, 

645 version=src_s_data.version, 

646 architecture=arch, 

647 suite=suite, 

648 ) 

649 if should_upgrade_srcarch(item): 

650 actionable_items_add(item) 

651 

652 # process the `remove' hints, if the given package is not yet in actionable_items 

653 for hint in self.hints["remove"]: 

654 src_r = hint.package 

655 assert src_r is not None 

656 

657 source_t = sources_t.get(src_r) 

658 if source_t is None: 

659 continue 

660 

661 existing_items = {x for x in actionable_items if x.package == src_r} 

662 if existing_items: 

663 self.logger.info( 

664 "removal hint '%s' ignored due to existing item(s) %s", 

665 hint, 

666 [i.name for i in existing_items], 

667 ) 

668 continue 

669 

670 tsrcv = source_t.version 

671 item = MigrationItem( 

672 package=src_r, 

673 version=tsrcv, 

674 suite=suite_info.target_suite, 

675 is_removal=True, 

676 ) 

677 

678 # check if the version specified in the hint is the same as the considered package 

679 if tsrcv != hint.version: 679 ↛ 680line 679 didn't jump to line 680 because the condition on line 679 was never true

680 continue 

681 

682 # add the removal of the package to actionable_items and build a new excuse 

683 excuse = Excuse(item) 

684 excuse.set_vers(tsrcv, None) 

685 excuse.addinfo(f"Removal request by {hint.user}") 

686 # if the removal of the package is blocked, skip it 

687 if ( 

688 blockhint := self.hints.search_first( 

689 "block", package=src_r, removal=True 

690 ) 

691 ) is not None: 

692 excuse.policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY 

693 excuse.add_verdict_info( 

694 excuse.policy_verdict, 

695 "Not removing package, due to block hint by %s " 

696 "(contact %s-release if update is needed)" 

697 % (blockhint.user, self.options.distribution), 

698 ) 

699 excuse.addreason("block") 

700 excuses[excuse.name] = excuse 

701 continue 

702 

703 actionable_items_add(item) 

704 excuse.addinfo("Package is broken, will try to remove") 

705 excuse.add_hint(hint) 

706 # Using "PASS" here as "Created by a hint" != "accepted due to hint". In a future 

707 # where there might be policy checks on removals, it would make sense to distinguish 

708 # those two states. Not sure that future will ever be. 

709 excuse.policy_verdict = PolicyVerdict.PASS 

710 excuses[excuse.name] = excuse 

711 

712 return actionable_items 

713 

714 def find_actionable_excuses(self) -> tuple[dict[str, Excuse], set[MigrationItem]]: 

715 excuses = self.excuses 

716 actionable_items = self._compute_excuses_and_initial_actionable_items() 

717 valid = {x.name for x in actionable_items} 

718 

719 # extract the not considered packages, which are in the excuses but not in upgrade_me 

720 unconsidered = {ename for ename in excuses if ename not in valid} 

721 invalidated: set[str] = set() 

722 

723 invalidate_excuses(excuses, valid, unconsidered, invalidated) 

724 

725 # check that the list of actionable items matches the list of valid 

726 # excuses 

727 assert_sets_equal(valid, {x for x, e in excuses.items() if e.is_valid}) 

728 

729 # check that the rdeps for all invalid excuses were invalidated 

730 assert_sets_equal( 

731 invalidated, {x for x, e in excuses.items() if not e.is_valid} 

732 ) 

733 

734 actionable_items = {x for x in actionable_items if x.name in valid} 

735 return excuses, actionable_items 

736 

737 

738_T = TypeVar("_T") 

739 

740 

741def assert_sets_equal(a: set[_T], b: set[_T]) -> None: 

742 if a != b: 742 ↛ 743line 742 didn't jump to line 743 because the condition on line 742 was never true

743 raise AssertionError(f"sets not equal a-b {a - b} b-a {b - a}")