Coverage for britney2/britney.py: 83%
773 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-06-28 21:36 +0000
1#!/usr/bin/python3 -u
3# Copyright (C) 2001-2008 Anthony Towns <ajt@debian.org>
4# Andreas Barth <aba@debian.org>
5# Fabio Tranchitella <kobold@debian.org>
6# Copyright (C) 2010-2013 Adam D. Barratt <adsb@debian.org>
8# This program is free software; you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation; either version 2 of the License, or
11# (at your option) any later version.
13# This program is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16# GNU General Public License for more details.
18"""
19= Introduction =
21This is the Debian testing updater script, also known as "Britney".
23Packages are usually installed into the `testing' distribution after
24they have undergone some degree of testing in unstable. The goal of
25this software is to do this task in a smart way, allowing testing
26to always be fully installable and close to being a release candidate.
28Britney's source code is split between two different but related tasks:
29the first one is the generation of the update excuses, while the
30second tries to update testing with the valid candidates; first
31each package alone, then larger and even larger sets of packages
32together. Each try is accepted if testing is not more uninstallable
33after the update than before.
35= Data Loading =
37In order to analyze the entire Debian distribution, Britney needs to
38load in memory the whole archive: this means more than 10.000 packages
39for twelve architectures, as well as the dependency interconnections
40between them. For this reason, the memory requirements for running this
41software are quite high and at least 1 gigabyte of RAM should be available.
43Britney loads the source packages from the `Sources' file and the binary
44packages from the `Packages_${arch}' files, where ${arch} is substituted
45with the supported architectures. While loading the data, the software
46analyzes the dependencies and builds a directed weighted graph in memory
47with all the interconnections between the packages (see Britney.read_sources
48and Britney.read_binaries).
50Other than source and binary packages, Britney loads the following data:
52 * rc-bugs-*, which contains the list of release-critical bugs for a given
53 version of a source or binary package (see RCBugPolicy.read_bugs).
55 * age-policy-dates, which contains the date of the upload of a given version
56 of a source package (see Britney.read_dates).
58 * age-policy-urgencies, which contains the urgency of the upload of a given
59 version of a source package (see AgePolicy._read_urgencies).
61 * Hints, which contains lists of commands which modify the standard behaviour
62 of Britney (see Britney.read_hints).
64 * Other policies typically require their own data.
66For a more detailed explanation about the format of these files, please read
67the documentation of the related methods. The exact meaning of them will be
68instead explained in the chapter "Excuses Generation".
70= Excuses =
72An excuse is a detailed explanation of why a package can or cannot
73be updated in the testing distribution from a newer package in
74another distribution (like for example unstable). The main purpose
75of the excuses is to be written in an HTML file which will be
76published over HTTP, as well as a YAML file. The maintainers will be able
77to parse it manually or automatically to find the explanation of why their
78packages have been updated or not.
80== Excuses generation ==
82These are the steps (with references to method names) that Britney
83does for the generation of the update excuses.
85 * If a source package is available in testing but it is not
86 present in unstable and no binary packages in unstable are
87 built from it, then it is marked for removal.
89 * Every source package in unstable and testing-proposed-updates,
90 if already present in testing, is checked for binary-NMUs, new
91 or dropped binary packages in all the supported architectures
92 (see Britney.should_upgrade_srcarch). The steps to detect if an
93 upgrade is needed are:
95 1. If there is a `remove' hint for the source package, the package
96 is ignored: it will be removed and not updated.
98 2. For every binary package built from the new source, it checks
99 for unsatisfied dependencies, new binary packages and updated
100 binary packages (binNMU), excluding the architecture-independent
101 ones, and packages not built from the same source.
103 3. For every binary package built from the old source, it checks
104 if it is still built from the new source; if this is not true
105 and the package is not architecture-independent, the script
106 removes it from testing.
108 4. Finally, if there is something worth doing (eg. a new or updated
109 binary package) and nothing wrong it marks the source package
110 as "Valid candidate", or "Not considered" if there is something
111 wrong which prevented the update.
113 * Every source package in unstable and testing-proposed-updates is
114 checked for upgrade (see Britney.should_upgrade_src). The steps
115 to detect if an upgrade is needed are:
117 1. If the source package in testing is more recent the new one
118 is ignored.
120 2. If the source package doesn't exist (is fake), which means that
121 a binary package refers to it but it is not present in the
122 `Sources' file, the new one is ignored.
124 3. If the package doesn't exist in testing, the urgency of the
125 upload is ignored and set to the default (actually `low').
127 4. If there is a `remove' hint for the source package, the package
128 is ignored: it will be removed and not updated.
130 5. If there is a `block' hint for the source package without an
131 `unblock` hint or a `block-all source`, the package is ignored.
133 6. If there is a `block-udeb' hint for the source package, it will
134 have the same effect as `block', but may only be cancelled by
135 a subsequent `unblock-udeb' hint.
137 7. If the suite is unstable, the update can go ahead only if the
138 upload happened more than the minimum days specified by the
139 urgency of the upload; if this is not true, the package is
140 ignored as `too-young'. Note that the urgency is sticky, meaning
141 that the highest urgency uploaded since the previous testing
142 transition is taken into account.
144 8. If the suite is unstable, all the architecture-dependent binary
145 packages and the architecture-independent ones for the `nobreakall'
146 architectures have to be built from the source we are considering.
147 If this is not true, then these are called `out-of-date'
148 architectures and the package is ignored.
150 9. The source package must have at least one binary package, otherwise
151 it is ignored.
153 10. If the suite is unstable, the new source package must have no
154 release critical bugs which do not also apply to the testing
155 one. If this is not true, the package is ignored as `buggy'.
157 11. If there is a `force' hint for the source package, then it is
158 updated even if it is marked as ignored from the previous steps.
160 12. If the suite is {testing-,}proposed-updates, the source package can
161 be updated only if there is an explicit approval for it. Unless
162 a `force' hint exists, the new package must also be available
163 on all of the architectures for which it has binary packages in
164 testing.
166 13. If the package will be ignored, mark it as "Valid candidate",
167 otherwise mark it as "Not considered".
169 * The list of `remove' hints is processed: if the requested source
170 package is not already being updated or removed and the version
171 actually in testing is the same specified with the `remove' hint,
172 it is marked for removal.
174 * The excuses are sorted by the number of days from the last upload
175 (days-old) and by name.
177 * A list of unconsidered excuses (for which the package is not upgraded)
178 is built. Using this list, all of the excuses depending on them are
179 marked as invalid "impossible dependencies".
181 * The excuses are written in an HTML file.
182"""
183import contextlib
184import logging
185import optparse
186import os
187import sys
188import time
189from collections import defaultdict
190from collections.abc import Iterator
191from functools import reduce
192from itertools import chain
193from operator import attrgetter
194from typing import TYPE_CHECKING, Any, Optional, cast
196import apt_pkg
198from britney2 import BinaryPackage, BinaryPackageId, MultiArch, SourcePackage, Suites
199from britney2.excusefinder import ExcuseFinder
200from britney2.hints import Hint, HintCollection, HintParser
201from britney2.inputs.suiteloader import (
202 DebMirrorLikeSuiteContentLoader,
203 MissingRequiredConfigurationError,
204)
205from britney2.installability.builder import build_installability_tester
206from britney2.installability.solver import InstallabilitySolver
207from britney2.migration import MigrationManager
208from britney2.migrationitem import MigrationItem, MigrationItemFactory
209from britney2.policies.autopkgtest import AutopkgtestPolicy
210from britney2.policies.lintian import LintianPolicy
211from britney2.policies.policy import (
212 AgePolicy,
213 BlockPolicy,
214 BuildDependsPolicy,
215 BuiltOnBuilddPolicy,
216 BuiltUsingPolicy,
217 DependsPolicy,
218 ImplicitDependencyPolicy,
219 PiupartsPolicy,
220 PolicyEngine,
221 PolicyLoadRequest,
222 RCBugPolicy,
223 ReproduciblePolicy,
224 ReverseRemovalPolicy,
225)
226from britney2.utils import (
227 ExcusesOutputFormat,
228 MigrationConstraintException,
229 clone_nuninst,
230 compile_nuninst,
231 format_and_log_uninst,
232 is_nuninst_asgood_generous,
233 log_and_format_old_libraries,
234 newly_uninst,
235 old_libraries,
236 parse_option,
237 parse_provides,
238 read_nuninst,
239 write_excuses,
240 write_heidi,
241 write_heidi_delta,
242 write_nuninst,
243)
245if TYPE_CHECKING: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 from .excuse import Excuse
247 from .installability.tester import InstallabilityTester
248 from .installability.universe import BinaryPackageUniverse
249 from .transaction import MigrationTransactionState
252__author__ = "Fabio Tranchitella and the Debian Release Team"
253__version__ = "2.0"
256MIGRATION_POLICIES = [
257 PolicyLoadRequest.always_load(DependsPolicy),
258 PolicyLoadRequest.conditionally_load(RCBugPolicy, "rcbug_enable", True),
259 PolicyLoadRequest.conditionally_load(PiupartsPolicy, "piuparts_enable", True),
260 PolicyLoadRequest.always_load(ImplicitDependencyPolicy),
261 PolicyLoadRequest.conditionally_load(AutopkgtestPolicy, "adt_enable", True),
262 PolicyLoadRequest.conditionally_load(LintianPolicy, "lintian_enable", False),
263 PolicyLoadRequest.conditionally_load(ReproduciblePolicy, "repro_enable", False),
264 PolicyLoadRequest.conditionally_load(AgePolicy, "age_enable", True),
265 PolicyLoadRequest.always_load(BuildDependsPolicy),
266 PolicyLoadRequest.always_load(BlockPolicy),
267 PolicyLoadRequest.conditionally_load(
268 BuiltUsingPolicy, "built_using_policy_enable", True
269 ),
270 PolicyLoadRequest.conditionally_load(BuiltOnBuilddPolicy, "check_buildd", False),
271 PolicyLoadRequest.always_load(ReverseRemovalPolicy),
272]
275class Britney:
276 """Britney, the Debian testing updater script
278 This is the script that updates the testing distribution. It is executed
279 each day after the installation of the updated packages. It generates the
280 `Packages' files for the testing distribution, but it does so in an
281 intelligent manner; it tries to avoid any inconsistency and to use only
282 non-buggy packages.
284 For more documentation on this script, please read the Developers Reference.
285 """
287 HINTS_HELPERS = (
288 "easy",
289 "hint",
290 "remove",
291 "block",
292 "block-udeb",
293 "unblock",
294 "unblock-udeb",
295 "approve",
296 "remark",
297 "ignore-lintian",
298 "ignore-piuparts",
299 "ignore-rc-bugs",
300 "ignore-reproducible",
301 "ignore-reproducible-src",
302 "force-skiptest",
303 "force-badtest",
304 )
305 HINTS_STANDARD = ("urgent", "age-days") + HINTS_HELPERS
306 # ALL = {"force", "force-hint", "block-all"} | HINTS_STANDARD | registered policy hints (not covered above)
307 HINTS_ALL = "ALL"
308 pkg_universe: "BinaryPackageUniverse"
309 _inst_tester: "InstallabilityTester"
310 constraints: dict[str, list[str]]
311 suite_info: Suites
313 def __init__(self) -> None:
314 """Class constructor
316 This method initializes and populates the data lists, which contain all
317 the information needed by the other methods of the class.
318 """
320 # setup logging - provide the "short level name" (i.e. INFO -> I) that
321 # we used to use prior to using the logging module.
323 old_factory = logging.getLogRecordFactory()
324 short_level_mapping = {
325 "CRITICAL": "F",
326 "INFO": "I",
327 "WARNING": "W",
328 "ERROR": "E",
329 "DEBUG": "N",
330 }
332 def record_factory(
333 *args: Any, **kwargs: Any
334 ) -> logging.LogRecord: # pragma: no cover
335 record = old_factory(*args, **kwargs)
336 try:
337 record.shortlevelname = short_level_mapping[record.levelname]
338 except KeyError:
339 record.shortlevelname = record.levelname
340 return record
342 logging.setLogRecordFactory(record_factory)
343 logging.basicConfig(
344 format="{shortlevelname}: [{asctime}] - {message}",
345 style="{",
346 datefmt="%Y-%m-%dT%H:%M:%S%z",
347 stream=sys.stdout,
348 )
350 self.logger = logging.getLogger()
352 # Logger for "upgrade_output"; the file handler will be attached later when
353 # we are ready to open the file.
354 self.output_logger = logging.getLogger("britney2.output.upgrade_output")
355 self.output_logger.setLevel(logging.INFO)
357 # initialize the apt_pkg back-end
358 apt_pkg.init()
360 # parse the command line arguments
361 self._policy_engine = PolicyEngine()
362 self.__parse_arguments()
363 assert self.suite_info is not None # for type checking
365 self.all_selected: list[MigrationItem] = []
366 self.excuses: dict[str, "Excuse"] = {}
367 self.upgrade_me: list[MigrationItem] = []
369 if self.options.nuninst_cache: 369 ↛ 370line 369 didn't jump to line 370 because the condition on line 369 was never true
370 self.logger.info(
371 "Not building the list of non-installable packages, as requested"
372 )
373 if self.options.print_uninst:
374 nuninst = read_nuninst(
375 self.options.noninst_status, self.options.architectures
376 )
377 print("* summary")
378 print(
379 "\n".join(
380 "%4d %s" % (len(nuninst[x]), x)
381 for x in self.options.architectures
382 )
383 )
384 return
386 try:
387 constraints_file = os.path.join(
388 self.options.static_input_dir, "constraints"
389 )
390 faux_packages = os.path.join(self.options.static_input_dir, "faux-packages")
391 except AttributeError:
392 self.logger.info("The static_input_dir option is not set")
393 constraints_file = None
394 faux_packages = None
395 if faux_packages is not None and os.path.exists(faux_packages):
396 self.logger.info("Loading faux packages from %s", faux_packages)
397 self._load_faux_packages(faux_packages)
398 elif faux_packages is not None: 398 ↛ 401line 398 didn't jump to line 401 because the condition on line 398 was always true
399 self.logger.info("No Faux packages as %s does not exist", faux_packages)
401 if constraints_file is not None and os.path.exists(constraints_file):
402 self.logger.info("Loading constraints from %s", constraints_file)
403 self.constraints = self._load_constraints(constraints_file)
404 else:
405 if constraints_file is not None: 405 ↛ 409line 405 didn't jump to line 409
406 self.logger.info(
407 "No constraints as %s does not exist", constraints_file
408 )
409 self.constraints = {
410 "keep-installable": [],
411 }
413 self.logger.info("Compiling Installability tester")
414 self.pkg_universe, self._inst_tester = build_installability_tester(
415 self.suite_info, self.options.architectures
416 )
417 target_suite = self.suite_info.target_suite
418 target_suite.inst_tester = self._inst_tester
420 self.allow_uninst: dict[str, set[str | None]] = {}
421 for arch in self.options.architectures:
422 self.allow_uninst[arch] = set()
423 self._migration_item_factory: MigrationItemFactory = MigrationItemFactory(
424 self.suite_info
425 )
426 self._hint_parser: HintParser = HintParser(self._migration_item_factory)
427 self._migration_manager: MigrationManager = MigrationManager(
428 self.options,
429 self.suite_info,
430 self.all_binaries,
431 self.pkg_universe,
432 self.constraints,
433 self.allow_uninst,
434 self._migration_item_factory,
435 self.hints,
436 )
438 if not self.options.nuninst_cache: 438 ↛ 478line 438 didn't jump to line 478 because the condition on line 438 was always true
439 self.logger.info(
440 "Building the list of non-installable packages for the full archive"
441 )
442 self._inst_tester.compute_installability()
443 nuninst = compile_nuninst(
444 target_suite, self.options.architectures, self.options.nobreakall_arches
445 )
446 self.nuninst_orig: dict[str, set[str]] = nuninst
447 for arch in self.options.architectures:
448 self.logger.info(
449 "> Found %d non-installable packages for %s",
450 len(nuninst[arch]),
451 arch,
452 )
453 if self.options.print_uninst: 453 ↛ 454line 453 didn't jump to line 454 because the condition on line 453 was never true
454 self.nuninst_arch_report(nuninst, arch)
456 if self.options.print_uninst: 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true
457 print("* summary")
458 print(
459 "\n".join(
460 map(
461 lambda x: "%4d %s" % (len(nuninst[x]), x),
462 self.options.architectures,
463 )
464 )
465 )
466 return
467 else:
468 write_nuninst(self.options.noninst_status, nuninst)
470 stats = self._inst_tester.compute_stats()
471 self.logger.info("> Installability tester statistics (per architecture)")
472 for arch in self.options.architectures:
473 arch_stat = stats[arch]
474 self.logger.info("> %s", arch)
475 for stat in arch_stat.stat_summary():
476 self.logger.info("> - %s", stat)
477 else:
478 self.logger.info("Loading uninstallability counters from cache")
479 self.nuninst_orig = read_nuninst(
480 self.options.noninst_status, self.options.architectures
481 )
483 # nuninst_orig may get updated during the upgrade process
484 self.nuninst_orig_save: dict[str, set[str]] = clone_nuninst(
485 self.nuninst_orig, architectures=self.options.architectures
486 )
488 self._policy_engine.register_policy_hints(self._hint_parser)
490 try:
491 self.read_hints(self.options.hintsdir)
492 except AttributeError:
493 self.read_hints(os.path.join(self.suite_info["unstable"].path, "Hints"))
495 self._policy_engine.initialise(self, self.hints)
497 def __parse_arguments(self) -> None:
498 """Parse the command line arguments
500 This method parses and initializes the command line arguments.
501 While doing so, it preprocesses some of the options to be converted
502 in a suitable form for the other methods of the class.
503 """
504 # initialize the parser
505 parser = optparse.OptionParser(version="%prog")
506 parser.add_option(
507 "-v", "", action="count", dest="verbose", help="enable verbose output"
508 )
509 parser.add_option(
510 "-c",
511 "--config",
512 action="store",
513 dest="config",
514 default="/etc/britney.conf",
515 help="path for the configuration file",
516 )
517 parser.add_option(
518 "",
519 "--architectures",
520 action="store",
521 dest="architectures",
522 default=None,
523 help="override architectures from configuration file",
524 )
525 parser.add_option(
526 "",
527 "--actions",
528 action="store",
529 dest="actions",
530 default=None,
531 help="override the list of actions to be performed",
532 )
533 parser.add_option(
534 "",
535 "--hints",
536 action="store",
537 dest="hints",
538 default=None,
539 help="additional hints, separated by semicolons",
540 )
541 parser.add_option(
542 "",
543 "--hint-tester",
544 action="store_true",
545 dest="hint_tester",
546 default=None,
547 help="provide a command line interface to test hints",
548 )
549 parser.add_option(
550 "",
551 "--dry-run",
552 action="store_true",
553 dest="dry_run",
554 default=False,
555 help="disable all outputs to the testing directory",
556 )
557 parser.add_option(
558 "",
559 "--nuninst-cache",
560 action="store_true",
561 dest="nuninst_cache",
562 default=False,
563 help="do not build the non-installability status, use the cache from file",
564 )
565 parser.add_option(
566 "",
567 "--print-uninst",
568 action="store_true",
569 dest="print_uninst",
570 default=False,
571 help="just print a summary of uninstallable packages",
572 )
573 parser.add_option(
574 "",
575 "--compute-migrations",
576 action="store_true",
577 dest="compute_migrations",
578 default=True,
579 help="Compute which packages can migrate (the default)",
580 )
581 parser.add_option(
582 "",
583 "--no-compute-migrations",
584 action="store_false",
585 dest="compute_migrations",
586 help="Do not compute which packages can migrate.",
587 )
588 parser.add_option(
589 "",
590 "--series",
591 action="store",
592 dest="series",
593 default="",
594 help="set distribution series name",
595 )
596 parser.add_option(
597 "",
598 "--distribution",
599 action="store",
600 dest="distribution",
601 default="debian",
602 help="set distribution name",
603 )
604 (self.options, self.args) = parser.parse_args()
606 if self.options.verbose: 606 ↛ 612line 606 didn't jump to line 612 because the condition on line 606 was always true
607 if self.options.verbose > 1: 607 ↛ 608line 607 didn't jump to line 608 because the condition on line 607 was never true
608 self.logger.setLevel(logging.DEBUG)
609 else:
610 self.logger.setLevel(logging.INFO)
611 else:
612 self.logger.setLevel(logging.WARNING)
613 # Historical way to get debug information (equivalent to -vv)
614 try: # pragma: no cover
615 if int(os.environ.get("BRITNEY_DEBUG", "0")):
616 self.logger.setLevel(logging.DEBUG)
617 except ValueError: # pragma: no cover
618 pass
620 # integrity checks
621 if self.options.nuninst_cache and self.options.print_uninst: # pragma: no cover
622 self.logger.error("nuninst_cache and print_uninst are mutually exclusive!")
623 sys.exit(1)
625 # if the configuration file exists, then read it and set the additional options
626 if not os.path.isfile(self.options.config): # pragma: no cover
627 self.logger.error(
628 "Unable to read the configuration file (%s), exiting!",
629 self.options.config,
630 )
631 sys.exit(1)
633 self.HINTS: dict[str, Any] = {"command-line": self.HINTS_ALL}
634 with open(self.options.config, encoding="utf-8") as config:
635 for line in config:
636 if "=" in line and not line.strip().startswith("#"):
637 k, v = line.split("=", 1)
638 k = k.strip()
639 v = v.strip()
640 if k.startswith("HINTS_"):
641 self.HINTS[k.split("_")[1].lower()] = reduce( 641 ↛ exitline 641 didn't jump to the function exit
642 lambda x, y: x + y,
643 [
644 hasattr(self, "HINTS_" + i)
645 and getattr(self, "HINTS_" + i)
646 or (i,)
647 for i in v.split()
648 ],
649 )
650 elif not hasattr(self.options, k.lower()) or not getattr(
651 self.options, k.lower()
652 ):
653 setattr(self.options, k.lower(), v)
655 parse_option(self.options, "archall_inconsistency_allowed", to_bool=True)
656 parse_option(
657 self.options, "be_strict_with_build_deps", default=True, to_bool=True
658 )
660 suite_loader = DebMirrorLikeSuiteContentLoader(self.options)
662 try:
663 self.suite_info = suite_loader.load_suites()
664 except MissingRequiredConfigurationError as e: # pragma: no cover
665 self.logger.error(
666 "Could not load the suite content due to missing configuration: %s",
667 str(e),
668 )
669 sys.exit(1)
670 self.all_binaries = suite_loader.all_binaries()
671 self.options.components = suite_loader.components
672 self.options.architectures = suite_loader.architectures
673 self.options.nobreakall_arches = suite_loader.nobreakall_arches
674 self.options.outofsync_arches = suite_loader.outofsync_arches
675 self.options.break_arches = suite_loader.break_arches
676 self.options.new_arches = suite_loader.new_arches
677 if self.options.series == "": 677 ↛ 680line 677 didn't jump to line 680 because the condition on line 677 was always true
678 self.options.series = self.suite_info.target_suite.name
680 if self.options.heidi_output and not hasattr( 680 ↛ 685line 680 didn't jump to line 685 because the condition on line 680 was always true
681 self.options, "heidi_delta_output"
682 ):
683 self.options.heidi_delta_output = self.options.heidi_output + "Delta"
685 self.options.smooth_updates = self.options.smooth_updates.split()
687 parse_option(self.options, "ignore_cruft", to_bool=True)
688 parse_option(self.options, "check_consistency_level", default=2, to_int=True)
689 parse_option(self.options, "build_url")
691 self._policy_engine.load_policies(
692 self.options, self.suite_info, MIGRATION_POLICIES
693 )
695 @property
696 def hints(self) -> HintCollection:
697 return self._hint_parser.hints
699 def _load_faux_packages(self, faux_packages_file: str) -> None:
700 """Loads fake packages
702 In rare cases, it is useful to create a "fake" package that can be used to satisfy
703 dependencies. This is usually needed for packages that are not shipped directly
704 on this mirror but is a prerequisite for using this mirror (e.g. some vendors provide
705 non-distributable "setup" packages and contrib/non-free packages depend on these).
707 :param faux_packages_file: Path to the file containing the fake package definitions
708 """
709 tag_file = apt_pkg.TagFile(faux_packages_file)
710 get_field = tag_file.section.get
711 step = tag_file.step
712 no = 0
713 pri_source_suite = self.suite_info.primary_source_suite
714 target_suite = self.suite_info.target_suite
716 while step():
717 no += 1
718 pkg_name = get_field("Package", None)
719 if pkg_name is None: # pragma: no cover
720 raise ValueError(
721 "Missing Package field in paragraph %d (file %s)"
722 % (no, faux_packages_file)
723 )
724 pkg_name = sys.intern(pkg_name)
725 version = sys.intern(get_field("Version", "1.0-1"))
726 provides_raw = get_field("Provides")
727 archs_raw = get_field("Architecture", None)
728 component = get_field("Component", "non-free")
729 if archs_raw: 729 ↛ 730line 729 didn't jump to line 730 because the condition on line 729 was never true
730 archs = archs_raw.split()
731 else:
732 archs = self.options.architectures
733 faux_section = "faux"
734 if component != "main": 734 ↛ 736line 734 didn't jump to line 736 because the condition on line 734 was always true
735 faux_section = "%s/faux" % component
736 src_data = SourcePackage(
737 pkg_name,
738 version,
739 sys.intern(faux_section),
740 set(),
741 None,
742 True,
743 None,
744 None,
745 [],
746 [],
747 )
749 target_suite.sources[pkg_name] = src_data
750 pri_source_suite.sources[pkg_name] = src_data
752 for arch in archs:
753 pkg_id = BinaryPackageId(pkg_name, version, arch)
754 if provides_raw: 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 provides = parse_provides(
756 provides_raw, pkg_id=pkg_id, logger=self.logger
757 )
758 else:
759 provides = None
760 bin_data = BinaryPackage(
761 faux_section,
762 pkg_name,
763 version,
764 arch,
765 MultiArch.from_str(get_field("Multi-Arch")),
766 None,
767 None,
768 provides,
769 False,
770 pkg_id,
771 None,
772 )
774 src_data.binaries.add(pkg_id)
775 target_suite.binaries[arch][pkg_name] = bin_data
776 pri_source_suite.binaries[arch][pkg_name] = bin_data
778 # register provided packages with the target suite provides table
779 for provided_pkg, provided_version, _ in ( 779 ↛ 782line 779 didn't jump to line 782
780 bin_data.provides if bin_data.provides is not None else []
781 ):
782 target_suite.provides_table[arch][provided_pkg].add(
783 (pkg_name, provided_version)
784 )
786 self.all_binaries[pkg_id] = bin_data
788 def _load_constraints(self, constraints_file: str) -> dict[str, list[str]]:
789 """Loads configurable constraints
791 The constraints file can contain extra rules that Britney should attempt
792 to satisfy. Examples can be "keep package X in testing and ensure it is
793 installable".
795 :param constraints_file: Path to the file containing the constraints
796 """
797 tag_file = apt_pkg.TagFile(constraints_file)
798 get_field = tag_file.section.get
799 step = tag_file.step
800 no = 0
801 faux_version = sys.intern("1")
802 faux_section = sys.intern("faux")
803 keep_installable: list[str] = []
804 constraints = {"keep-installable": keep_installable}
805 pri_source_suite = self.suite_info.primary_source_suite
806 target_suite = self.suite_info.target_suite
808 while step():
809 no += 1
810 pkg_name = get_field("Fake-Package-Name", None)
811 if pkg_name is None: # pragma: no cover
812 raise ValueError(
813 "Missing Fake-Package-Name field in paragraph %d (file %s)"
814 % (no, constraints_file)
815 )
816 pkg_name = sys.intern(pkg_name)
818 def mandatory_field(x: str) -> str:
819 v: str = get_field(x, None)
820 if v is None: # pragma: no cover
821 raise ValueError(
822 "Missing %s field for %s (file %s)"
823 % (x, pkg_name, constraints_file)
824 )
825 return v
827 constraint = mandatory_field("Constraint")
828 if constraint not in {"present-and-installable"}: # pragma: no cover
829 raise ValueError(
830 "Unsupported constraint %s for %s (file %s)"
831 % (constraint, pkg_name, constraints_file)
832 )
834 self.logger.info(" - constraint %s", pkg_name)
836 pkg_list = [
837 x.strip()
838 for x in mandatory_field("Package-List").split("\n")
839 if x.strip() != "" and not x.strip().startswith("#")
840 ]
841 src_data = SourcePackage(
842 pkg_name,
843 faux_version,
844 faux_section,
845 set(),
846 None,
847 True,
848 None,
849 None,
850 [],
851 [],
852 )
853 target_suite.sources[pkg_name] = src_data
854 pri_source_suite.sources[pkg_name] = src_data
855 keep_installable.append(pkg_name)
856 for arch in self.options.architectures:
857 deps = []
858 for pkg_spec in pkg_list:
859 s = pkg_spec.split(None, 1)
860 if len(s) == 1:
861 deps.append(s[0])
862 else:
863 pkg, arch_res = s
864 if not (
865 arch_res.startswith("[") and arch_res.endswith("]")
866 ): # pragma: no cover
867 raise ValueError(
868 "Invalid arch-restriction on %s - should be [arch1 arch2] (for %s file %s)"
869 % (pkg, pkg_name, constraints_file)
870 )
871 arch_res_l = arch_res[1:-1].split()
872 if not arch_res_l: # pragma: no cover
873 msg = "Empty arch-restriction for %s: Uses comma or negation (for %s file %s)"
874 raise ValueError(msg % (pkg, pkg_name, constraints_file))
875 for a in arch_res_l:
876 if a == arch:
877 deps.append(pkg)
878 elif "," in a or "!" in a: # pragma: no cover
879 msg = "Invalid arch-restriction for %s: Uses comma or negation (for %s file %s)"
880 raise ValueError(
881 msg % (pkg, pkg_name, constraints_file)
882 )
883 pkg_id = BinaryPackageId(pkg_name, faux_version, arch)
884 bin_data = BinaryPackage(
885 faux_section,
886 pkg_name,
887 faux_version,
888 arch,
889 MultiArch.NO,
890 ", ".join(deps),
891 None,
892 [],
893 False,
894 pkg_id,
895 [],
896 )
897 src_data.binaries.add(pkg_id)
898 target_suite.binaries[arch][pkg_name] = bin_data
899 pri_source_suite.binaries[arch][pkg_name] = bin_data
900 self.all_binaries[pkg_id] = bin_data
902 return constraints
904 # Data reading/writing methods
905 # ----------------------------
907 def read_hints(self, hintsdir: str) -> None:
908 """Read the hint commands from the specified directory
910 The hint commands are read from the files contained in the directory
911 specified by the `hintsdir' parameter.
912 The names of the files have to be the same as the authorized users
913 for the hints.
915 The file contains rows with the format:
917 <command> <package-name>[/<version>]
919 The method returns a dictionary where the key is the command, and
920 the value is the list of affected packages.
921 """
923 for who in self.HINTS.keys():
924 if who == "command-line":
925 lines = self.options.hints and self.options.hints.split(";") or ()
926 filename = "<cmd-line>"
927 self._hint_parser.parse_hints(who, self.HINTS[who], filename, lines)
928 else:
929 filename = os.path.join(hintsdir, who)
930 if not os.path.isfile(filename): 930 ↛ 931line 930 didn't jump to line 931 because the condition on line 930 was never true
931 self.logger.error(
932 "Cannot read hints list from %s, no such file!", filename
933 )
934 continue
935 self.logger.info("Loading hints list from %s", filename)
936 with open(filename, encoding="utf-8") as f:
937 self._hint_parser.parse_hints(who, self.HINTS[who], filename, f)
939 hints = self._hint_parser.hints
941 for x in (
942 "block",
943 "block-all",
944 "block-udeb",
945 "unblock",
946 "unblock-udeb",
947 "force",
948 "urgent",
949 "remove",
950 "age-days",
951 ):
952 z: dict[str | None, dict[str | None, tuple[Hint, str]]] = defaultdict(dict)
953 for hint in hints[x]:
954 package = hint.package
955 architecture = hint.architecture
956 key = (hint, hint.user)
957 if (
958 package in z
959 and architecture in z[package]
960 and z[package][architecture] != key
961 ):
962 hint2 = z[package][architecture][0]
963 if x in ("unblock", "unblock-udeb", "age-days"): 963 ↛ 995line 963 didn't jump to line 995 because the condition on line 963 was always true
964 assert hint.version is not None
965 assert hint2.version is not None
966 if apt_pkg.version_compare(hint2.version, hint.version) < 0:
967 # This hint is for a newer version, so discard the old one
968 self.logger.warning(
969 "Overriding %s[%s] = ('%s', '%s', '%s') with ('%s', '%s', '%s')",
970 x,
971 package,
972 hint2.version,
973 hint2.architecture,
974 hint2.user,
975 hint.version,
976 hint.architecture,
977 hint.user,
978 )
979 hint2.set_active(False)
980 else:
981 # This hint is for an older version, so ignore it in favour of the new one
982 self.logger.warning(
983 "Ignoring %s[%s] = ('%s', '%s', '%s'), ('%s', '%s', '%s') is higher or equal",
984 x,
985 package,
986 hint.version,
987 hint.architecture,
988 hint.user,
989 hint2.version,
990 hint2.architecture,
991 hint2.user,
992 )
993 hint.set_active(False)
994 else:
995 self.logger.warning(
996 "Overriding %s[%s] = ('%s', '%s') with ('%s', '%s')",
997 x,
998 package,
999 hint2.user,
1000 hint2,
1001 hint.user,
1002 hint,
1003 )
1004 hint2.set_active(False)
1006 z[package][architecture] = key
1008 for hint in hints["allow-uninst"]:
1009 if hint.architecture == "source":
1010 for arch in self.options.architectures:
1011 self.allow_uninst[arch].add(hint.package)
1012 else:
1013 assert hint.architecture is not None
1014 self.allow_uninst[hint.architecture].add(hint.package)
1016 # Sanity check the hints hash
1017 if len(hints["block"]) == 0 and len(hints["block-udeb"]) == 0: 1017 ↛ 1018line 1017 didn't jump to line 1018 because the condition on line 1017 was never true
1018 self.logger.warning("WARNING: No block hints at all, not even udeb ones!")
1020 # Remove all hints that were set inactive.
1021 # We don't need to keep unused hints in memory.
1022 hints.remove_inactive_hints()
1024 def write_excuses(self) -> None:
1025 """Produce and write the update excuses
1027 This method handles the update excuses generation: the packages are
1028 looked at to determine whether they are valid candidates. For the details
1029 of this procedure, please refer to the module docstring.
1030 """
1032 self.logger.info("Update Excuses generation started")
1034 mi_factory = self._migration_item_factory
1035 excusefinder = ExcuseFinder(
1036 self.options,
1037 self.suite_info,
1038 self.all_binaries,
1039 self.pkg_universe,
1040 self._policy_engine,
1041 mi_factory,
1042 self.hints,
1043 )
1045 excuses, upgrade_me = excusefinder.find_actionable_excuses()
1046 self.excuses = excuses
1048 # sort the list of candidates
1049 self.upgrade_me = sorted(upgrade_me)
1050 old_lib_removals = old_libraries(
1051 mi_factory, self.suite_info, self.options.outofsync_arches
1052 )
1053 self.upgrade_me.extend(old_lib_removals)
1054 self.output_logger.info(
1055 "List of old libraries added to upgrade_me (%d):", len(old_lib_removals)
1056 )
1057 log_and_format_old_libraries(self.output_logger, old_lib_removals)
1059 # write excuses to the output file
1060 if not self.options.dry_run: 1060 ↛ 1077line 1060 didn't jump to line 1077 because the condition on line 1060 was always true
1061 self.logger.info("> Writing Excuses to %s", self.options.excuses_output)
1062 write_excuses(
1063 excuses,
1064 self.options.excuses_output,
1065 output_format=ExcusesOutputFormat.LEGACY_HTML,
1066 )
1067 if hasattr(self.options, "excuses_yaml_output"): 1067 ↛ 1077line 1067 didn't jump to line 1077 because the condition on line 1067 was always true
1068 self.logger.info(
1069 "> Writing YAML Excuses to %s", self.options.excuses_yaml_output
1070 )
1071 write_excuses(
1072 excuses,
1073 self.options.excuses_yaml_output,
1074 output_format=ExcusesOutputFormat.YAML,
1075 )
1077 self.logger.info("Update Excuses generation completed")
1079 # Upgrade run
1080 # -----------
1082 def eval_nuninst(
1083 self,
1084 nuninst: dict[str, set[str]],
1085 original: dict[str, set[str]] | None = None,
1086 ) -> str:
1087 """Return a string which represents the uninstallability counters
1089 This method returns a string which represents the uninstallability
1090 counters reading the uninstallability statistics `nuninst` and, if
1091 present, merging the results with the `original` one.
1093 An example of the output string is:
1094 1+2: i-0:a-0:a-0:h-0:i-1:m-0:m-0:p-0:a-0:m-0:s-2:s-0
1096 where the first part is the number of broken packages in non-break
1097 architectures + the total number of broken packages for all the
1098 architectures.
1099 """
1100 res = []
1101 total = 0
1102 totalbreak = 0
1103 for arch in self.options.architectures:
1104 if arch in nuninst: 1104 ↛ 1106line 1104 didn't jump to line 1106 because the condition on line 1104 was always true
1105 n = len(nuninst[arch])
1106 elif original and arch in original:
1107 n = len(original[arch])
1108 else:
1109 continue
1110 if arch in self.options.break_arches:
1111 totalbreak = totalbreak + n
1112 else:
1113 total = total + n
1114 res.append("%s-%d" % (arch[0], n))
1115 return "%d+%d: %s" % (total, totalbreak, ":".join(res))
1117 def iter_packages(
1118 self,
1119 packages: list[MigrationItem],
1120 selected: list[MigrationItem],
1121 nuninst: dict[str, set[str]] | None = None,
1122 ) -> tuple[dict[str, set[str]] | None, list[MigrationItem]]:
1123 """Iter on the list of actions and apply them one-by-one
1125 This method applies the changes from `packages` to testing, checking the uninstallability
1126 counters for every action performed. If the action does not improve them, it is reverted.
1127 The method returns the new uninstallability counters and the remaining actions if the
1128 final result is successful, otherwise (None, []).
1130 :param selected: list of MigrationItem?
1131 :param nuninst: dict with sets ? of ? per architecture
1132 """
1133 assert self.suite_info is not None # for type checking
1134 group_info = {}
1135 rescheduled_packages = packages
1136 maybe_rescheduled_packages: list[MigrationItem] = []
1137 output_logger = self.output_logger
1138 solver = InstallabilitySolver(self.pkg_universe, self._inst_tester)
1139 mm = self._migration_manager
1140 target_suite = self.suite_info.target_suite
1142 for y in sorted((y for y in packages), key=attrgetter("uvname")):
1143 try:
1144 _, updates, rms, _ = mm.compute_groups(y)
1145 result = (y, sorted(updates), sorted(rms))
1146 group_info[y] = result
1147 except MigrationConstraintException as e:
1148 rescheduled_packages.remove(y)
1149 output_logger.info("not adding package to list: %s", (y.package))
1150 output_logger.info(" got exception: %s" % (repr(e)))
1152 if nuninst:
1153 nuninst_orig = nuninst
1154 else:
1155 nuninst_orig = self.nuninst_orig
1157 nuninst_last_accepted = nuninst_orig
1159 output_logger.info(
1160 "recur: [] %s %d/0", ",".join(x.uvname for x in selected), len(packages)
1161 )
1162 while rescheduled_packages:
1163 groups = [group_info[x] for x in rescheduled_packages]
1164 worklist = solver.solve_groups(groups)
1165 rescheduled_packages = []
1167 worklist.reverse()
1169 while worklist:
1170 comp = worklist.pop()
1171 comp_name = " ".join(item.uvname for item in comp)
1172 output_logger.info("trying: %s" % comp_name)
1173 with mm.start_transaction() as transaction:
1174 accepted = False
1175 try:
1176 (
1177 accepted,
1178 nuninst_after,
1179 failed_arch,
1180 new_cruft,
1181 ) = mm.migrate_items_to_target_suite(
1182 comp, nuninst_last_accepted
1183 )
1184 if accepted:
1185 selected.extend(comp)
1186 transaction.commit()
1187 output_logger.info("accepted: %s", comp_name)
1188 output_logger.info(
1189 " ori: %s", self.eval_nuninst(nuninst_orig)
1190 )
1191 output_logger.info(
1192 " pre: %s", self.eval_nuninst(nuninst_last_accepted)
1193 )
1194 output_logger.info(
1195 " now: %s", self.eval_nuninst(nuninst_after)
1196 )
1197 if new_cruft:
1198 output_logger.info(
1199 " added new cruft items to list: %s",
1200 " ".join(x.uvname for x in sorted(new_cruft)),
1201 )
1203 if len(selected) <= 20:
1204 output_logger.info(
1205 " all: %s", " ".join(x.uvname for x in selected)
1206 )
1207 else:
1208 output_logger.info(
1209 " most: (%d) .. %s",
1210 len(selected),
1211 " ".join(x.uvname for x in selected[-20:]),
1212 )
1213 if self.options.check_consistency_level >= 3: 1213 ↛ 1214line 1213 didn't jump to line 1214 because the condition on line 1213 was never true
1214 target_suite.check_suite_source_pkg_consistency(
1215 "iter_packages after commit"
1216 )
1217 nuninst_last_accepted = nuninst_after
1218 for cruft_item in new_cruft:
1219 try:
1220 _, updates, rms, _ = mm.compute_groups(cruft_item)
1221 result = (cruft_item, sorted(updates), sorted(rms))
1222 group_info[cruft_item] = result
1223 worklist.append([cruft_item])
1224 except MigrationConstraintException as e:
1225 output_logger.info(
1226 " got exception adding cruft item %s to list: %s"
1227 % (cruft_item.uvname, repr(e))
1228 )
1229 rescheduled_packages.extend(maybe_rescheduled_packages)
1230 maybe_rescheduled_packages.clear()
1231 else:
1232 transaction.rollback()
1233 assert failed_arch # type checking
1234 broken = sorted(
1235 b
1236 for b in nuninst_after[failed_arch]
1237 if b not in nuninst_last_accepted[failed_arch]
1238 )
1239 compare_nuninst = None
1240 if any(
1241 item for item in comp if item.architecture != "source"
1242 ):
1243 compare_nuninst = nuninst_last_accepted
1244 # NB: try_migration already reverted this for us, so just print the results and move on
1245 output_logger.info(
1246 "skipped: %s (%d, %d, %d)",
1247 comp_name,
1248 len(rescheduled_packages),
1249 len(maybe_rescheduled_packages),
1250 len(worklist),
1251 )
1252 output_logger.info(
1253 " got: %s",
1254 self.eval_nuninst(nuninst_after, compare_nuninst),
1255 )
1256 output_logger.info(
1257 " * %s: %s", failed_arch, ", ".join(broken)
1258 )
1259 if self.options.check_consistency_level >= 3: 1259 ↛ 1260line 1259 didn't jump to line 1260 because the condition on line 1259 was never true
1260 target_suite.check_suite_source_pkg_consistency(
1261 "iter_package after rollback (not accepted)"
1262 )
1264 except MigrationConstraintException as e:
1265 transaction.rollback()
1266 output_logger.info(
1267 "skipped: %s (%d, %d, %d)",
1268 comp_name,
1269 len(rescheduled_packages),
1270 len(maybe_rescheduled_packages),
1271 len(worklist),
1272 )
1273 output_logger.info(" got exception: %s" % (repr(e)))
1274 if self.options.check_consistency_level >= 3: 1274 ↛ 1275line 1274 didn't jump to line 1275 because the condition on line 1274 was never true
1275 target_suite.check_suite_source_pkg_consistency(
1276 "iter_package after rollback (MigrationConstraintException)"
1277 )
1279 if not accepted:
1280 if len(comp) > 1:
1281 output_logger.info(
1282 " - splitting the component into single items and retrying them"
1283 )
1284 worklist.extend([item] for item in comp)
1285 else:
1286 maybe_rescheduled_packages.append(comp[0])
1288 output_logger.info(" finish: [%s]", ",".join(x.uvname for x in selected))
1289 output_logger.info("endloop: %s", self.eval_nuninst(self.nuninst_orig))
1290 output_logger.info(" now: %s", self.eval_nuninst(nuninst_last_accepted))
1291 format_and_log_uninst(
1292 output_logger,
1293 self.options.architectures,
1294 newly_uninst(self.nuninst_orig, nuninst_last_accepted),
1295 )
1296 output_logger.info("")
1298 return (nuninst_last_accepted, maybe_rescheduled_packages)
1300 def do_all(
1301 self,
1302 hinttype: str | None = None,
1303 init: list[MigrationItem] | None = None,
1304 actions: list[MigrationItem] | None = None,
1305 ) -> None:
1306 """Testing update runner
1308 This method tries to update testing checking the uninstallability
1309 counters before and after the actions to decide if the update was
1310 successful or not.
1311 """
1312 selected = []
1313 if actions:
1314 upgrade_me = actions[:]
1315 else:
1316 upgrade_me = self.upgrade_me[:]
1317 nuninst_start = self.nuninst_orig
1318 output_logger = self.output_logger
1319 target_suite = self.suite_info.target_suite
1321 # these are special parameters for hints processing
1322 force = False
1323 recurse = True
1324 nuninst_end = None
1325 extra: list[MigrationItem] = []
1326 mm = self._migration_manager
1328 if hinttype == "easy" or hinttype == "force-hint":
1329 force = hinttype == "force-hint"
1330 recurse = False
1332 # if we have a list of initial packages, check them
1333 if init:
1334 for x in init:
1335 if x not in upgrade_me:
1336 output_logger.warning(
1337 "failed: %s is not a valid candidate (or it already migrated)",
1338 x.uvname,
1339 )
1340 return None
1341 selected.append(x)
1342 upgrade_me.remove(x)
1344 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1345 output_logger.info("orig: %s", self.eval_nuninst(nuninst_start))
1347 if not (init and not force):
1348 # No "outer" transaction needed as we will never need to rollback
1349 # (e.g. "force-hint" or a regular "main run"). Emulate the start_transaction
1350 # call from the MigrationManager, so the rest of the code follows the
1351 # same flow regardless of whether we need the transaction or not.
1353 @contextlib.contextmanager
1354 def _start_transaction() -> Iterator[Optional["MigrationTransactionState"]]:
1355 yield None
1357 else:
1358 # We will need to be able to roll back (e.g. easy or a "hint"-hint)
1359 _start_transaction = mm.start_transaction
1361 with _start_transaction() as transaction:
1362 if init:
1363 # init => a hint (e.g. "easy") - so do the hint run
1364 (_, nuninst_end, _, new_cruft) = mm.migrate_items_to_target_suite(
1365 selected, self.nuninst_orig, stop_on_first_regression=False
1366 )
1368 if recurse:
1369 # Ensure upgrade_me and selected do not overlap, if we
1370 # follow-up with a recurse ("hint"-hint).
1371 selected_set = set(selected)
1372 upgrade_me = [x for x in upgrade_me if x not in selected_set]
1373 else:
1374 # On non-recursive hints check for cruft and purge it proactively in case it "fixes" the hint.
1375 cruft = [x for x in upgrade_me if x.is_cruft_removal]
1376 if new_cruft:
1377 output_logger.info(
1378 "Change added new cruft items to list: %s",
1379 " ".join(x.uvname for x in sorted(new_cruft)),
1380 )
1381 cruft.extend(new_cruft)
1382 if cruft:
1383 output_logger.info("Checking if changes enables cruft removal")
1384 (nuninst_end, remaining_cruft) = self.iter_packages(
1385 cruft, selected, nuninst=nuninst_end
1386 )
1387 output_logger.info(
1388 "Removed %d of %d cruft item(s) after the changes",
1389 len(cruft) - len(remaining_cruft),
1390 len(cruft),
1391 )
1392 new_cruft.difference_update(remaining_cruft)
1394 # Add new cruft items regardless of whether we recurse. A future run might clean
1395 # them for us.
1396 upgrade_me.extend(new_cruft)
1398 if recurse:
1399 # Either the main run or the recursive run of a "hint"-hint.
1400 (nuninst_end, extra) = self.iter_packages(
1401 upgrade_me, selected, nuninst=nuninst_end
1402 )
1404 assert nuninst_end is not None
1405 nuninst_end_str = self.eval_nuninst(nuninst_end)
1407 if not recurse:
1408 # easy or force-hint
1409 output_logger.info("easy: %s", nuninst_end_str)
1411 if not force:
1412 format_and_log_uninst(
1413 self.output_logger,
1414 self.options.architectures,
1415 newly_uninst(nuninst_start, nuninst_end),
1416 )
1418 if force:
1419 # Force implies "unconditionally better"
1420 better = True
1421 else:
1422 break_arches: set[str] = set(self.options.break_arches)
1423 if all(x.architecture in break_arches for x in selected):
1424 # If we only migrated items from break-arches, then we
1425 # do not allow any regressions on these architectures.
1426 # This usually only happens with hints
1427 break_arches = set()
1428 better = is_nuninst_asgood_generous(
1429 self.constraints,
1430 self.allow_uninst,
1431 self.options.architectures,
1432 self.nuninst_orig,
1433 nuninst_end,
1434 break_arches,
1435 )
1437 if better:
1438 # Result accepted either by force or by being better than the original result.
1439 output_logger.info(
1440 "final: %s", ",".join(sorted(x.uvname for x in selected))
1441 )
1442 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1443 output_logger.info(" orig: %s", self.eval_nuninst(self.nuninst_orig))
1444 output_logger.info(" end: %s", nuninst_end_str)
1445 if force:
1446 broken = newly_uninst(nuninst_start, nuninst_end)
1447 if broken:
1448 output_logger.warning("force breaks:")
1449 format_and_log_uninst(
1450 self.output_logger,
1451 self.options.architectures,
1452 broken,
1453 loglevel=logging.WARNING,
1454 )
1455 else:
1456 output_logger.info("force did not break any packages")
1457 output_logger.info(
1458 "SUCCESS (%d/%d)", len(actions or self.upgrade_me), len(extra)
1459 )
1460 self.nuninst_orig = nuninst_end
1461 self.all_selected += selected
1462 if transaction:
1463 transaction.commit()
1464 if self.options.check_consistency_level >= 2: 1464 ↛ 1468line 1464 didn't jump to line 1468 because the condition on line 1464 was always true
1465 target_suite.check_suite_source_pkg_consistency(
1466 "do_all after commit"
1467 )
1468 if not actions:
1469 if recurse:
1470 self.upgrade_me = extra
1471 else:
1472 selected_set = set(selected)
1473 self.upgrade_me = [
1474 x for x in self.upgrade_me if x not in selected_set
1475 ]
1476 else:
1477 output_logger.info("FAILED\n")
1478 if not transaction: 1478 ↛ 1482line 1478 didn't jump to line 1482 because the condition on line 1478 was never true
1479 # if we 'FAILED', but we cannot rollback, we will probably
1480 # leave a broken state behind
1481 # this should not happen
1482 raise AssertionError("do_all FAILED but no transaction to rollback")
1483 transaction.rollback()
1484 if self.options.check_consistency_level >= 2: 1484 ↛ 1361line 1484 didn't jump to line 1361
1485 target_suite.check_suite_source_pkg_consistency(
1486 "do_all after rollback"
1487 )
1489 output_logger.info("")
1491 def assert_nuninst_is_correct(self) -> None:
1492 self.logger.info("> Update complete - Verifying non-installability counters")
1494 cached_nuninst = self.nuninst_orig
1495 self._inst_tester.compute_installability()
1496 computed_nuninst = compile_nuninst(
1497 self.suite_info.target_suite,
1498 self.options.architectures,
1499 self.options.nobreakall_arches,
1500 )
1501 if cached_nuninst != computed_nuninst: # pragma: no cover
1502 only_on_break_archs = True
1503 msg_l = [
1504 "==================== NUNINST OUT OF SYNC ========================="
1505 ]
1506 for arch in self.options.architectures:
1507 expected_nuninst = set(cached_nuninst[arch])
1508 actual_nuninst = set(computed_nuninst[arch])
1509 false_negatives = actual_nuninst - expected_nuninst
1510 false_positives = expected_nuninst - actual_nuninst
1511 # Britney does not quite work correctly with
1512 # break/fucked arches, so ignore issues there for now.
1513 if (
1514 false_negatives or false_positives
1515 ) and arch not in self.options.break_arches:
1516 only_on_break_archs = False
1517 if false_negatives:
1518 msg_l.append(f" {arch} - unnoticed nuninst: {str(false_negatives)}")
1519 if false_positives:
1520 msg_l.append(f" {arch} - invalid nuninst: {str(false_positives)}")
1521 if false_negatives or false_positives:
1522 msg_l.append(
1523 f" {arch} - actual nuninst: {str(sorted(actual_nuninst))}"
1524 )
1525 msg_l.append(msg_l[0])
1526 for msg in msg_l:
1527 if only_on_break_archs:
1528 self.logger.warning(msg)
1529 else:
1530 self.logger.error(msg)
1531 if not only_on_break_archs:
1532 raise AssertionError("NUNINST OUT OF SYNC")
1533 else:
1534 self.logger.warning("Nuninst is out of sync on some break arches")
1536 self.logger.info("> All non-installability counters are ok")
1538 def upgrade_testing(self) -> None:
1539 """Upgrade testing using the packages from the source suites
1541 This method tries to upgrade testing using the packages from the
1542 source suites.
1543 Before running the do_all method, it tries the easy and force-hint
1544 commands.
1545 """
1547 output_logger = self.output_logger
1548 self.logger.info("Starting the upgrade test")
1549 output_logger.info(
1550 "Generated on: %s",
1551 time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time())),
1552 )
1553 output_logger.info("Arch order is: %s", ", ".join(self.options.architectures))
1555 if not self.options.actions: 1555 ↛ 1566line 1555 didn't jump to line 1566 because the condition on line 1555 was always true
1556 # process `easy' hints
1557 for x in self.hints["easy"]:
1558 self.do_hint("easy", x.user, x.packages)
1560 # process `force-hint' hints
1561 for x in self.hints["force-hint"]:
1562 self.do_hint("force-hint", x.user, x.packages)
1564 # run the first round of the upgrade
1565 # - do separate runs for break arches
1566 allpackages = []
1567 normpackages = self.upgrade_me[:]
1568 archpackages = {}
1569 for a in self.options.break_arches:
1570 archpackages[a] = [p for p in normpackages if p.architecture == a]
1571 normpackages = [p for p in normpackages if p.architecture != a]
1572 self.upgrade_me = normpackages
1573 output_logger.info("info: main run")
1574 self.do_all()
1575 allpackages += self.upgrade_me
1576 for a in self.options.break_arches:
1577 backup = self.options.break_arches
1578 self.options.break_arches = " ".join(
1579 x for x in self.options.break_arches if x != a
1580 )
1581 self.upgrade_me = archpackages[a]
1582 output_logger.info("info: broken arch run for %s", a)
1583 self.do_all()
1584 allpackages += self.upgrade_me
1585 self.options.break_arches = backup
1586 self.upgrade_me = allpackages
1588 if self.options.actions: 1588 ↛ 1589line 1588 didn't jump to line 1589 because the condition on line 1588 was never true
1589 self.printuninstchange()
1590 return
1592 # process `hint' hints
1593 hintcnt = 0
1594 for x in self.hints["hint"][:50]:
1595 if hintcnt > 50: 1595 ↛ 1596line 1595 didn't jump to line 1596 because the condition on line 1595 was never true
1596 output_logger.info("Skipping remaining hints...")
1597 break
1598 if self.do_hint("hint", x.user, x.packages): 1598 ↛ 1594line 1598 didn't jump to line 1594 because the condition on line 1598 was always true
1599 hintcnt += 1
1601 # run the auto hinter
1602 self.run_auto_hinter()
1604 if getattr(self.options, "remove_obsolete", "yes") == "yes":
1605 # obsolete source packages
1606 # a package is obsolete if none of the binary packages in testing
1607 # are built by it
1608 self.logger.info(
1609 "> Removing obsolete source packages from the target suite"
1610 )
1611 # local copies for performance
1612 target_suite = self.suite_info.target_suite
1613 sources_t = target_suite.sources
1614 binaries_t = target_suite.binaries
1615 mi_factory = self._migration_item_factory
1616 used = {
1617 binaries_t[arch][binary].source
1618 for arch in binaries_t
1619 for binary in binaries_t[arch]
1620 if not binary.endswith("-faux-build-depends")
1621 }
1622 removals = [
1623 mi_factory.parse_item(
1624 f"-{source}/{sources_t[source].version}", auto_correct=False
1625 )
1626 for source in sources_t
1627 if source not in used
1628 ]
1629 if removals:
1630 output_logger.info(
1631 "Removing obsolete source packages from the target suite (%d):",
1632 len(removals),
1633 )
1634 self.do_all(actions=removals)
1636 # smooth updates
1637 removals = old_libraries(
1638 self._migration_item_factory, self.suite_info, self.options.outofsync_arches
1639 )
1640 if removals:
1641 output_logger.info(
1642 "Removing packages left in the target suite (e.g. smooth updates or cruft)"
1643 )
1644 log_and_format_old_libraries(self.output_logger, removals)
1645 self.do_all(actions=removals)
1646 removals = old_libraries(
1647 self._migration_item_factory,
1648 self.suite_info,
1649 self.options.outofsync_arches,
1650 )
1652 output_logger.info(
1653 "List of old libraries in the target suite (%d):", len(removals)
1654 )
1655 log_and_format_old_libraries(self.output_logger, removals)
1657 self.printuninstchange()
1658 if self.options.check_consistency_level >= 1: 1658 ↛ 1664line 1658 didn't jump to line 1664 because the condition on line 1658 was always true
1659 target_suite = self.suite_info.target_suite
1660 self.assert_nuninst_is_correct()
1661 target_suite.check_suite_source_pkg_consistency("end")
1663 # output files
1664 if self.options.heidi_output and not self.options.dry_run: 1664 ↛ 1678line 1664 didn't jump to line 1678 because the condition on line 1664 was always true
1665 target_suite = self.suite_info.target_suite
1667 # write HeidiResult
1668 self.logger.info("Writing Heidi results to %s", self.options.heidi_output)
1669 write_heidi(
1670 self.options.heidi_output,
1671 target_suite,
1672 outofsync_arches=self.options.outofsync_arches,
1673 )
1675 self.logger.info("Writing delta to %s", self.options.heidi_delta_output)
1676 write_heidi_delta(self.options.heidi_delta_output, self.all_selected)
1678 self.logger.info("Test completed!")
1680 def printuninstchange(self) -> None:
1681 self.logger.info("Checking for newly uninstallable packages")
1682 uninst = newly_uninst(self.nuninst_orig_save, self.nuninst_orig)
1684 if uninst:
1685 self.output_logger.info("")
1686 self.output_logger.info(
1687 "Newly uninstallable packages in the target suite (arch:all on BREAKALL_ARCHES not shown)"
1688 )
1689 format_and_log_uninst(
1690 self.output_logger,
1691 self.options.architectures,
1692 uninst,
1693 loglevel=logging.WARNING,
1694 )
1696 def hint_tester(self) -> None:
1697 """Run a command line interface to test hints
1699 This method provides a command line interface for the release team to
1700 try hints and evaluate the results.
1701 """
1702 import readline
1704 from britney2.completer import Completer
1706 histfile = os.path.expanduser("~/.britney2_history")
1707 if os.path.exists(histfile):
1708 readline.read_history_file(histfile)
1710 readline.parse_and_bind("tab: complete")
1711 readline.set_completer(Completer(self).completer)
1712 # Package names can contain "-" and we use "/" in our presentation of them as well,
1713 # so ensure readline does not split on these characters.
1714 readline.set_completer_delims(
1715 readline.get_completer_delims().replace("-", "").replace("/", "")
1716 )
1718 known_hints = self._hint_parser.registered_hint_names
1720 print("Britney hint tester")
1721 print()
1722 print(
1723 "Besides inputting known britney hints, the follow commands are also available"
1724 )
1725 print(" * quit/exit - terminates the shell")
1726 print(
1727 " * python-console - jump into an interactive python shell (with the current loaded dataset)"
1728 )
1729 print()
1731 while True:
1732 # read the command from the command line
1733 try:
1734 user_input = input("britney> ").split()
1735 except EOFError:
1736 print("")
1737 break
1738 except KeyboardInterrupt:
1739 print("")
1740 continue
1741 match user_input:
1742 case ("quit" | "exit", *_):
1743 # quit the hint tester
1744 break
1745 case ("python-console", *_):
1746 try:
1747 import britney2.console
1748 except ImportError as e:
1749 print("Failed to import britney.console module: %s" % repr(e))
1750 continue
1751 britney2.console.run_python_console(self)
1752 print("Returning to the britney hint-tester console")
1753 # run a hint
1754 case ("easy" | "hint" | "force-hint" as choice, *items):
1755 mi_factory = self._migration_item_factory
1756 try:
1757 self.do_hint(
1758 choice,
1759 "hint-tester",
1760 mi_factory.parse_items(items),
1761 )
1762 self.printuninstchange()
1763 except KeyboardInterrupt:
1764 continue
1765 case (str() as hint, *_) if hint in known_hints:
1766 self._hint_parser.parse_hints(
1767 "hint-tester", self.HINTS_ALL, "<stdin>", [" ".join(user_input)]
1768 )
1769 self.write_excuses()
1771 try:
1772 readline.write_history_file(histfile)
1773 except OSError as e:
1774 self.logger.warning("Could not write %s: %s", histfile, e)
1776 def do_hint(self, hinttype: str, who: str, pkgvers: list[MigrationItem]) -> bool:
1777 """Process hints
1779 This method process `easy`, `hint` and `force-hint` hints. If the
1780 requested version is not in the relevant source suite, then the hint
1781 is skipped.
1782 """
1784 output_logger = self.output_logger
1786 self.logger.info("> Processing '%s' hint from %s", hinttype, who)
1787 output_logger.info(
1788 "Trying %s from %s: %s",
1789 hinttype,
1790 who,
1791 " ".join(f"{x.uvname}/{x.version}" for x in pkgvers),
1792 )
1794 issues = []
1795 # loop on the requested packages and versions
1796 for idx in range(len(pkgvers)):
1797 pkg = pkgvers[idx]
1798 # skip removal requests
1799 if pkg.is_removal:
1800 continue
1802 suite = pkg.suite
1804 assert pkg.version is not None
1805 if pkg.package not in suite.sources: 1805 ↛ 1806line 1805 didn't jump to line 1806 because the condition on line 1805 was never true
1806 issues.append(f"Source {pkg.package} has no version in {suite.name}")
1807 elif ( 1807 ↛ 1811line 1807 didn't jump to line 1811
1808 apt_pkg.version_compare(suite.sources[pkg.package].version, pkg.version)
1809 != 0
1810 ):
1811 issues.append(
1812 "Version mismatch, %s %s != %s"
1813 % (pkg.package, pkg.version, suite.sources[pkg.package].version)
1814 )
1815 if issues: 1815 ↛ 1816line 1815 didn't jump to line 1816 because the condition on line 1815 was never true
1816 output_logger.warning("%s: Not using hint", ", ".join(issues))
1817 return False
1819 self.do_all(hinttype, pkgvers)
1820 return True
1822 def get_auto_hinter_hints(
1823 self, upgrade_me: list[MigrationItem]
1824 ) -> list[list[frozenset[MigrationItem]]]:
1825 """Auto-generate "easy" hints.
1827 This method attempts to generate "easy" hints for sets of packages which
1828 must migrate together. Beginning with a package which does not depend on
1829 any other package (in terms of excuses), a list of dependencies and
1830 reverse dependencies is recursively created.
1832 Once all such lists have been generated, any which are subsets of other
1833 lists are ignored in favour of the larger lists. The remaining lists are
1834 then attempted in turn as "easy" hints.
1836 We also try to auto hint circular dependencies analyzing the update
1837 excuses relationships. If they build a circular dependency, which we already
1838 know as not-working with the standard do_all algorithm, try to `easy` them.
1839 """
1840 self.logger.info("> Processing hints from the auto hinter")
1842 sources_t = self.suite_info.target_suite.sources
1843 excuses = self.excuses
1845 def excuse_still_valid(excuse: "Excuse") -> bool:
1846 source = excuse.source
1847 assert isinstance(excuse.item, MigrationItem)
1848 arch = excuse.item.architecture
1849 # TODO for binNMUs, this check is always ok, even if the item
1850 # migrated already
1851 valid = (
1852 arch != "source"
1853 or source not in sources_t
1854 or sources_t[source].version != excuse.ver[1]
1855 )
1856 # TODO migrated items should be removed from upgrade_me, so this
1857 # should not happen
1858 if not valid: 1858 ↛ 1859line 1858 didn't jump to line 1859 because the condition on line 1858 was never true
1859 raise AssertionError("excuse no longer valid %s" % (excuse.item))
1860 return valid
1862 # consider only excuses which are valid candidates and still relevant.
1863 valid_excuses = frozenset(
1864 e.name
1865 for n, e in excuses.items()
1866 if e.item in upgrade_me and excuse_still_valid(e)
1867 )
1868 excuses_deps = {
1869 name: valid_excuses.intersection(excuse.get_deps())
1870 for name, excuse in excuses.items()
1871 if name in valid_excuses
1872 }
1873 excuses_rdeps = defaultdict(set)
1874 for name, deps in excuses_deps.items():
1875 for dep in deps:
1876 excuses_rdeps[dep].add(name)
1878 # loop on them
1879 candidates = []
1880 mincands = []
1881 seen_hints = set()
1882 for e in valid_excuses:
1883 excuse = excuses[e]
1884 if not excuse.get_deps():
1885 assert isinstance(excuse.item, MigrationItem)
1886 items = [excuse.item]
1887 orig_size = 1
1888 looped = False
1889 seen_items = set()
1890 seen_items.update(items)
1892 for item in items:
1893 assert isinstance(item, MigrationItem)
1894 # excuses which depend on "item" or are depended on by it
1895 new_items = cast(
1896 set[MigrationItem],
1897 {
1898 excuses[x].item
1899 for x in chain(
1900 excuses_deps[item.name], excuses_rdeps[item.name]
1901 )
1902 },
1903 )
1904 new_items -= seen_items
1905 items.extend(new_items)
1906 seen_items.update(new_items)
1908 if not looped and len(items) > 1:
1909 orig_size = len(items)
1910 h = frozenset(seen_items)
1911 if h not in seen_hints: 1911 ↛ 1914line 1911 didn't jump to line 1914 because the condition on line 1911 was always true
1912 mincands.append(h)
1913 seen_hints.add(h)
1914 looped = True
1915 if len(items) != orig_size: 1915 ↛ 1916line 1915 didn't jump to line 1916 because the condition on line 1915 was never true
1916 h = frozenset(seen_items)
1917 if h != mincands[-1] and h not in seen_hints:
1918 candidates.append(h)
1919 seen_hints.add(h)
1920 return [candidates, mincands]
1922 def run_auto_hinter(self) -> None:
1923 for lst in self.get_auto_hinter_hints(self.upgrade_me):
1924 for hint in lst:
1925 self.do_hint("easy", "autohinter", sorted(hint))
1927 def nuninst_arch_report(self, nuninst: dict[str, set[str]], arch: str) -> None:
1928 """Print a report of uninstallable packages for one architecture."""
1929 all = defaultdict(set)
1930 binaries_t = self.suite_info.target_suite.binaries
1931 for p in nuninst[arch]:
1932 pkg = binaries_t[arch][p]
1933 all[(pkg.source, pkg.source_version)].add(p)
1935 print("* %s" % arch)
1937 for (src, ver), pkgs in sorted(all.items()):
1938 print(" {} ({}): {}".format(src, ver, " ".join(sorted(pkgs))))
1940 print()
1942 def _remove_archall_faux_packages(self) -> None:
1943 """Remove faux packages added for the excuses phase
1945 To prevent binary packages from going missing while they are listed by
1946 their source package we add bin:faux packages during reading in the
1947 Sources. They are used during the excuses phase to prevent packages
1948 from becoming candidates. However, they interfere in complex ways
1949 during the installability phase, so instead of having all code during
1950 migration be aware of this excuses phase implementation detail, let's
1951 remove them again.
1953 """
1954 if not self.options.archall_inconsistency_allowed:
1955 all_binaries = self.all_binaries
1956 faux_a = {x for x in all_binaries.keys() if x.architecture == "faux"}
1957 for pkg_a in faux_a:
1958 del all_binaries[pkg_a]
1960 for suite in self.suite_info._suites.values():
1961 for arch in suite.binaries.keys():
1962 binaries = suite.binaries[arch]
1963 faux_b = {
1964 x for x in binaries if binaries[x].pkg_id.architecture == "faux"
1965 }
1966 for pkg_b in faux_b:
1967 del binaries[pkg_b]
1968 sources = suite.sources
1969 for src in sources.keys():
1970 faux_s = {
1971 x for x in sources[src].binaries if x.architecture == "faux"
1972 }
1973 sources[src].binaries -= faux_s
1975 def main(self) -> None:
1976 """Main method
1978 This is the entry point for the class: it includes the list of calls
1979 for the member methods which will produce the output files.
1980 """
1981 # if running in --print-uninst mode, quit
1982 if self.options.print_uninst: 1982 ↛ 1983line 1982 didn't jump to line 1983 because the condition on line 1982 was never true
1983 return
1984 # if no actions are provided, build the excuses and sort them
1985 elif not self.options.actions: 1985 ↛ 1989line 1985 didn't jump to line 1989 because the condition on line 1985 was always true
1986 self.write_excuses()
1987 # otherwise, use the actions provided by the command line
1988 else:
1989 self.upgrade_me = self.options.actions.split()
1991 self._remove_archall_faux_packages()
1993 if self.options.compute_migrations or self.options.hint_tester:
1994 if self.options.dry_run: 1994 ↛ 1995line 1994 didn't jump to line 1995 because the condition on line 1994 was never true
1995 self.logger.info(
1996 "Upgrade output not (also) written to a separate file"
1997 " as this is a dry-run."
1998 )
1999 elif hasattr(self.options, "upgrade_output"): 1999 ↛ 2009line 1999 didn't jump to line 2009 because the condition on line 1999 was always true
2000 upgrade_output = getattr(self.options, "upgrade_output")
2001 file_handler = logging.FileHandler(
2002 upgrade_output, mode="w", encoding="utf-8"
2003 )
2004 output_formatter = logging.Formatter("%(message)s")
2005 file_handler.setFormatter(output_formatter)
2006 self.output_logger.addHandler(file_handler)
2007 self.logger.info("Logging upgrade output to %s", upgrade_output)
2008 else:
2009 self.logger.info(
2010 "Upgrade output not (also) written to a separate file"
2011 " as the UPGRADE_OUTPUT configuration is not provided."
2012 )
2014 # run the hint tester
2015 if self.options.hint_tester: 2015 ↛ 2016line 2015 didn't jump to line 2016 because the condition on line 2015 was never true
2016 self.hint_tester()
2017 # run the upgrade test
2018 else:
2019 self.upgrade_testing()
2021 self.logger.info("> Stats from the installability tester")
2022 for stat in self._inst_tester.stats.stats():
2023 self.logger.info("> %s", stat)
2024 else:
2025 self.logger.info("Migration computation skipped as requested.")
2026 if not self.options.dry_run: 2026 ↛ 2028line 2026 didn't jump to line 2028 because the condition on line 2026 was always true
2027 self._policy_engine.save_state(self)
2028 logging.shutdown()
2031if __name__ == "__main__": 2031 ↛ 2032line 2031 didn't jump to line 2032 because the condition on line 2031 was never true
2032 Britney().main()