Coverage for britney2/britney.py: 83%
773 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-06-17 09:00 +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-piuparts",
298 "ignore-rc-bugs",
299 "force-skiptest",
300 "force-badtest",
301 )
302 HINTS_STANDARD = ("urgent", "age-days") + HINTS_HELPERS
303 # ALL = {"force", "force-hint", "block-all"} | HINTS_STANDARD | registered policy hints (not covered above)
304 HINTS_ALL = "ALL"
305 pkg_universe: "BinaryPackageUniverse"
306 _inst_tester: "InstallabilityTester"
307 constraints: dict[str, list[str]]
308 suite_info: Suites
310 def __init__(self) -> None:
311 """Class constructor
313 This method initializes and populates the data lists, which contain all
314 the information needed by the other methods of the class.
315 """
317 # setup logging - provide the "short level name" (i.e. INFO -> I) that
318 # we used to use prior to using the logging module.
320 old_factory = logging.getLogRecordFactory()
321 short_level_mapping = {
322 "CRITICAL": "F",
323 "INFO": "I",
324 "WARNING": "W",
325 "ERROR": "E",
326 "DEBUG": "N",
327 }
329 def record_factory(
330 *args: Any, **kwargs: Any
331 ) -> logging.LogRecord: # pragma: no cover
332 record = old_factory(*args, **kwargs)
333 try:
334 record.shortlevelname = short_level_mapping[record.levelname]
335 except KeyError:
336 record.shortlevelname = record.levelname
337 return record
339 logging.setLogRecordFactory(record_factory)
340 logging.basicConfig(
341 format="{shortlevelname}: [{asctime}] - {message}",
342 style="{",
343 datefmt="%Y-%m-%dT%H:%M:%S%z",
344 stream=sys.stdout,
345 )
347 self.logger = logging.getLogger()
349 # Logger for "upgrade_output"; the file handler will be attached later when
350 # we are ready to open the file.
351 self.output_logger = logging.getLogger("britney2.output.upgrade_output")
352 self.output_logger.setLevel(logging.INFO)
354 # initialize the apt_pkg back-end
355 apt_pkg.init()
357 # parse the command line arguments
358 self._policy_engine = PolicyEngine()
359 self.__parse_arguments()
360 assert self.suite_info is not None # for type checking
362 self.all_selected: list[MigrationItem] = []
363 self.excuses: dict[str, "Excuse"] = {}
364 self.upgrade_me: list[MigrationItem] = []
366 if self.options.nuninst_cache: 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 self.logger.info(
368 "Not building the list of non-installable packages, as requested"
369 )
370 if self.options.print_uninst:
371 nuninst = read_nuninst(
372 self.options.noninst_status, self.options.architectures
373 )
374 print("* summary")
375 print(
376 "\n".join(
377 "%4d %s" % (len(nuninst[x]), x)
378 for x in self.options.architectures
379 )
380 )
381 return
383 try:
384 constraints_file = os.path.join(
385 self.options.static_input_dir, "constraints"
386 )
387 faux_packages = os.path.join(self.options.static_input_dir, "faux-packages")
388 except AttributeError:
389 self.logger.info("The static_input_dir option is not set")
390 constraints_file = None
391 faux_packages = None
392 if faux_packages is not None and os.path.exists(faux_packages):
393 self.logger.info("Loading faux packages from %s", faux_packages)
394 self._load_faux_packages(faux_packages)
395 elif faux_packages is not None: 395 ↛ 398line 395 didn't jump to line 398 because the condition on line 395 was always true
396 self.logger.info("No Faux packages as %s does not exist", faux_packages)
398 if constraints_file is not None and os.path.exists(constraints_file):
399 self.logger.info("Loading constraints from %s", constraints_file)
400 self.constraints = self._load_constraints(constraints_file)
401 else:
402 if constraints_file is not None: 402 ↛ 406line 402 didn't jump to line 406
403 self.logger.info(
404 "No constraints as %s does not exist", constraints_file
405 )
406 self.constraints = {
407 "keep-installable": [],
408 }
410 self.logger.info("Compiling Installability tester")
411 self.pkg_universe, self._inst_tester = build_installability_tester(
412 self.suite_info, self.options.architectures
413 )
414 target_suite = self.suite_info.target_suite
415 target_suite.inst_tester = self._inst_tester
417 self.allow_uninst: dict[str, set[str | None]] = {}
418 for arch in self.options.architectures:
419 self.allow_uninst[arch] = set()
420 self._migration_item_factory: MigrationItemFactory = MigrationItemFactory(
421 self.suite_info
422 )
423 self._hint_parser: HintParser = HintParser(self._migration_item_factory)
424 self._migration_manager: MigrationManager = MigrationManager(
425 self.options,
426 self.suite_info,
427 self.all_binaries,
428 self.pkg_universe,
429 self.constraints,
430 self.allow_uninst,
431 self._migration_item_factory,
432 self.hints,
433 )
435 if not self.options.nuninst_cache: 435 ↛ 475line 435 didn't jump to line 475 because the condition on line 435 was always true
436 self.logger.info(
437 "Building the list of non-installable packages for the full archive"
438 )
439 self._inst_tester.compute_installability()
440 nuninst = compile_nuninst(
441 target_suite, self.options.architectures, self.options.nobreakall_arches
442 )
443 self.nuninst_orig: dict[str, set[str]] = nuninst
444 for arch in self.options.architectures:
445 self.logger.info(
446 "> Found %d non-installable packages for %s",
447 len(nuninst[arch]),
448 arch,
449 )
450 if self.options.print_uninst: 450 ↛ 451line 450 didn't jump to line 451 because the condition on line 450 was never true
451 self.nuninst_arch_report(nuninst, arch)
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 print("* summary")
455 print(
456 "\n".join(
457 map(
458 lambda x: "%4d %s" % (len(nuninst[x]), x),
459 self.options.architectures,
460 )
461 )
462 )
463 return
464 else:
465 write_nuninst(self.options.noninst_status, nuninst)
467 stats = self._inst_tester.compute_stats()
468 self.logger.info("> Installability tester statistics (per architecture)")
469 for arch in self.options.architectures:
470 arch_stat = stats[arch]
471 self.logger.info("> %s", arch)
472 for stat in arch_stat.stat_summary():
473 self.logger.info("> - %s", stat)
474 else:
475 self.logger.info("Loading uninstallability counters from cache")
476 self.nuninst_orig = read_nuninst(
477 self.options.noninst_status, self.options.architectures
478 )
480 # nuninst_orig may get updated during the upgrade process
481 self.nuninst_orig_save: dict[str, set[str]] = clone_nuninst(
482 self.nuninst_orig, architectures=self.options.architectures
483 )
485 self._policy_engine.register_policy_hints(self._hint_parser)
487 try:
488 self.read_hints(self.options.hintsdir)
489 except AttributeError:
490 self.read_hints(os.path.join(self.suite_info["unstable"].path, "Hints"))
492 self._policy_engine.initialise(self, self.hints)
494 def __parse_arguments(self) -> None:
495 """Parse the command line arguments
497 This method parses and initializes the command line arguments.
498 While doing so, it preprocesses some of the options to be converted
499 in a suitable form for the other methods of the class.
500 """
501 # initialize the parser
502 parser = optparse.OptionParser(version="%prog")
503 parser.add_option(
504 "-v", "", action="count", dest="verbose", help="enable verbose output"
505 )
506 parser.add_option(
507 "-c",
508 "--config",
509 action="store",
510 dest="config",
511 default="/etc/britney.conf",
512 help="path for the configuration file",
513 )
514 parser.add_option(
515 "",
516 "--architectures",
517 action="store",
518 dest="architectures",
519 default=None,
520 help="override architectures from configuration file",
521 )
522 parser.add_option(
523 "",
524 "--actions",
525 action="store",
526 dest="actions",
527 default=None,
528 help="override the list of actions to be performed",
529 )
530 parser.add_option(
531 "",
532 "--hints",
533 action="store",
534 dest="hints",
535 default=None,
536 help="additional hints, separated by semicolons",
537 )
538 parser.add_option(
539 "",
540 "--hint-tester",
541 action="store_true",
542 dest="hint_tester",
543 default=None,
544 help="provide a command line interface to test hints",
545 )
546 parser.add_option(
547 "",
548 "--dry-run",
549 action="store_true",
550 dest="dry_run",
551 default=False,
552 help="disable all outputs to the testing directory",
553 )
554 parser.add_option(
555 "",
556 "--nuninst-cache",
557 action="store_true",
558 dest="nuninst_cache",
559 default=False,
560 help="do not build the non-installability status, use the cache from file",
561 )
562 parser.add_option(
563 "",
564 "--print-uninst",
565 action="store_true",
566 dest="print_uninst",
567 default=False,
568 help="just print a summary of uninstallable packages",
569 )
570 parser.add_option(
571 "",
572 "--compute-migrations",
573 action="store_true",
574 dest="compute_migrations",
575 default=True,
576 help="Compute which packages can migrate (the default)",
577 )
578 parser.add_option(
579 "",
580 "--no-compute-migrations",
581 action="store_false",
582 dest="compute_migrations",
583 help="Do not compute which packages can migrate.",
584 )
585 parser.add_option(
586 "",
587 "--series",
588 action="store",
589 dest="series",
590 default="",
591 help="set distribution series name",
592 )
593 parser.add_option(
594 "",
595 "--distribution",
596 action="store",
597 dest="distribution",
598 default="debian",
599 help="set distribution name",
600 )
601 (self.options, self.args) = parser.parse_args()
603 if self.options.verbose: 603 ↛ 609line 603 didn't jump to line 609 because the condition on line 603 was always true
604 if self.options.verbose > 1: 604 ↛ 605line 604 didn't jump to line 605 because the condition on line 604 was never true
605 self.logger.setLevel(logging.DEBUG)
606 else:
607 self.logger.setLevel(logging.INFO)
608 else:
609 self.logger.setLevel(logging.WARNING)
610 # Historical way to get debug information (equivalent to -vv)
611 try: # pragma: no cover
612 if int(os.environ.get("BRITNEY_DEBUG", "0")):
613 self.logger.setLevel(logging.DEBUG)
614 except ValueError: # pragma: no cover
615 pass
617 # integrity checks
618 if self.options.nuninst_cache and self.options.print_uninst: # pragma: no cover
619 self.logger.error("nuninst_cache and print_uninst are mutually exclusive!")
620 sys.exit(1)
622 # if the configuration file exists, then read it and set the additional options
623 if not os.path.isfile(self.options.config): # pragma: no cover
624 self.logger.error(
625 "Unable to read the configuration file (%s), exiting!",
626 self.options.config,
627 )
628 sys.exit(1)
630 self.HINTS: dict[str, Any] = {"command-line": self.HINTS_ALL}
631 with open(self.options.config, encoding="utf-8") as config:
632 for line in config:
633 if "=" in line and not line.strip().startswith("#"):
634 k, v = line.split("=", 1)
635 k = k.strip()
636 v = v.strip()
637 if k.startswith("HINTS_"):
638 self.HINTS[k.split("_")[1].lower()] = reduce( 638 ↛ exitline 638 didn't jump to the function exit
639 lambda x, y: x + y,
640 [
641 hasattr(self, "HINTS_" + i)
642 and getattr(self, "HINTS_" + i)
643 or (i,)
644 for i in v.split()
645 ],
646 )
647 elif not hasattr(self.options, k.lower()) or not getattr(
648 self.options, k.lower()
649 ):
650 setattr(self.options, k.lower(), v)
652 parse_option(self.options, "archall_inconsistency_allowed", to_bool=True)
653 parse_option(
654 self.options, "be_strict_with_build_deps", default=True, to_bool=True
655 )
657 suite_loader = DebMirrorLikeSuiteContentLoader(self.options)
659 try:
660 self.suite_info = suite_loader.load_suites()
661 except MissingRequiredConfigurationError as e: # pragma: no cover
662 self.logger.error(
663 "Could not load the suite content due to missing configuration: %s",
664 str(e),
665 )
666 sys.exit(1)
667 self.all_binaries = suite_loader.all_binaries()
668 self.options.components = suite_loader.components
669 self.options.architectures = suite_loader.architectures
670 self.options.nobreakall_arches = suite_loader.nobreakall_arches
671 self.options.outofsync_arches = suite_loader.outofsync_arches
672 self.options.break_arches = suite_loader.break_arches
673 self.options.new_arches = suite_loader.new_arches
674 if self.options.series == "": 674 ↛ 677line 674 didn't jump to line 677 because the condition on line 674 was always true
675 self.options.series = self.suite_info.target_suite.name
677 if self.options.heidi_output and not hasattr( 677 ↛ 682line 677 didn't jump to line 682 because the condition on line 677 was always true
678 self.options, "heidi_delta_output"
679 ):
680 self.options.heidi_delta_output = self.options.heidi_output + "Delta"
682 self.options.smooth_updates = self.options.smooth_updates.split()
684 parse_option(self.options, "ignore_cruft", to_bool=True)
685 parse_option(self.options, "check_consistency_level", default=2, to_int=True)
686 parse_option(self.options, "build_url")
688 self._policy_engine.load_policies(
689 self.options, self.suite_info, MIGRATION_POLICIES
690 )
692 @property
693 def hints(self) -> HintCollection:
694 return self._hint_parser.hints
696 def _load_faux_packages(self, faux_packages_file: str) -> None:
697 """Loads fake packages
699 In rare cases, it is useful to create a "fake" package that can be used to satisfy
700 dependencies. This is usually needed for packages that are not shipped directly
701 on this mirror but is a prerequisite for using this mirror (e.g. some vendors provide
702 non-distributable "setup" packages and contrib/non-free packages depend on these).
704 :param faux_packages_file: Path to the file containing the fake package definitions
705 """
706 tag_file = apt_pkg.TagFile(faux_packages_file)
707 get_field = tag_file.section.get
708 step = tag_file.step
709 no = 0
710 pri_source_suite = self.suite_info.primary_source_suite
711 target_suite = self.suite_info.target_suite
713 while step():
714 no += 1
715 pkg_name = get_field("Package", None)
716 if pkg_name is None: # pragma: no cover
717 raise ValueError(
718 "Missing Package field in paragraph %d (file %s)"
719 % (no, faux_packages_file)
720 )
721 pkg_name = sys.intern(pkg_name)
722 version = sys.intern(get_field("Version", "1.0-1"))
723 provides_raw = get_field("Provides")
724 archs_raw = get_field("Architecture", None)
725 component = get_field("Component", "non-free")
726 if archs_raw: 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true
727 archs = archs_raw.split()
728 else:
729 archs = self.options.architectures
730 faux_section = "faux"
731 if component != "main": 731 ↛ 733line 731 didn't jump to line 733 because the condition on line 731 was always true
732 faux_section = "%s/faux" % component
733 src_data = SourcePackage(
734 pkg_name,
735 version,
736 sys.intern(faux_section),
737 set(),
738 None,
739 True,
740 None,
741 None,
742 [],
743 [],
744 )
746 target_suite.sources[pkg_name] = src_data
747 pri_source_suite.sources[pkg_name] = src_data
749 for arch in archs:
750 pkg_id = BinaryPackageId(pkg_name, version, arch)
751 if provides_raw: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true
752 provides = parse_provides(
753 provides_raw, pkg_id=pkg_id, logger=self.logger
754 )
755 else:
756 provides = None
757 bin_data = BinaryPackage(
758 faux_section,
759 pkg_name,
760 version,
761 arch,
762 MultiArch.from_str(get_field("Multi-Arch")),
763 None,
764 None,
765 provides,
766 False,
767 pkg_id,
768 None,
769 )
771 src_data.binaries.add(pkg_id)
772 target_suite.binaries[arch][pkg_name] = bin_data
773 pri_source_suite.binaries[arch][pkg_name] = bin_data
775 # register provided packages with the target suite provides table
776 for provided_pkg, provided_version, _ in ( 776 ↛ 779line 776 didn't jump to line 779
777 bin_data.provides if bin_data.provides is not None else []
778 ):
779 target_suite.provides_table[arch][provided_pkg].add(
780 (pkg_name, provided_version)
781 )
783 self.all_binaries[pkg_id] = bin_data
785 def _load_constraints(self, constraints_file: str) -> dict[str, list[str]]:
786 """Loads configurable constraints
788 The constraints file can contain extra rules that Britney should attempt
789 to satisfy. Examples can be "keep package X in testing and ensure it is
790 installable".
792 :param constraints_file: Path to the file containing the constraints
793 """
794 tag_file = apt_pkg.TagFile(constraints_file)
795 get_field = tag_file.section.get
796 step = tag_file.step
797 no = 0
798 faux_version = sys.intern("1")
799 faux_section = sys.intern("faux")
800 keep_installable: list[str] = []
801 constraints = {"keep-installable": keep_installable}
802 pri_source_suite = self.suite_info.primary_source_suite
803 target_suite = self.suite_info.target_suite
805 while step():
806 no += 1
807 pkg_name = get_field("Fake-Package-Name", None)
808 if pkg_name is None: # pragma: no cover
809 raise ValueError(
810 "Missing Fake-Package-Name field in paragraph %d (file %s)"
811 % (no, constraints_file)
812 )
813 pkg_name = sys.intern(pkg_name)
815 def mandatory_field(x: str) -> str:
816 v: str = get_field(x, None)
817 if v is None: # pragma: no cover
818 raise ValueError(
819 "Missing %s field for %s (file %s)"
820 % (x, pkg_name, constraints_file)
821 )
822 return v
824 constraint = mandatory_field("Constraint")
825 if constraint not in {"present-and-installable"}: # pragma: no cover
826 raise ValueError(
827 "Unsupported constraint %s for %s (file %s)"
828 % (constraint, pkg_name, constraints_file)
829 )
831 self.logger.info(" - constraint %s", pkg_name)
833 pkg_list = [
834 x.strip()
835 for x in mandatory_field("Package-List").split("\n")
836 if x.strip() != "" and not x.strip().startswith("#")
837 ]
838 src_data = SourcePackage(
839 pkg_name,
840 faux_version,
841 faux_section,
842 set(),
843 None,
844 True,
845 None,
846 None,
847 [],
848 [],
849 )
850 target_suite.sources[pkg_name] = src_data
851 pri_source_suite.sources[pkg_name] = src_data
852 keep_installable.append(pkg_name)
853 for arch in self.options.architectures:
854 deps = []
855 for pkg_spec in pkg_list:
856 s = pkg_spec.split(None, 1)
857 if len(s) == 1:
858 deps.append(s[0])
859 else:
860 pkg, arch_res = s
861 if not (
862 arch_res.startswith("[") and arch_res.endswith("]")
863 ): # pragma: no cover
864 raise ValueError(
865 "Invalid arch-restriction on %s - should be [arch1 arch2] (for %s file %s)"
866 % (pkg, pkg_name, constraints_file)
867 )
868 arch_res_l = arch_res[1:-1].split()
869 if not arch_res_l: # pragma: no cover
870 msg = "Empty arch-restriction for %s: Uses comma or negation (for %s file %s)"
871 raise ValueError(msg % (pkg, pkg_name, constraints_file))
872 for a in arch_res_l:
873 if a == arch:
874 deps.append(pkg)
875 elif "," in a or "!" in a: # pragma: no cover
876 msg = "Invalid arch-restriction for %s: Uses comma or negation (for %s file %s)"
877 raise ValueError(
878 msg % (pkg, pkg_name, constraints_file)
879 )
880 pkg_id = BinaryPackageId(pkg_name, faux_version, arch)
881 bin_data = BinaryPackage(
882 faux_section,
883 pkg_name,
884 faux_version,
885 arch,
886 MultiArch.NO,
887 ", ".join(deps),
888 None,
889 [],
890 False,
891 pkg_id,
892 [],
893 )
894 src_data.binaries.add(pkg_id)
895 target_suite.binaries[arch][pkg_name] = bin_data
896 pri_source_suite.binaries[arch][pkg_name] = bin_data
897 self.all_binaries[pkg_id] = bin_data
899 return constraints
901 # Data reading/writing methods
902 # ----------------------------
904 def read_hints(self, hintsdir: str) -> None:
905 """Read the hint commands from the specified directory
907 The hint commands are read from the files contained in the directory
908 specified by the `hintsdir' parameter.
909 The names of the files have to be the same as the authorized users
910 for the hints.
912 The file contains rows with the format:
914 <command> <package-name>[/<version>]
916 The method returns a dictionary where the key is the command, and
917 the value is the list of affected packages.
918 """
920 for who in self.HINTS.keys():
921 if who == "command-line":
922 lines = self.options.hints and self.options.hints.split(";") or ()
923 filename = "<cmd-line>"
924 self._hint_parser.parse_hints(who, self.HINTS[who], filename, lines)
925 else:
926 filename = os.path.join(hintsdir, who)
927 if not os.path.isfile(filename): 927 ↛ 928line 927 didn't jump to line 928 because the condition on line 927 was never true
928 self.logger.error(
929 "Cannot read hints list from %s, no such file!", filename
930 )
931 continue
932 self.logger.info("Loading hints list from %s", filename)
933 with open(filename, encoding="utf-8") as f:
934 self._hint_parser.parse_hints(who, self.HINTS[who], filename, f)
936 hints = self._hint_parser.hints
938 for x in (
939 "block",
940 "block-all",
941 "block-udeb",
942 "unblock",
943 "unblock-udeb",
944 "force",
945 "urgent",
946 "remove",
947 "age-days",
948 ):
949 z: dict[str | None, dict[str | None, tuple[Hint, str]]] = defaultdict(dict)
950 for hint in hints[x]:
951 package = hint.package
952 architecture = hint.architecture
953 key = (hint, hint.user)
954 if (
955 package in z
956 and architecture in z[package]
957 and z[package][architecture] != key
958 ):
959 hint2 = z[package][architecture][0]
960 if x in ("unblock", "unblock-udeb", "age-days"): 960 ↛ 992line 960 didn't jump to line 992 because the condition on line 960 was always true
961 assert hint.version is not None
962 assert hint2.version is not None
963 if apt_pkg.version_compare(hint2.version, hint.version) < 0:
964 # This hint is for a newer version, so discard the old one
965 self.logger.warning(
966 "Overriding %s[%s] = ('%s', '%s', '%s') with ('%s', '%s', '%s')",
967 x,
968 package,
969 hint2.version,
970 hint2.architecture,
971 hint2.user,
972 hint.version,
973 hint.architecture,
974 hint.user,
975 )
976 hint2.set_active(False)
977 else:
978 # This hint is for an older version, so ignore it in favour of the new one
979 self.logger.warning(
980 "Ignoring %s[%s] = ('%s', '%s', '%s'), ('%s', '%s', '%s') is higher or equal",
981 x,
982 package,
983 hint.version,
984 hint.architecture,
985 hint.user,
986 hint2.version,
987 hint2.architecture,
988 hint2.user,
989 )
990 hint.set_active(False)
991 else:
992 self.logger.warning(
993 "Overriding %s[%s] = ('%s', '%s') with ('%s', '%s')",
994 x,
995 package,
996 hint2.user,
997 hint2,
998 hint.user,
999 hint,
1000 )
1001 hint2.set_active(False)
1003 z[package][architecture] = key
1005 for hint in hints["allow-uninst"]:
1006 if hint.architecture == "source":
1007 for arch in self.options.architectures:
1008 self.allow_uninst[arch].add(hint.package)
1009 else:
1010 assert hint.architecture is not None
1011 self.allow_uninst[hint.architecture].add(hint.package)
1013 # Sanity check the hints hash
1014 if len(hints["block"]) == 0 and len(hints["block-udeb"]) == 0: 1014 ↛ 1015line 1014 didn't jump to line 1015 because the condition on line 1014 was never true
1015 self.logger.warning("WARNING: No block hints at all, not even udeb ones!")
1017 # Remove all hints that were set inactive.
1018 # We don't need to keep unused hints in memory.
1019 hints.remove_inactive_hints()
1021 def write_excuses(self) -> None:
1022 """Produce and write the update excuses
1024 This method handles the update excuses generation: the packages are
1025 looked at to determine whether they are valid candidates. For the details
1026 of this procedure, please refer to the module docstring.
1027 """
1029 self.logger.info("Update Excuses generation started")
1031 mi_factory = self._migration_item_factory
1032 excusefinder = ExcuseFinder(
1033 self.options,
1034 self.suite_info,
1035 self.all_binaries,
1036 self.pkg_universe,
1037 self._policy_engine,
1038 mi_factory,
1039 self.hints,
1040 )
1042 excuses, upgrade_me = excusefinder.find_actionable_excuses()
1043 self.excuses = excuses
1045 # sort the list of candidates
1046 self.upgrade_me = sorted(upgrade_me)
1047 old_lib_removals = old_libraries(
1048 mi_factory, self.suite_info, self.options.outofsync_arches
1049 )
1050 self.upgrade_me.extend(old_lib_removals)
1051 self.output_logger.info(
1052 "List of old libraries added to upgrade_me (%d):", len(old_lib_removals)
1053 )
1054 log_and_format_old_libraries(self.output_logger, old_lib_removals)
1056 # write excuses to the output file
1057 if not self.options.dry_run: 1057 ↛ 1074line 1057 didn't jump to line 1074 because the condition on line 1057 was always true
1058 self.logger.info("> Writing Excuses to %s", self.options.excuses_output)
1059 write_excuses(
1060 excuses,
1061 self.options.excuses_output,
1062 output_format=ExcusesOutputFormat.LEGACY_HTML,
1063 )
1064 if hasattr(self.options, "excuses_yaml_output"): 1064 ↛ 1074line 1064 didn't jump to line 1074 because the condition on line 1064 was always true
1065 self.logger.info(
1066 "> Writing YAML Excuses to %s", self.options.excuses_yaml_output
1067 )
1068 write_excuses(
1069 excuses,
1070 self.options.excuses_yaml_output,
1071 output_format=ExcusesOutputFormat.YAML,
1072 )
1074 self.logger.info("Update Excuses generation completed")
1076 # Upgrade run
1077 # -----------
1079 def eval_nuninst(
1080 self,
1081 nuninst: dict[str, set[str]],
1082 original: dict[str, set[str]] | None = None,
1083 ) -> str:
1084 """Return a string which represents the uninstallability counters
1086 This method returns a string which represents the uninstallability
1087 counters reading the uninstallability statistics `nuninst` and, if
1088 present, merging the results with the `original` one.
1090 An example of the output string is:
1091 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
1093 where the first part is the number of broken packages in non-break
1094 architectures + the total number of broken packages for all the
1095 architectures.
1096 """
1097 res = []
1098 total = 0
1099 totalbreak = 0
1100 for arch in self.options.architectures:
1101 if arch in nuninst: 1101 ↛ 1103line 1101 didn't jump to line 1103 because the condition on line 1101 was always true
1102 n = len(nuninst[arch])
1103 elif original and arch in original:
1104 n = len(original[arch])
1105 else:
1106 continue
1107 if arch in self.options.break_arches:
1108 totalbreak = totalbreak + n
1109 else:
1110 total = total + n
1111 res.append("%s-%d" % (arch[0], n))
1112 return "%d+%d: %s" % (total, totalbreak, ":".join(res))
1114 def iter_packages(
1115 self,
1116 packages: list[MigrationItem],
1117 selected: list[MigrationItem],
1118 nuninst: dict[str, set[str]] | None = None,
1119 ) -> tuple[dict[str, set[str]] | None, list[MigrationItem]]:
1120 """Iter on the list of actions and apply them one-by-one
1122 This method applies the changes from `packages` to testing, checking the uninstallability
1123 counters for every action performed. If the action does not improve them, it is reverted.
1124 The method returns the new uninstallability counters and the remaining actions if the
1125 final result is successful, otherwise (None, []).
1127 :param selected: list of MigrationItem?
1128 :param nuninst: dict with sets ? of ? per architecture
1129 """
1130 assert self.suite_info is not None # for type checking
1131 group_info = {}
1132 rescheduled_packages = packages
1133 maybe_rescheduled_packages: list[MigrationItem] = []
1134 output_logger = self.output_logger
1135 solver = InstallabilitySolver(self.pkg_universe, self._inst_tester)
1136 mm = self._migration_manager
1137 target_suite = self.suite_info.target_suite
1139 for y in sorted((y for y in packages), key=attrgetter("uvname")):
1140 try:
1141 _, updates, rms, _ = mm.compute_groups(y)
1142 result = (y, sorted(updates), sorted(rms))
1143 group_info[y] = result
1144 except MigrationConstraintException as e:
1145 rescheduled_packages.remove(y)
1146 output_logger.info("not adding package to list: %s", (y.package))
1147 output_logger.info(" got exception: %s" % (repr(e)))
1149 if nuninst:
1150 nuninst_orig = nuninst
1151 else:
1152 nuninst_orig = self.nuninst_orig
1154 nuninst_last_accepted = nuninst_orig
1156 output_logger.info(
1157 "recur: [] %s %d/0", ",".join(x.uvname for x in selected), len(packages)
1158 )
1159 while rescheduled_packages:
1160 groups = [group_info[x] for x in rescheduled_packages]
1161 worklist = solver.solve_groups(groups)
1162 rescheduled_packages = []
1164 worklist.reverse()
1166 while worklist:
1167 comp = worklist.pop()
1168 comp_name = " ".join(item.uvname for item in comp)
1169 output_logger.info("trying: %s" % comp_name)
1170 with mm.start_transaction() as transaction:
1171 accepted = False
1172 try:
1173 (
1174 accepted,
1175 nuninst_after,
1176 failed_arch,
1177 new_cruft,
1178 ) = mm.migrate_items_to_target_suite(
1179 comp, nuninst_last_accepted
1180 )
1181 if accepted:
1182 selected.extend(comp)
1183 transaction.commit()
1184 output_logger.info("accepted: %s", comp_name)
1185 output_logger.info(
1186 " ori: %s", self.eval_nuninst(nuninst_orig)
1187 )
1188 output_logger.info(
1189 " pre: %s", self.eval_nuninst(nuninst_last_accepted)
1190 )
1191 output_logger.info(
1192 " now: %s", self.eval_nuninst(nuninst_after)
1193 )
1194 if new_cruft:
1195 output_logger.info(
1196 " added new cruft items to list: %s",
1197 " ".join(x.uvname for x in sorted(new_cruft)),
1198 )
1200 if len(selected) <= 20:
1201 output_logger.info(
1202 " all: %s", " ".join(x.uvname for x in selected)
1203 )
1204 else:
1205 output_logger.info(
1206 " most: (%d) .. %s",
1207 len(selected),
1208 " ".join(x.uvname for x in selected[-20:]),
1209 )
1210 if self.options.check_consistency_level >= 3: 1210 ↛ 1211line 1210 didn't jump to line 1211 because the condition on line 1210 was never true
1211 target_suite.check_suite_source_pkg_consistency(
1212 "iter_packages after commit"
1213 )
1214 nuninst_last_accepted = nuninst_after
1215 for cruft_item in new_cruft:
1216 try:
1217 _, updates, rms, _ = mm.compute_groups(cruft_item)
1218 result = (cruft_item, sorted(updates), sorted(rms))
1219 group_info[cruft_item] = result
1220 worklist.append([cruft_item])
1221 except MigrationConstraintException as e:
1222 output_logger.info(
1223 " got exception adding cruft item %s to list: %s"
1224 % (cruft_item.uvname, repr(e))
1225 )
1226 rescheduled_packages.extend(maybe_rescheduled_packages)
1227 maybe_rescheduled_packages.clear()
1228 else:
1229 transaction.rollback()
1230 assert failed_arch # type checking
1231 broken = sorted(
1232 b
1233 for b in nuninst_after[failed_arch]
1234 if b not in nuninst_last_accepted[failed_arch]
1235 )
1236 compare_nuninst = None
1237 if any(
1238 item for item in comp if item.architecture != "source"
1239 ):
1240 compare_nuninst = nuninst_last_accepted
1241 # NB: try_migration already reverted this for us, so just print the results and move on
1242 output_logger.info(
1243 "skipped: %s (%d, %d, %d)",
1244 comp_name,
1245 len(rescheduled_packages),
1246 len(maybe_rescheduled_packages),
1247 len(worklist),
1248 )
1249 output_logger.info(
1250 " got: %s",
1251 self.eval_nuninst(nuninst_after, compare_nuninst),
1252 )
1253 output_logger.info(
1254 " * %s: %s", failed_arch, ", ".join(broken)
1255 )
1256 if self.options.check_consistency_level >= 3: 1256 ↛ 1257line 1256 didn't jump to line 1257 because the condition on line 1256 was never true
1257 target_suite.check_suite_source_pkg_consistency(
1258 "iter_package after rollback (not accepted)"
1259 )
1261 except MigrationConstraintException as e:
1262 transaction.rollback()
1263 output_logger.info(
1264 "skipped: %s (%d, %d, %d)",
1265 comp_name,
1266 len(rescheduled_packages),
1267 len(maybe_rescheduled_packages),
1268 len(worklist),
1269 )
1270 output_logger.info(" got exception: %s" % (repr(e)))
1271 if self.options.check_consistency_level >= 3: 1271 ↛ 1272line 1271 didn't jump to line 1272 because the condition on line 1271 was never true
1272 target_suite.check_suite_source_pkg_consistency(
1273 "iter_package after rollback (MigrationConstraintException)"
1274 )
1276 if not accepted:
1277 if len(comp) > 1:
1278 output_logger.info(
1279 " - splitting the component into single items and retrying them"
1280 )
1281 worklist.extend([item] for item in comp)
1282 else:
1283 maybe_rescheduled_packages.append(comp[0])
1285 output_logger.info(" finish: [%s]", ",".join(x.uvname for x in selected))
1286 output_logger.info("endloop: %s", self.eval_nuninst(self.nuninst_orig))
1287 output_logger.info(" now: %s", self.eval_nuninst(nuninst_last_accepted))
1288 format_and_log_uninst(
1289 output_logger,
1290 self.options.architectures,
1291 newly_uninst(self.nuninst_orig, nuninst_last_accepted),
1292 )
1293 output_logger.info("")
1295 return (nuninst_last_accepted, maybe_rescheduled_packages)
1297 def do_all(
1298 self,
1299 hinttype: str | None = None,
1300 init: list[MigrationItem] | None = None,
1301 actions: list[MigrationItem] | None = None,
1302 ) -> None:
1303 """Testing update runner
1305 This method tries to update testing checking the uninstallability
1306 counters before and after the actions to decide if the update was
1307 successful or not.
1308 """
1309 selected = []
1310 if actions:
1311 upgrade_me = actions[:]
1312 else:
1313 upgrade_me = self.upgrade_me[:]
1314 nuninst_start = self.nuninst_orig
1315 output_logger = self.output_logger
1316 target_suite = self.suite_info.target_suite
1318 # these are special parameters for hints processing
1319 force = False
1320 recurse = True
1321 nuninst_end = None
1322 extra: list[MigrationItem] = []
1323 mm = self._migration_manager
1325 if hinttype == "easy" or hinttype == "force-hint":
1326 force = hinttype == "force-hint"
1327 recurse = False
1329 # if we have a list of initial packages, check them
1330 if init:
1331 for x in init:
1332 if x not in upgrade_me:
1333 output_logger.warning(
1334 "failed: %s is not a valid candidate (or it already migrated)",
1335 x.uvname,
1336 )
1337 return None
1338 selected.append(x)
1339 upgrade_me.remove(x)
1341 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1342 output_logger.info("orig: %s", self.eval_nuninst(nuninst_start))
1344 if not (init and not force):
1345 # No "outer" transaction needed as we will never need to rollback
1346 # (e.g. "force-hint" or a regular "main run"). Emulate the start_transaction
1347 # call from the MigrationManager, so the rest of the code follows the
1348 # same flow regardless of whether we need the transaction or not.
1350 @contextlib.contextmanager
1351 def _start_transaction() -> Iterator[Optional["MigrationTransactionState"]]:
1352 yield None
1354 else:
1355 # We will need to be able to roll back (e.g. easy or a "hint"-hint)
1356 _start_transaction = mm.start_transaction
1358 with _start_transaction() as transaction:
1359 if init:
1360 # init => a hint (e.g. "easy") - so do the hint run
1361 (_, nuninst_end, _, new_cruft) = mm.migrate_items_to_target_suite(
1362 selected, self.nuninst_orig, stop_on_first_regression=False
1363 )
1365 if recurse:
1366 # Ensure upgrade_me and selected do not overlap, if we
1367 # follow-up with a recurse ("hint"-hint).
1368 selected_set = set(selected)
1369 upgrade_me = [x for x in upgrade_me if x not in selected_set]
1370 else:
1371 # On non-recursive hints check for cruft and purge it proactively in case it "fixes" the hint.
1372 cruft = [x for x in upgrade_me if x.is_cruft_removal]
1373 if new_cruft:
1374 output_logger.info(
1375 "Change added new cruft items to list: %s",
1376 " ".join(x.uvname for x in sorted(new_cruft)),
1377 )
1378 cruft.extend(new_cruft)
1379 if cruft:
1380 output_logger.info("Checking if changes enables cruft removal")
1381 (nuninst_end, remaining_cruft) = self.iter_packages(
1382 cruft, selected, nuninst=nuninst_end
1383 )
1384 output_logger.info(
1385 "Removed %d of %d cruft item(s) after the changes",
1386 len(cruft) - len(remaining_cruft),
1387 len(cruft),
1388 )
1389 new_cruft.difference_update(remaining_cruft)
1391 # Add new cruft items regardless of whether we recurse. A future run might clean
1392 # them for us.
1393 upgrade_me.extend(new_cruft)
1395 if recurse:
1396 # Either the main run or the recursive run of a "hint"-hint.
1397 (nuninst_end, extra) = self.iter_packages(
1398 upgrade_me, selected, nuninst=nuninst_end
1399 )
1401 assert nuninst_end is not None
1402 nuninst_end_str = self.eval_nuninst(nuninst_end)
1404 if not recurse:
1405 # easy or force-hint
1406 output_logger.info("easy: %s", nuninst_end_str)
1408 if not force:
1409 format_and_log_uninst(
1410 self.output_logger,
1411 self.options.architectures,
1412 newly_uninst(nuninst_start, nuninst_end),
1413 )
1415 if force:
1416 # Force implies "unconditionally better"
1417 better = True
1418 else:
1419 break_arches: set[str] = set(self.options.break_arches)
1420 if all(x.architecture in break_arches for x in selected):
1421 # If we only migrated items from break-arches, then we
1422 # do not allow any regressions on these architectures.
1423 # This usually only happens with hints
1424 break_arches = set()
1425 better = is_nuninst_asgood_generous(
1426 self.constraints,
1427 self.allow_uninst,
1428 self.options.architectures,
1429 self.nuninst_orig,
1430 nuninst_end,
1431 break_arches,
1432 )
1434 if better:
1435 # Result accepted either by force or by being better than the original result.
1436 output_logger.info(
1437 "final: %s", ",".join(sorted(x.uvname for x in selected))
1438 )
1439 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1440 output_logger.info(" orig: %s", self.eval_nuninst(self.nuninst_orig))
1441 output_logger.info(" end: %s", nuninst_end_str)
1442 if force:
1443 broken = newly_uninst(nuninst_start, nuninst_end)
1444 if broken:
1445 output_logger.warning("force breaks:")
1446 format_and_log_uninst(
1447 self.output_logger,
1448 self.options.architectures,
1449 broken,
1450 loglevel=logging.WARNING,
1451 )
1452 else:
1453 output_logger.info("force did not break any packages")
1454 output_logger.info(
1455 "SUCCESS (%d/%d)", len(actions or self.upgrade_me), len(extra)
1456 )
1457 self.nuninst_orig = nuninst_end
1458 self.all_selected += selected
1459 if transaction:
1460 transaction.commit()
1461 if self.options.check_consistency_level >= 2: 1461 ↛ 1465line 1461 didn't jump to line 1465 because the condition on line 1461 was always true
1462 target_suite.check_suite_source_pkg_consistency(
1463 "do_all after commit"
1464 )
1465 if not actions:
1466 if recurse:
1467 self.upgrade_me = extra
1468 else:
1469 selected_set = set(selected)
1470 self.upgrade_me = [
1471 x for x in self.upgrade_me if x not in selected_set
1472 ]
1473 else:
1474 output_logger.info("FAILED\n")
1475 if not transaction: 1475 ↛ 1479line 1475 didn't jump to line 1479 because the condition on line 1475 was never true
1476 # if we 'FAILED', but we cannot rollback, we will probably
1477 # leave a broken state behind
1478 # this should not happen
1479 raise AssertionError("do_all FAILED but no transaction to rollback")
1480 transaction.rollback()
1481 if self.options.check_consistency_level >= 2: 1481 ↛ 1358line 1481 didn't jump to line 1358
1482 target_suite.check_suite_source_pkg_consistency(
1483 "do_all after rollback"
1484 )
1486 output_logger.info("")
1488 def assert_nuninst_is_correct(self) -> None:
1489 self.logger.info("> Update complete - Verifying non-installability counters")
1491 cached_nuninst = self.nuninst_orig
1492 self._inst_tester.compute_installability()
1493 computed_nuninst = compile_nuninst(
1494 self.suite_info.target_suite,
1495 self.options.architectures,
1496 self.options.nobreakall_arches,
1497 )
1498 if cached_nuninst != computed_nuninst: # pragma: no cover
1499 only_on_break_archs = True
1500 msg_l = [
1501 "==================== NUNINST OUT OF SYNC ========================="
1502 ]
1503 for arch in self.options.architectures:
1504 expected_nuninst = set(cached_nuninst[arch])
1505 actual_nuninst = set(computed_nuninst[arch])
1506 false_negatives = actual_nuninst - expected_nuninst
1507 false_positives = expected_nuninst - actual_nuninst
1508 # Britney does not quite work correctly with
1509 # break/fucked arches, so ignore issues there for now.
1510 if (
1511 false_negatives or false_positives
1512 ) and arch not in self.options.break_arches:
1513 only_on_break_archs = False
1514 if false_negatives:
1515 msg_l.append(f" {arch} - unnoticed nuninst: {str(false_negatives)}")
1516 if false_positives:
1517 msg_l.append(f" {arch} - invalid nuninst: {str(false_positives)}")
1518 if false_negatives or false_positives:
1519 msg_l.append(
1520 f" {arch} - actual nuninst: {str(sorted(actual_nuninst))}"
1521 )
1522 msg_l.append(msg_l[0])
1523 for msg in msg_l:
1524 if only_on_break_archs:
1525 self.logger.warning(msg)
1526 else:
1527 self.logger.error(msg)
1528 if not only_on_break_archs:
1529 raise AssertionError("NUNINST OUT OF SYNC")
1530 else:
1531 self.logger.warning("Nuninst is out of sync on some break arches")
1533 self.logger.info("> All non-installability counters are ok")
1535 def upgrade_testing(self) -> None:
1536 """Upgrade testing using the packages from the source suites
1538 This method tries to upgrade testing using the packages from the
1539 source suites.
1540 Before running the do_all method, it tries the easy and force-hint
1541 commands.
1542 """
1544 output_logger = self.output_logger
1545 self.logger.info("Starting the upgrade test")
1546 output_logger.info(
1547 "Generated on: %s",
1548 time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time())),
1549 )
1550 output_logger.info("Arch order is: %s", ", ".join(self.options.architectures))
1552 if not self.options.actions: 1552 ↛ 1563line 1552 didn't jump to line 1563 because the condition on line 1552 was always true
1553 # process `easy' hints
1554 for x in self.hints["easy"]:
1555 self.do_hint("easy", x.user, x.packages)
1557 # process `force-hint' hints
1558 for x in self.hints["force-hint"]:
1559 self.do_hint("force-hint", x.user, x.packages)
1561 # run the first round of the upgrade
1562 # - do separate runs for break arches
1563 allpackages = []
1564 normpackages = self.upgrade_me[:]
1565 archpackages = {}
1566 for a in self.options.break_arches:
1567 archpackages[a] = [p for p in normpackages if p.architecture == a]
1568 normpackages = [p for p in normpackages if p.architecture != a]
1569 self.upgrade_me = normpackages
1570 output_logger.info("info: main run")
1571 self.do_all()
1572 allpackages += self.upgrade_me
1573 for a in self.options.break_arches:
1574 backup = self.options.break_arches
1575 self.options.break_arches = " ".join(
1576 x for x in self.options.break_arches if x != a
1577 )
1578 self.upgrade_me = archpackages[a]
1579 output_logger.info("info: broken arch run for %s", a)
1580 self.do_all()
1581 allpackages += self.upgrade_me
1582 self.options.break_arches = backup
1583 self.upgrade_me = allpackages
1585 if self.options.actions: 1585 ↛ 1586line 1585 didn't jump to line 1586 because the condition on line 1585 was never true
1586 self.printuninstchange()
1587 return
1589 # process `hint' hints
1590 hintcnt = 0
1591 for x in self.hints["hint"][:50]:
1592 if hintcnt > 50: 1592 ↛ 1593line 1592 didn't jump to line 1593 because the condition on line 1592 was never true
1593 output_logger.info("Skipping remaining hints...")
1594 break
1595 if self.do_hint("hint", x.user, x.packages): 1595 ↛ 1591line 1595 didn't jump to line 1591 because the condition on line 1595 was always true
1596 hintcnt += 1
1598 # run the auto hinter
1599 self.run_auto_hinter()
1601 if getattr(self.options, "remove_obsolete", "yes") == "yes":
1602 # obsolete source packages
1603 # a package is obsolete if none of the binary packages in testing
1604 # are built by it
1605 self.logger.info(
1606 "> Removing obsolete source packages from the target suite"
1607 )
1608 # local copies for performance
1609 target_suite = self.suite_info.target_suite
1610 sources_t = target_suite.sources
1611 binaries_t = target_suite.binaries
1612 mi_factory = self._migration_item_factory
1613 used = {
1614 binaries_t[arch][binary].source
1615 for arch in binaries_t
1616 for binary in binaries_t[arch]
1617 if not binary.endswith("-faux-build-depends")
1618 }
1619 removals = [
1620 mi_factory.parse_item(
1621 f"-{source}/{sources_t[source].version}", auto_correct=False
1622 )
1623 for source in sources_t
1624 if source not in used
1625 ]
1626 if removals:
1627 output_logger.info(
1628 "Removing obsolete source packages from the target suite (%d):",
1629 len(removals),
1630 )
1631 self.do_all(actions=removals)
1633 # smooth updates
1634 removals = old_libraries(
1635 self._migration_item_factory, self.suite_info, self.options.outofsync_arches
1636 )
1637 if removals:
1638 output_logger.info(
1639 "Removing packages left in the target suite (e.g. smooth updates or cruft)"
1640 )
1641 log_and_format_old_libraries(self.output_logger, removals)
1642 self.do_all(actions=removals)
1643 removals = old_libraries(
1644 self._migration_item_factory,
1645 self.suite_info,
1646 self.options.outofsync_arches,
1647 )
1649 output_logger.info(
1650 "List of old libraries in the target suite (%d):", len(removals)
1651 )
1652 log_and_format_old_libraries(self.output_logger, removals)
1654 self.printuninstchange()
1655 if self.options.check_consistency_level >= 1: 1655 ↛ 1661line 1655 didn't jump to line 1661 because the condition on line 1655 was always true
1656 target_suite = self.suite_info.target_suite
1657 self.assert_nuninst_is_correct()
1658 target_suite.check_suite_source_pkg_consistency("end")
1660 # output files
1661 if self.options.heidi_output and not self.options.dry_run: 1661 ↛ 1675line 1661 didn't jump to line 1675 because the condition on line 1661 was always true
1662 target_suite = self.suite_info.target_suite
1664 # write HeidiResult
1665 self.logger.info("Writing Heidi results to %s", self.options.heidi_output)
1666 write_heidi(
1667 self.options.heidi_output,
1668 target_suite,
1669 outofsync_arches=self.options.outofsync_arches,
1670 )
1672 self.logger.info("Writing delta to %s", self.options.heidi_delta_output)
1673 write_heidi_delta(self.options.heidi_delta_output, self.all_selected)
1675 self.logger.info("Test completed!")
1677 def printuninstchange(self) -> None:
1678 self.logger.info("Checking for newly uninstallable packages")
1679 uninst = newly_uninst(self.nuninst_orig_save, self.nuninst_orig)
1681 if uninst:
1682 self.output_logger.info("")
1683 self.output_logger.info(
1684 "Newly uninstallable packages in the target suite (arch:all on BREAKALL_ARCHES not shown)"
1685 )
1686 format_and_log_uninst(
1687 self.output_logger,
1688 self.options.architectures,
1689 uninst,
1690 loglevel=logging.WARNING,
1691 )
1693 def hint_tester(self) -> None:
1694 """Run a command line interface to test hints
1696 This method provides a command line interface for the release team to
1697 try hints and evaluate the results.
1698 """
1699 import readline
1701 from britney2.completer import Completer
1703 histfile = os.path.expanduser("~/.britney2_history")
1704 if os.path.exists(histfile):
1705 readline.read_history_file(histfile)
1707 readline.parse_and_bind("tab: complete")
1708 readline.set_completer(Completer(self).completer)
1709 # Package names can contain "-" and we use "/" in our presentation of them as well,
1710 # so ensure readline does not split on these characters.
1711 readline.set_completer_delims(
1712 readline.get_completer_delims().replace("-", "").replace("/", "")
1713 )
1715 known_hints = self._hint_parser.registered_hint_names
1717 print("Britney hint tester")
1718 print()
1719 print(
1720 "Besides inputting known britney hints, the follow commands are also available"
1721 )
1722 print(" * quit/exit - terminates the shell")
1723 print(
1724 " * python-console - jump into an interactive python shell (with the current loaded dataset)"
1725 )
1726 print()
1728 while True:
1729 # read the command from the command line
1730 try:
1731 user_input = input("britney> ").split()
1732 except EOFError:
1733 print("")
1734 break
1735 except KeyboardInterrupt:
1736 print("")
1737 continue
1738 match user_input:
1739 case ("quit" | "exit", *_):
1740 # quit the hint tester
1741 break
1742 case ("python-console", *_):
1743 try:
1744 import britney2.console
1745 except ImportError as e:
1746 print("Failed to import britney.console module: %s" % repr(e))
1747 continue
1748 britney2.console.run_python_console(self)
1749 print("Returning to the britney hint-tester console")
1750 # run a hint
1751 case ("easy" | "hint" | "force-hint" as choice, *items):
1752 mi_factory = self._migration_item_factory
1753 try:
1754 self.do_hint(
1755 choice,
1756 "hint-tester",
1757 mi_factory.parse_items(items),
1758 )
1759 self.printuninstchange()
1760 except KeyboardInterrupt:
1761 continue
1762 case (str() as hint, *_) if hint in known_hints:
1763 self._hint_parser.parse_hints(
1764 "hint-tester", self.HINTS_ALL, "<stdin>", [" ".join(user_input)]
1765 )
1766 self.write_excuses()
1768 try:
1769 readline.write_history_file(histfile)
1770 except OSError as e:
1771 self.logger.warning("Could not write %s: %s", histfile, e)
1773 def do_hint(self, hinttype: str, who: str, pkgvers: list[MigrationItem]) -> bool:
1774 """Process hints
1776 This method process `easy`, `hint` and `force-hint` hints. If the
1777 requested version is not in the relevant source suite, then the hint
1778 is skipped.
1779 """
1781 output_logger = self.output_logger
1783 self.logger.info("> Processing '%s' hint from %s", hinttype, who)
1784 output_logger.info(
1785 "Trying %s from %s: %s",
1786 hinttype,
1787 who,
1788 " ".join(f"{x.uvname}/{x.version}" for x in pkgvers),
1789 )
1791 issues = []
1792 # loop on the requested packages and versions
1793 for idx in range(len(pkgvers)):
1794 pkg = pkgvers[idx]
1795 # skip removal requests
1796 if pkg.is_removal:
1797 continue
1799 suite = pkg.suite
1801 assert pkg.version is not None
1802 if pkg.package not in suite.sources: 1802 ↛ 1803line 1802 didn't jump to line 1803 because the condition on line 1802 was never true
1803 issues.append(f"Source {pkg.package} has no version in {suite.name}")
1804 elif ( 1804 ↛ 1808line 1804 didn't jump to line 1808
1805 apt_pkg.version_compare(suite.sources[pkg.package].version, pkg.version)
1806 != 0
1807 ):
1808 issues.append(
1809 "Version mismatch, %s %s != %s"
1810 % (pkg.package, pkg.version, suite.sources[pkg.package].version)
1811 )
1812 if issues: 1812 ↛ 1813line 1812 didn't jump to line 1813 because the condition on line 1812 was never true
1813 output_logger.warning("%s: Not using hint", ", ".join(issues))
1814 return False
1816 self.do_all(hinttype, pkgvers)
1817 return True
1819 def get_auto_hinter_hints(
1820 self, upgrade_me: list[MigrationItem]
1821 ) -> list[list[frozenset[MigrationItem]]]:
1822 """Auto-generate "easy" hints.
1824 This method attempts to generate "easy" hints for sets of packages which
1825 must migrate together. Beginning with a package which does not depend on
1826 any other package (in terms of excuses), a list of dependencies and
1827 reverse dependencies is recursively created.
1829 Once all such lists have been generated, any which are subsets of other
1830 lists are ignored in favour of the larger lists. The remaining lists are
1831 then attempted in turn as "easy" hints.
1833 We also try to auto hint circular dependencies analyzing the update
1834 excuses relationships. If they build a circular dependency, which we already
1835 know as not-working with the standard do_all algorithm, try to `easy` them.
1836 """
1837 self.logger.info("> Processing hints from the auto hinter")
1839 sources_t = self.suite_info.target_suite.sources
1840 excuses = self.excuses
1842 def excuse_still_valid(excuse: "Excuse") -> bool:
1843 source = excuse.source
1844 assert isinstance(excuse.item, MigrationItem)
1845 arch = excuse.item.architecture
1846 # TODO for binNMUs, this check is always ok, even if the item
1847 # migrated already
1848 valid = (
1849 arch != "source"
1850 or source not in sources_t
1851 or sources_t[source].version != excuse.ver[1]
1852 )
1853 # TODO migrated items should be removed from upgrade_me, so this
1854 # should not happen
1855 if not valid: 1855 ↛ 1856line 1855 didn't jump to line 1856 because the condition on line 1855 was never true
1856 raise AssertionError("excuse no longer valid %s" % (item))
1857 return valid
1859 # consider only excuses which are valid candidates and still relevant.
1860 valid_excuses = frozenset(
1861 e.name
1862 for n, e in excuses.items()
1863 if e.item in upgrade_me and excuse_still_valid(e)
1864 )
1865 excuses_deps = {
1866 name: valid_excuses.intersection(excuse.get_deps())
1867 for name, excuse in excuses.items()
1868 if name in valid_excuses
1869 }
1870 excuses_rdeps = defaultdict(set)
1871 for name, deps in excuses_deps.items():
1872 for dep in deps:
1873 excuses_rdeps[dep].add(name)
1875 # loop on them
1876 candidates = []
1877 mincands = []
1878 seen_hints = set()
1879 for e in valid_excuses:
1880 excuse = excuses[e]
1881 if not excuse.get_deps():
1882 assert isinstance(excuse.item, MigrationItem)
1883 items = [excuse.item]
1884 orig_size = 1
1885 looped = False
1886 seen_items = set()
1887 seen_items.update(items)
1889 for item in items:
1890 assert isinstance(item, MigrationItem)
1891 # excuses which depend on "item" or are depended on by it
1892 new_items = cast(
1893 set[MigrationItem],
1894 {
1895 excuses[x].item
1896 for x in chain(
1897 excuses_deps[item.name], excuses_rdeps[item.name]
1898 )
1899 },
1900 )
1901 new_items -= seen_items
1902 items.extend(new_items)
1903 seen_items.update(new_items)
1905 if not looped and len(items) > 1:
1906 orig_size = len(items)
1907 h = frozenset(seen_items)
1908 if h not in seen_hints: 1908 ↛ 1911line 1908 didn't jump to line 1911 because the condition on line 1908 was always true
1909 mincands.append(h)
1910 seen_hints.add(h)
1911 looped = True
1912 if len(items) != orig_size: 1912 ↛ 1913line 1912 didn't jump to line 1913 because the condition on line 1912 was never true
1913 h = frozenset(seen_items)
1914 if h != mincands[-1] and h not in seen_hints:
1915 candidates.append(h)
1916 seen_hints.add(h)
1917 return [candidates, mincands]
1919 def run_auto_hinter(self) -> None:
1920 for lst in self.get_auto_hinter_hints(self.upgrade_me):
1921 for hint in lst:
1922 self.do_hint("easy", "autohinter", sorted(hint))
1924 def nuninst_arch_report(self, nuninst: dict[str, set[str]], arch: str) -> None:
1925 """Print a report of uninstallable packages for one architecture."""
1926 all = defaultdict(set)
1927 binaries_t = self.suite_info.target_suite.binaries
1928 for p in nuninst[arch]:
1929 pkg = binaries_t[arch][p]
1930 all[(pkg.source, pkg.source_version)].add(p)
1932 print("* %s" % arch)
1934 for (src, ver), pkgs in sorted(all.items()):
1935 print(" {} ({}): {}".format(src, ver, " ".join(sorted(pkgs))))
1937 print()
1939 def _remove_archall_faux_packages(self) -> None:
1940 """Remove faux packages added for the excuses phase
1942 To prevent binary packages from going missing while they are listed by
1943 their source package we add bin:faux packages during reading in the
1944 Sources. They are used during the excuses phase to prevent packages
1945 from becoming candidates. However, they interfere in complex ways
1946 during the installability phase, so instead of having all code during
1947 migration be aware of this excuses phase implementation detail, let's
1948 remove them again.
1950 """
1951 if not self.options.archall_inconsistency_allowed:
1952 all_binaries = self.all_binaries
1953 faux_a = {x for x in all_binaries.keys() if x.architecture == "faux"}
1954 for pkg_a in faux_a:
1955 del all_binaries[pkg_a]
1957 for suite in self.suite_info._suites.values():
1958 for arch in suite.binaries.keys():
1959 binaries = suite.binaries[arch]
1960 faux_b = {
1961 x for x in binaries if binaries[x].pkg_id.architecture == "faux"
1962 }
1963 for pkg_b in faux_b:
1964 del binaries[pkg_b]
1965 sources = suite.sources
1966 for src in sources.keys():
1967 faux_s = {
1968 x for x in sources[src].binaries if x.architecture == "faux"
1969 }
1970 sources[src].binaries -= faux_s
1972 def main(self) -> None:
1973 """Main method
1975 This is the entry point for the class: it includes the list of calls
1976 for the member methods which will produce the output files.
1977 """
1978 # if running in --print-uninst mode, quit
1979 if self.options.print_uninst: 1979 ↛ 1980line 1979 didn't jump to line 1980 because the condition on line 1979 was never true
1980 return
1981 # if no actions are provided, build the excuses and sort them
1982 elif not self.options.actions: 1982 ↛ 1986line 1982 didn't jump to line 1986 because the condition on line 1982 was always true
1983 self.write_excuses()
1984 # otherwise, use the actions provided by the command line
1985 else:
1986 self.upgrade_me = self.options.actions.split()
1988 self._remove_archall_faux_packages()
1990 if self.options.compute_migrations or self.options.hint_tester:
1991 if self.options.dry_run: 1991 ↛ 1992line 1991 didn't jump to line 1992 because the condition on line 1991 was never true
1992 self.logger.info(
1993 "Upgrade output not (also) written to a separate file"
1994 " as this is a dry-run."
1995 )
1996 elif hasattr(self.options, "upgrade_output"): 1996 ↛ 2006line 1996 didn't jump to line 2006 because the condition on line 1996 was always true
1997 upgrade_output = getattr(self.options, "upgrade_output")
1998 file_handler = logging.FileHandler(
1999 upgrade_output, mode="w", encoding="utf-8"
2000 )
2001 output_formatter = logging.Formatter("%(message)s")
2002 file_handler.setFormatter(output_formatter)
2003 self.output_logger.addHandler(file_handler)
2004 self.logger.info("Logging upgrade output to %s", upgrade_output)
2005 else:
2006 self.logger.info(
2007 "Upgrade output not (also) written to a separate file"
2008 " as the UPGRADE_OUTPUT configuration is not provided."
2009 )
2011 # run the hint tester
2012 if self.options.hint_tester: 2012 ↛ 2013line 2012 didn't jump to line 2013 because the condition on line 2012 was never true
2013 self.hint_tester()
2014 # run the upgrade test
2015 else:
2016 self.upgrade_testing()
2018 self.logger.info("> Stats from the installability tester")
2019 for stat in self._inst_tester.stats.stats():
2020 self.logger.info("> %s", stat)
2021 else:
2022 self.logger.info("Migration computation skipped as requested.")
2023 if not self.options.dry_run: 2023 ↛ 2025line 2023 didn't jump to line 2025 because the condition on line 2023 was always true
2024 self._policy_engine.save_state(self)
2025 logging.shutdown()
2028if __name__ == "__main__": 2028 ↛ 2029line 2028 didn't jump to line 2029 because the condition on line 2028 was never true
2029 Britney().main()