Coverage for britney2/britney.py: 83%
770 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +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 Generator
191from functools import reduce
192from itertools import chain
193from operator import attrgetter
194from typing import TYPE_CHECKING, Any, Optional
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 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 f"Missing Package field in paragraph {no} (file {faux_packages_file})"
722 )
723 version = get_field("Version", "1.0-1")
724 provides_raw = get_field("Provides")
725 archs_raw = get_field("Architecture", None)
726 component = get_field("Component", "non-free")
727 if archs_raw: 727 ↛ 728line 727 didn't jump to line 728 because the condition on line 727 was never true
728 archs = archs_raw.split()
729 else:
730 archs = self.options.architectures
731 faux_section = "faux"
732 if component != "main": 732 ↛ 734line 732 didn't jump to line 734 because the condition on line 732 was always true
733 faux_section = f"{component}/faux"
734 src_data = SourcePackage(
735 pkg_name,
736 version,
737 faux_section,
738 set(),
739 None,
740 True,
741 )
743 target_suite.sources[pkg_name] = src_data
744 pri_source_suite.sources[pkg_name] = src_data
746 for arch in archs:
747 pkg_id = BinaryPackageId(pkg_name, version, arch)
748 if provides_raw: 748 ↛ 749line 748 didn't jump to line 749 because the condition on line 748 was never true
749 provides = parse_provides(
750 provides_raw, pkg_id=pkg_id, logger=self.logger
751 )
752 else:
753 provides = None
754 bin_data = BinaryPackage(
755 faux_section,
756 pkg_name,
757 version,
758 arch,
759 MultiArch.from_str(get_field("Multi-Arch")),
760 None,
761 None,
762 provides,
763 False,
764 pkg_id,
765 None,
766 )
768 src_data.binaries.add(pkg_id)
769 target_suite.binaries[arch][pkg_name] = bin_data
770 pri_source_suite.binaries[arch][pkg_name] = bin_data
772 # register provided packages with the target suite provides table
773 for provided_pkg, provided_version in ( 773 ↛ 776line 773 didn't jump to line 776
774 bin_data.provides if bin_data.provides is not None else []
775 ):
776 target_suite.provides_table[arch][provided_pkg].add(
777 (pkg_name, provided_version)
778 )
780 self.all_binaries[pkg_id] = bin_data
782 def _load_constraints(self, constraints_file: str) -> dict[str, list[str]]:
783 """Loads configurable constraints
785 The constraints file can contain extra rules that Britney should attempt
786 to satisfy. Examples can be "keep package X in testing and ensure it is
787 installable".
789 :param constraints_file: Path to the file containing the constraints
790 """
791 tag_file = apt_pkg.TagFile(constraints_file)
792 get_field = tag_file.section.get
793 step = tag_file.step
794 no = 0
795 faux_version = "1"
796 faux_section = "faux"
797 keep_installable: list[str] = []
798 constraints = {"keep-installable": keep_installable}
799 pri_source_suite = self.suite_info.primary_source_suite
800 target_suite = self.suite_info.target_suite
802 while step():
803 no += 1
804 pkg_name = get_field("Fake-Package-Name", None)
805 if pkg_name is None: # pragma: no cover
806 raise ValueError(
807 f"Missing Fake-Package-Name field in paragraph {no} (file {constraints_file})"
808 )
810 def mandatory_field(x: str) -> str:
811 v: str = get_field(x, None)
812 if v is None: # pragma: no cover
813 raise ValueError(
814 f"Missing {x} field for {pkg_name} (file {constraints_file})"
815 )
816 return v
818 constraint = mandatory_field("Constraint")
819 if constraint not in {"present-and-installable"}: # pragma: no cover
820 raise ValueError(
821 f"Unsupported constraint {constraint} for {pkg_name} (file {constraints_file})"
822 )
824 self.logger.info(" - constraint %s", pkg_name)
826 pkg_list = [
827 x.strip()
828 for x in mandatory_field("Package-List").split("\n")
829 if x.strip() != "" and not x.strip().startswith("#")
830 ]
831 src_data = SourcePackage(
832 pkg_name,
833 faux_version,
834 faux_section,
835 set(),
836 None,
837 True,
838 )
839 target_suite.sources[pkg_name] = src_data
840 pri_source_suite.sources[pkg_name] = src_data
841 keep_installable.append(pkg_name)
842 for arch in self.options.architectures:
843 deps = []
844 for pkg_spec in pkg_list:
845 s = pkg_spec.split(None, 1)
846 if len(s) == 1:
847 deps.append(s[0])
848 else:
849 pkg, arch_res = s
850 if not (
851 arch_res.startswith("[") and arch_res.endswith("]")
852 ): # pragma: no cover
853 raise ValueError(
854 f"Invalid arch-restriction on {pkg} - should be [arch1 arch2] "
855 f"(for {pkg_name} file {constraints_file})"
856 )
857 arch_res_l = arch_res[1:-1].split()
858 if not arch_res_l: # pragma: no cover
859 raise ValueError(
860 f"Empty arch-restriction for {pkg}: Uses comma or negation "
861 f"(for {pkg_name} file {constraints_file})"
862 )
863 for a in arch_res_l:
864 if a == arch:
865 deps.append(pkg)
866 elif "," in a or "!" in a: # pragma: no cover
867 raise ValueError(
868 f"Invalid arch-restriction for {pkg}: Uses comma or negation "
869 f"(for {pkg_name} file {constraints_file})"
870 )
871 pkg_id = BinaryPackageId(pkg_name, faux_version, arch)
872 bin_data = BinaryPackage(
873 faux_section,
874 pkg_name,
875 faux_version,
876 arch,
877 MultiArch.NO,
878 ", ".join(deps),
879 None,
880 None,
881 False,
882 pkg_id,
883 None,
884 )
885 src_data.binaries.add(pkg_id)
886 target_suite.binaries[arch][pkg_name] = bin_data
887 pri_source_suite.binaries[arch][pkg_name] = bin_data
888 self.all_binaries[pkg_id] = bin_data
890 return constraints
892 # Data reading/writing methods
893 # ----------------------------
895 def read_hints(self, hintsdir: str) -> None:
896 """Read the hint commands from the specified directory
898 The hint commands are read from the files contained in the directory
899 specified by the `hintsdir' parameter.
900 The names of the files have to be the same as the authorized users
901 for the hints.
903 The file contains rows with the format:
905 <command> <package-name>[/<version>]
907 The method returns a dictionary where the key is the command, and
908 the value is the list of affected packages.
909 """
911 for who in self.HINTS.keys():
912 if who == "command-line":
913 lines = self.options.hints and self.options.hints.split(";") or ()
914 filename = "<cmd-line>"
915 self._hint_parser.parse_hints(who, self.HINTS[who], filename, lines)
916 else:
917 filename = os.path.join(hintsdir, who)
918 if not os.path.isfile(filename): 918 ↛ 919line 918 didn't jump to line 919 because the condition on line 918 was never true
919 self.logger.error(
920 "Cannot read hints list from %s, no such file!", filename
921 )
922 continue
923 self.logger.info("Loading hints list from %s", filename)
924 with open(filename, encoding="utf-8") as f:
925 self._hint_parser.parse_hints(who, self.HINTS[who], filename, f)
927 hints = self._hint_parser.hints
929 for x in (
930 "block",
931 "block-all",
932 "block-udeb",
933 "unblock",
934 "unblock-udeb",
935 "force",
936 "urgent",
937 "remove",
938 "age-days",
939 ):
940 z: dict[str | None, dict[str | None, tuple[Hint, str]]] = defaultdict(dict)
941 for hint in hints[x]:
942 package = hint.package
943 architecture = hint.architecture
944 key = (hint, hint.user)
945 if (
946 package in z
947 and architecture in z[package]
948 and z[package][architecture] != key
949 ):
950 hint2 = z[package][architecture][0]
951 if x in ("unblock", "unblock-udeb", "age-days"): 951 ↛ 983line 951 didn't jump to line 983 because the condition on line 951 was always true
952 assert hint.version is not None
953 assert hint2.version is not None
954 if apt_pkg.version_compare(hint2.version, hint.version) < 0:
955 # This hint is for a newer version, so discard the old one
956 self.logger.warning(
957 "Overriding %s[%s] = ('%s', '%s', '%s') with ('%s', '%s', '%s')",
958 x,
959 package,
960 hint2.version,
961 hint2.architecture,
962 hint2.user,
963 hint.version,
964 hint.architecture,
965 hint.user,
966 )
967 hint2.set_active(False)
968 else:
969 # This hint is for an older version, so ignore it in favour of the new one
970 self.logger.warning(
971 "Ignoring %s[%s] = ('%s', '%s', '%s'), ('%s', '%s', '%s') is higher or equal",
972 x,
973 package,
974 hint.version,
975 hint.architecture,
976 hint.user,
977 hint2.version,
978 hint2.architecture,
979 hint2.user,
980 )
981 hint.set_active(False)
982 else:
983 self.logger.warning(
984 "Overriding %s[%s] = ('%s', '%s') with ('%s', '%s')",
985 x,
986 package,
987 hint2.user,
988 hint2,
989 hint.user,
990 hint,
991 )
992 hint2.set_active(False)
994 z[package][architecture] = key
996 for hint in hints["allow-uninst"]:
997 if hint.architecture == "source":
998 for arch in self.options.architectures:
999 self.allow_uninst[arch].add(hint.package)
1000 else:
1001 assert hint.architecture is not None
1002 self.allow_uninst[hint.architecture].add(hint.package)
1004 # Sanity check the hints hash
1005 if len(hints["block"]) == 0 and len(hints["block-udeb"]) == 0: 1005 ↛ 1006line 1005 didn't jump to line 1006 because the condition on line 1005 was never true
1006 self.logger.warning("WARNING: No block hints at all, not even udeb ones!")
1008 # Remove all hints that were set inactive.
1009 # We don't need to keep unused hints in memory.
1010 hints.remove_inactive_hints()
1012 def write_excuses(self) -> None:
1013 """Produce and write the update excuses
1015 This method handles the update excuses generation: the packages are
1016 looked at to determine whether they are valid candidates. For the details
1017 of this procedure, please refer to the module docstring.
1018 """
1020 self.logger.info("Update Excuses generation started")
1022 mi_factory = self._migration_item_factory
1023 excusefinder = ExcuseFinder(
1024 self.options,
1025 self.suite_info,
1026 self.all_binaries,
1027 self.pkg_universe,
1028 self._policy_engine,
1029 mi_factory,
1030 self.hints,
1031 )
1033 excuses, upgrade_me = excusefinder.find_actionable_excuses()
1034 self.excuses = excuses
1036 # sort the list of candidates
1037 self.upgrade_me = sorted(upgrade_me)
1038 old_lib_removals = old_libraries(
1039 mi_factory, self.suite_info, self.options.outofsync_arches
1040 )
1041 self.upgrade_me.extend(old_lib_removals)
1042 self.output_logger.info(
1043 "List of old libraries added to upgrade_me (%d):", len(old_lib_removals)
1044 )
1045 log_and_format_old_libraries(self.output_logger, old_lib_removals)
1047 # write excuses to the output file
1048 if not self.options.dry_run: 1048 ↛ 1065line 1048 didn't jump to line 1065 because the condition on line 1048 was always true
1049 self.logger.info("> Writing Excuses to %s", self.options.excuses_output)
1050 write_excuses(
1051 excuses,
1052 self.options.excuses_output,
1053 output_format=ExcusesOutputFormat.LEGACY_HTML,
1054 )
1055 if hasattr(self.options, "excuses_yaml_output"): 1055 ↛ 1065line 1055 didn't jump to line 1065 because the condition on line 1055 was always true
1056 self.logger.info(
1057 "> Writing YAML Excuses to %s", self.options.excuses_yaml_output
1058 )
1059 write_excuses(
1060 excuses,
1061 self.options.excuses_yaml_output,
1062 output_format=ExcusesOutputFormat.YAML,
1063 )
1065 self.logger.info("Update Excuses generation completed")
1067 # Upgrade run
1068 # -----------
1070 def eval_nuninst(
1071 self,
1072 nuninst: dict[str, set[str]],
1073 original: dict[str, set[str]] | None = None,
1074 ) -> str:
1075 """Return a string which represents the uninstallability counters
1077 This method returns a string which represents the uninstallability
1078 counters reading the uninstallability statistics `nuninst` and, if
1079 present, merging the results with the `original` one.
1081 An example of the output string is:
1082 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
1084 where the first part is the number of broken packages in non-break
1085 architectures + the total number of broken packages for all the
1086 architectures.
1087 """
1088 res = []
1089 total = 0
1090 totalbreak = 0
1091 for arch in self.options.architectures:
1092 if arch in nuninst: 1092 ↛ 1094line 1092 didn't jump to line 1094 because the condition on line 1092 was always true
1093 n = len(nuninst[arch])
1094 elif original and arch in original:
1095 n = len(original[arch])
1096 else:
1097 continue
1098 if arch in self.options.break_arches:
1099 totalbreak = totalbreak + n
1100 else:
1101 total = total + n
1102 res.append(f"{arch[0]}-{n}")
1103 return "%d+%d: %s" % (total, totalbreak, ":".join(res))
1105 def iter_packages(
1106 self,
1107 packages: list[MigrationItem],
1108 selected: list[MigrationItem],
1109 nuninst: dict[str, set[str]] | None = None,
1110 ) -> tuple[dict[str, set[str]] | None, list[MigrationItem]]:
1111 """Iter on the list of actions and apply them one-by-one
1113 This method applies the changes from `packages` to testing, checking the uninstallability
1114 counters for every action performed. If the action does not improve them, it is reverted.
1115 The method returns the new uninstallability counters and the remaining actions if the
1116 final result is successful, otherwise (None, []).
1118 :param selected: list of MigrationItem?
1119 :param nuninst: dict with sets ? of ? per architecture
1120 """
1121 assert self.suite_info is not None # for type checking
1122 group_info = {}
1123 rescheduled_packages = packages
1124 maybe_rescheduled_packages: list[MigrationItem] = []
1125 output_logger = self.output_logger
1126 solver = InstallabilitySolver(self.pkg_universe, self._inst_tester)
1127 mm = self._migration_manager
1128 target_suite = self.suite_info.target_suite
1130 for y in sorted((y for y in packages), key=attrgetter("uvname")):
1131 try:
1132 _, updates, rms, _ = mm.compute_groups(y)
1133 result = (y, sorted(updates), sorted(rms))
1134 group_info[y] = result
1135 except MigrationConstraintException as e:
1136 rescheduled_packages.remove(y)
1137 output_logger.info("not adding package to list: %s", y.package)
1138 output_logger.info(" got exception: %r", e)
1140 if nuninst:
1141 nuninst_orig = nuninst
1142 else:
1143 nuninst_orig = self.nuninst_orig
1145 nuninst_last_accepted = nuninst_orig
1147 output_logger.info(
1148 "recur: [] %s %d/0", ",".join(x.uvname for x in selected), len(packages)
1149 )
1150 while rescheduled_packages:
1151 worklist = solver.solve_groups(group_info[x] for x in rescheduled_packages)
1152 rescheduled_packages = []
1154 worklist.reverse()
1156 while worklist:
1157 comp = worklist.pop()
1158 comp_name = " ".join(item.uvname for item in comp)
1159 output_logger.info("trying: %s", comp_name)
1160 with mm.start_transaction() as transaction:
1161 accepted = False
1162 try:
1163 (
1164 accepted,
1165 nuninst_after,
1166 failed_arch,
1167 new_cruft,
1168 ) = mm.migrate_items_to_target_suite(
1169 comp, nuninst_last_accepted
1170 )
1171 if accepted:
1172 selected.extend(comp)
1173 transaction.commit()
1174 output_logger.info("accepted: %s", comp_name)
1175 output_logger.info(
1176 " ori: %s", self.eval_nuninst(nuninst_orig)
1177 )
1178 output_logger.info(
1179 " pre: %s", self.eval_nuninst(nuninst_last_accepted)
1180 )
1181 output_logger.info(
1182 " now: %s", self.eval_nuninst(nuninst_after)
1183 )
1184 if new_cruft:
1185 output_logger.info(
1186 " added new cruft items to list: %s",
1187 " ".join(x.uvname for x in sorted(new_cruft)),
1188 )
1190 if len(selected) <= 20:
1191 output_logger.info(
1192 " all: %s", " ".join(x.uvname for x in selected)
1193 )
1194 else:
1195 output_logger.info(
1196 " most: (%d) .. %s",
1197 len(selected),
1198 " ".join(x.uvname for x in selected[-20:]),
1199 )
1200 if self.options.check_consistency_level >= 3: 1200 ↛ 1201line 1200 didn't jump to line 1201 because the condition on line 1200 was never true
1201 target_suite.check_suite_source_pkg_consistency(
1202 "iter_packages after commit"
1203 )
1204 nuninst_last_accepted = nuninst_after
1205 for cruft_item in new_cruft:
1206 try:
1207 _, updates, rms, _ = mm.compute_groups(cruft_item)
1208 result = (cruft_item, sorted(updates), sorted(rms))
1209 group_info[cruft_item] = result
1210 worklist.append([cruft_item])
1211 except MigrationConstraintException as e:
1212 output_logger.info(
1213 " got exception adding cruft item %s to list: %r",
1214 cruft_item.uvname,
1215 e,
1216 )
1217 rescheduled_packages.extend(maybe_rescheduled_packages)
1218 maybe_rescheduled_packages.clear()
1219 else:
1220 transaction.rollback()
1221 assert failed_arch # type checking
1222 broken = sorted(
1223 b
1224 for b in nuninst_after[failed_arch]
1225 if b not in nuninst_last_accepted[failed_arch]
1226 )
1227 compare_nuninst = None
1228 if any(
1229 item for item in comp if item.architecture != "source"
1230 ):
1231 compare_nuninst = nuninst_last_accepted
1232 # NB: try_migration already reverted this for us, so just print the results and move on
1233 output_logger.info(
1234 "skipped: %s (%d, %d, %d)",
1235 comp_name,
1236 len(rescheduled_packages),
1237 len(maybe_rescheduled_packages),
1238 len(worklist),
1239 )
1240 output_logger.info(
1241 " got: %s",
1242 self.eval_nuninst(nuninst_after, compare_nuninst),
1243 )
1244 output_logger.info(
1245 " * %s: %s", failed_arch, ", ".join(broken)
1246 )
1247 if self.options.check_consistency_level >= 3: 1247 ↛ 1248line 1247 didn't jump to line 1248 because the condition on line 1247 was never true
1248 target_suite.check_suite_source_pkg_consistency(
1249 "iter_package after rollback (not accepted)"
1250 )
1252 except MigrationConstraintException as e:
1253 transaction.rollback()
1254 output_logger.info(
1255 "skipped: %s (%d, %d, %d)",
1256 comp_name,
1257 len(rescheduled_packages),
1258 len(maybe_rescheduled_packages),
1259 len(worklist),
1260 )
1261 output_logger.info(" got exception: %r", e)
1262 if self.options.check_consistency_level >= 3: 1262 ↛ 1263line 1262 didn't jump to line 1263 because the condition on line 1262 was never true
1263 target_suite.check_suite_source_pkg_consistency(
1264 "iter_package after rollback (MigrationConstraintException)"
1265 )
1267 if not accepted:
1268 if len(comp) > 1:
1269 output_logger.info(
1270 " - splitting the component into single items and retrying them"
1271 )
1272 worklist.extend([item] for item in comp)
1273 else:
1274 maybe_rescheduled_packages.append(comp[0])
1276 output_logger.info(" finish: [%s]", ",".join(x.uvname for x in selected))
1277 output_logger.info("endloop: %s", self.eval_nuninst(self.nuninst_orig))
1278 output_logger.info(" now: %s", self.eval_nuninst(nuninst_last_accepted))
1279 format_and_log_uninst(
1280 output_logger,
1281 self.options.architectures,
1282 newly_uninst(self.nuninst_orig, nuninst_last_accepted),
1283 )
1284 output_logger.info("")
1286 return (nuninst_last_accepted, maybe_rescheduled_packages)
1288 def do_all(
1289 self,
1290 hinttype: str | None = None,
1291 init: list[MigrationItem] | None = None,
1292 actions: list[MigrationItem] | None = None,
1293 ) -> None:
1294 """Testing update runner
1296 This method tries to update testing checking the uninstallability
1297 counters before and after the actions to decide if the update was
1298 successful or not.
1299 """
1300 selected = []
1301 if actions:
1302 upgrade_me = actions[:]
1303 else:
1304 upgrade_me = self.upgrade_me[:]
1305 nuninst_start = self.nuninst_orig
1306 output_logger = self.output_logger
1307 target_suite = self.suite_info.target_suite
1309 # these are special parameters for hints processing
1310 force = False
1311 recurse = True
1312 nuninst_end = None
1313 extra: list[MigrationItem] = []
1314 mm = self._migration_manager
1316 if hinttype == "easy" or hinttype == "force-hint":
1317 force = hinttype == "force-hint"
1318 recurse = False
1320 # if we have a list of initial packages, check them
1321 if init:
1322 for x in init:
1323 if x not in upgrade_me:
1324 output_logger.warning(
1325 "failed: %s is not a valid candidate (or it already migrated)",
1326 x.uvname,
1327 )
1328 return None
1329 selected.append(x)
1330 upgrade_me.remove(x)
1332 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1333 output_logger.info("orig: %s", self.eval_nuninst(nuninst_start))
1335 if not (init and not force):
1336 # No "outer" transaction needed as we will never need to rollback
1337 # (e.g. "force-hint" or a regular "main run"). Emulate the start_transaction
1338 # call from the MigrationManager, so the rest of the code follows the
1339 # same flow regardless of whether we need the transaction or not.
1341 @contextlib.contextmanager
1342 def _start_transaction() -> (
1343 Generator[Optional["MigrationTransactionState"]]
1344 ):
1345 yield None
1347 else:
1348 # We will need to be able to roll back (e.g. easy or a "hint"-hint)
1349 _start_transaction = mm.start_transaction
1351 with _start_transaction() as transaction:
1352 if init:
1353 # init => a hint (e.g. "easy") - so do the hint run
1354 (_, nuninst_end, _, new_cruft) = mm.migrate_items_to_target_suite(
1355 selected, self.nuninst_orig, stop_on_first_regression=False
1356 )
1358 if recurse:
1359 # Ensure upgrade_me and selected do not overlap, if we
1360 # follow-up with a recurse ("hint"-hint).
1361 selected_set = set(selected)
1362 upgrade_me = [x for x in upgrade_me if x not in selected_set]
1363 else:
1364 # On non-recursive hints check for cruft and purge it proactively in case it "fixes" the hint.
1365 cruft = [x for x in upgrade_me if x.is_cruft_removal]
1366 if new_cruft:
1367 output_logger.info(
1368 "Change added new cruft items to list: %s",
1369 " ".join(x.uvname for x in sorted(new_cruft)),
1370 )
1371 cruft.extend(new_cruft)
1372 if cruft:
1373 output_logger.info("Checking if changes enables cruft removal")
1374 (nuninst_end, remaining_cruft) = self.iter_packages(
1375 cruft, selected, nuninst=nuninst_end
1376 )
1377 output_logger.info(
1378 "Removed %d of %d cruft item(s) after the changes",
1379 len(cruft) - len(remaining_cruft),
1380 len(cruft),
1381 )
1382 new_cruft.difference_update(remaining_cruft)
1384 # Add new cruft items regardless of whether we recurse. A future run might clean
1385 # them for us.
1386 upgrade_me.extend(new_cruft)
1388 if recurse:
1389 # Either the main run or the recursive run of a "hint"-hint.
1390 (nuninst_end, extra) = self.iter_packages(
1391 upgrade_me, selected, nuninst=nuninst_end
1392 )
1394 assert nuninst_end is not None
1395 nuninst_end_str = self.eval_nuninst(nuninst_end)
1397 if not recurse:
1398 # easy or force-hint
1399 output_logger.info("easy: %s", nuninst_end_str)
1401 if not force:
1402 format_and_log_uninst(
1403 self.output_logger,
1404 self.options.architectures,
1405 newly_uninst(nuninst_start, nuninst_end),
1406 )
1408 if force:
1409 # Force implies "unconditionally better"
1410 better = True
1411 else:
1412 break_arches: set[str] = set(self.options.break_arches)
1413 if all(x.architecture in break_arches for x in selected):
1414 # If we only migrated items from break-arches, then we
1415 # do not allow any regressions on these architectures.
1416 # This usually only happens with hints
1417 break_arches = set()
1418 better = is_nuninst_asgood_generous(
1419 self.constraints,
1420 self.allow_uninst,
1421 self.options.architectures,
1422 self.nuninst_orig,
1423 nuninst_end,
1424 break_arches,
1425 )
1427 if better:
1428 # Result accepted either by force or by being better than the original result.
1429 output_logger.info(
1430 "final: %s", ",".join(sorted(x.uvname for x in selected))
1431 )
1432 output_logger.info("start: %s", self.eval_nuninst(nuninst_start))
1433 output_logger.info(" orig: %s", self.eval_nuninst(self.nuninst_orig))
1434 output_logger.info(" end: %s", nuninst_end_str)
1435 if force:
1436 broken = newly_uninst(nuninst_start, nuninst_end)
1437 if broken:
1438 output_logger.warning("force breaks:")
1439 format_and_log_uninst(
1440 self.output_logger,
1441 self.options.architectures,
1442 broken,
1443 loglevel=logging.WARNING,
1444 )
1445 else:
1446 output_logger.info("force did not break any packages")
1447 output_logger.info(
1448 "SUCCESS (%d/%d)", len(actions or self.upgrade_me), len(extra)
1449 )
1450 self.nuninst_orig = nuninst_end
1451 self.all_selected += selected
1452 if transaction:
1453 transaction.commit()
1454 if self.options.check_consistency_level >= 2: 1454 ↛ 1458line 1454 didn't jump to line 1458 because the condition on line 1454 was always true
1455 target_suite.check_suite_source_pkg_consistency(
1456 "do_all after commit"
1457 )
1458 if not actions:
1459 if recurse:
1460 self.upgrade_me = extra
1461 else:
1462 selected_set = set(selected)
1463 self.upgrade_me = [
1464 x for x in self.upgrade_me if x not in selected_set
1465 ]
1466 else:
1467 output_logger.info("FAILED\n")
1468 if not transaction: 1468 ↛ 1472line 1468 didn't jump to line 1472 because the condition on line 1468 was never true
1469 # if we 'FAILED', but we cannot rollback, we will probably
1470 # leave a broken state behind
1471 # this should not happen
1472 raise AssertionError("do_all FAILED but no transaction to rollback")
1473 transaction.rollback()
1474 if self.options.check_consistency_level >= 2: 1474 ↛ 1351line 1474 didn't jump to line 1351
1475 target_suite.check_suite_source_pkg_consistency(
1476 "do_all after rollback"
1477 )
1479 output_logger.info("")
1481 def assert_nuninst_is_correct(self) -> None:
1482 self.logger.info("> Update complete - Verifying non-installability counters")
1484 cached_nuninst = self.nuninst_orig
1485 self._inst_tester.compute_installability()
1486 computed_nuninst = compile_nuninst(
1487 self.suite_info.target_suite,
1488 self.options.architectures,
1489 self.options.nobreakall_arches,
1490 )
1491 if cached_nuninst != computed_nuninst: # pragma: no cover
1492 only_on_break_archs = True
1493 msg_l = [
1494 "==================== NUNINST OUT OF SYNC ========================="
1495 ]
1496 for arch in self.options.architectures:
1497 expected_nuninst = set(cached_nuninst[arch])
1498 actual_nuninst = set(computed_nuninst[arch])
1499 false_negatives = actual_nuninst - expected_nuninst
1500 false_positives = expected_nuninst - actual_nuninst
1501 # Britney does not quite work correctly with
1502 # break/fucked arches, so ignore issues there for now.
1503 if (
1504 false_negatives or false_positives
1505 ) and arch not in self.options.break_arches:
1506 only_on_break_archs = False
1507 if false_negatives:
1508 msg_l.append(f" {arch} - unnoticed nuninst: {str(false_negatives)}")
1509 if false_positives:
1510 msg_l.append(f" {arch} - invalid nuninst: {str(false_positives)}")
1511 if false_negatives or false_positives:
1512 msg_l.append(
1513 f" {arch} - actual nuninst: {str(sorted(actual_nuninst))}"
1514 )
1515 msg_l.append(msg_l[0])
1516 for msg in msg_l:
1517 if only_on_break_archs:
1518 self.logger.warning(msg)
1519 else:
1520 self.logger.error(msg)
1521 if not only_on_break_archs:
1522 raise AssertionError("NUNINST OUT OF SYNC")
1523 else:
1524 self.logger.warning("Nuninst is out of sync on some break arches")
1526 self.logger.info("> All non-installability counters are ok")
1528 def upgrade_testing(self) -> None:
1529 """Upgrade testing using the packages from the source suites
1531 This method tries to upgrade testing using the packages from the
1532 source suites.
1533 Before running the do_all method, it tries the easy and force-hint
1534 commands.
1535 """
1537 output_logger = self.output_logger
1538 self.logger.info("Starting the upgrade test")
1539 output_logger.info(
1540 "Generated on: %s",
1541 time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time())),
1542 )
1543 output_logger.info("Arch order is: %s", ", ".join(self.options.architectures))
1545 if not self.options.actions: 1545 ↛ 1556line 1545 didn't jump to line 1556 because the condition on line 1545 was always true
1546 # process `easy' hints
1547 for x in self.hints["easy"]:
1548 self.do_hint("easy", x.user, x.packages)
1550 # process `force-hint' hints
1551 for x in self.hints["force-hint"]:
1552 self.do_hint("force-hint", x.user, x.packages)
1554 # run the first round of the upgrade
1555 # - do separate runs for break arches
1556 allpackages = []
1557 normpackages = self.upgrade_me[:]
1558 archpackages = {}
1559 for a in self.options.break_arches:
1560 archpackages[a] = [p for p in normpackages if p.architecture == a]
1561 normpackages = [p for p in normpackages if p.architecture != a]
1562 self.upgrade_me = normpackages
1563 output_logger.info("info: main run")
1564 self.do_all()
1565 allpackages += self.upgrade_me
1566 for a in self.options.break_arches:
1567 backup = self.options.break_arches
1568 self.options.break_arches = " ".join(
1569 x for x in self.options.break_arches if x != a
1570 )
1571 self.upgrade_me = archpackages[a]
1572 output_logger.info("info: broken arch run for %s", a)
1573 self.do_all()
1574 allpackages += self.upgrade_me
1575 self.options.break_arches = backup
1576 self.upgrade_me = allpackages
1578 if self.options.actions: 1578 ↛ 1579line 1578 didn't jump to line 1579 because the condition on line 1578 was never true
1579 self.printuninstchange()
1580 return
1582 # process `hint' hints
1583 hintcnt = 0
1584 for x in self.hints["hint"][:50]:
1585 if hintcnt > 50: 1585 ↛ 1586line 1585 didn't jump to line 1586 because the condition on line 1585 was never true
1586 output_logger.info("Skipping remaining hints...")
1587 break
1588 if self.do_hint("hint", x.user, x.packages): 1588 ↛ 1584line 1588 didn't jump to line 1584 because the condition on line 1588 was always true
1589 hintcnt += 1
1591 # run the auto hinter
1592 self.run_auto_hinter()
1594 if getattr(self.options, "remove_obsolete", "yes") == "yes":
1595 # obsolete source packages
1596 # a package is obsolete if none of the binary packages in testing
1597 # are built by it
1598 self.logger.info(
1599 "> Removing obsolete source packages from the target suite"
1600 )
1601 # local copies for performance
1602 target_suite = self.suite_info.target_suite
1603 sources_t = target_suite.sources
1604 binaries_t = target_suite.binaries
1605 mi_factory = self._migration_item_factory
1606 used = {
1607 binaries_t[arch][binary].source
1608 for arch in binaries_t
1609 for binary in binaries_t[arch]
1610 if not binary.endswith("-faux-build-depends")
1611 }
1612 removals = [
1613 mi_factory.parse_item(
1614 f"-{source}/{sources_t[source].version}", auto_correct=False
1615 )
1616 for source in sources_t
1617 if source not in used
1618 ]
1619 if removals:
1620 output_logger.info(
1621 "Removing obsolete source packages from the target suite (%d):",
1622 len(removals),
1623 )
1624 self.do_all(actions=removals)
1626 # smooth updates
1627 removals = old_libraries(
1628 self._migration_item_factory, self.suite_info, self.options.outofsync_arches
1629 )
1630 if removals:
1631 output_logger.info(
1632 "Removing packages left in the target suite (e.g. smooth updates or cruft)"
1633 )
1634 log_and_format_old_libraries(self.output_logger, removals)
1635 self.do_all(actions=removals)
1636 removals = old_libraries(
1637 self._migration_item_factory,
1638 self.suite_info,
1639 self.options.outofsync_arches,
1640 )
1642 output_logger.info(
1643 "List of old libraries in the target suite (%d):", len(removals)
1644 )
1645 log_and_format_old_libraries(self.output_logger, removals)
1647 self.printuninstchange()
1648 if self.options.check_consistency_level >= 1: 1648 ↛ 1654line 1648 didn't jump to line 1654 because the condition on line 1648 was always true
1649 target_suite = self.suite_info.target_suite
1650 self.assert_nuninst_is_correct()
1651 target_suite.check_suite_source_pkg_consistency("end")
1653 # output files
1654 if self.options.heidi_output and not self.options.dry_run: 1654 ↛ 1668line 1654 didn't jump to line 1668 because the condition on line 1654 was always true
1655 target_suite = self.suite_info.target_suite
1657 # write HeidiResult
1658 self.logger.info("Writing Heidi results to %s", self.options.heidi_output)
1659 write_heidi(
1660 self.options.heidi_output,
1661 target_suite,
1662 outofsync_arches=self.options.outofsync_arches,
1663 )
1665 self.logger.info("Writing delta to %s", self.options.heidi_delta_output)
1666 write_heidi_delta(self.options.heidi_delta_output, self.all_selected)
1668 self.logger.info("Test completed!")
1670 def printuninstchange(self) -> None:
1671 self.logger.info("Checking for newly uninstallable packages")
1672 uninst = newly_uninst(self.nuninst_orig_save, self.nuninst_orig)
1674 if uninst:
1675 self.output_logger.info("")
1676 self.output_logger.info(
1677 "Newly uninstallable packages in the target suite (arch:all on BREAKALL_ARCHES not shown)"
1678 )
1679 format_and_log_uninst(
1680 self.output_logger,
1681 self.options.architectures,
1682 uninst,
1683 loglevel=logging.WARNING,
1684 )
1686 def hint_tester(self) -> None:
1687 """Run a command line interface to test hints
1689 This method provides a command line interface for the release team to
1690 try hints and evaluate the results.
1691 """
1692 import readline
1694 from britney2.completer import Completer
1696 histfile = os.path.expanduser("~/.britney2_history")
1697 if os.path.exists(histfile):
1698 readline.read_history_file(histfile)
1700 readline.parse_and_bind("tab: complete")
1701 readline.set_completer(Completer(self).completer)
1702 # Package names can contain "-" and we use "/" in our presentation of them as well,
1703 # so ensure readline does not split on these characters.
1704 readline.set_completer_delims(
1705 readline.get_completer_delims().replace("-", "").replace("/", "")
1706 )
1708 known_hints = self._hint_parser.registered_hint_names
1710 print("Britney hint tester")
1711 print()
1712 print(
1713 "Besides inputting known britney hints, the follow commands are also available"
1714 )
1715 print(" * quit/exit - terminates the shell")
1716 print(
1717 " * python-console - jump into an interactive python shell (with the current loaded dataset)"
1718 )
1719 print()
1721 while True:
1722 # read the command from the command line
1723 try:
1724 user_input = input("britney> ").split()
1725 except EOFError:
1726 print("")
1727 break
1728 except KeyboardInterrupt:
1729 print("")
1730 continue
1731 match user_input:
1732 case ("quit" | "exit", *_):
1733 # quit the hint tester
1734 break
1735 case ("python-console", *_):
1736 try:
1737 import britney2.console
1738 except ImportError as e:
1739 print(f"Failed to import britney.console module: {e!r}")
1740 continue
1741 britney2.console.run_python_console(self)
1742 print("Returning to the britney hint-tester console")
1743 # run a hint
1744 case ("easy" | "hint" | "force-hint" as choice, *items):
1745 mi_factory = self._migration_item_factory
1746 try:
1747 self.do_hint(
1748 choice,
1749 "hint-tester",
1750 list(mi_factory.parse_items(items)),
1751 )
1752 self.printuninstchange()
1753 except KeyboardInterrupt:
1754 continue
1755 case (str() as hint, *_) if hint in known_hints:
1756 self._hint_parser.parse_hints(
1757 "hint-tester", self.HINTS_ALL, "<stdin>", [" ".join(user_input)]
1758 )
1759 self.write_excuses()
1761 try:
1762 readline.write_history_file(histfile)
1763 except OSError as e:
1764 self.logger.warning("Could not write %s: %s", histfile, e)
1766 def do_hint(self, hinttype: str, who: str, pkgvers: list[MigrationItem]) -> bool:
1767 """Process hints
1769 This method process `easy`, `hint` and `force-hint` hints. If the
1770 requested version is not in the relevant source suite, then the hint
1771 is skipped.
1772 """
1774 output_logger = self.output_logger
1776 self.logger.info("> Processing '%s' hint from %s", hinttype, who)
1777 output_logger.info(
1778 "Trying %s from %s: %s",
1779 hinttype,
1780 who,
1781 " ".join(f"{x.uvname}/{x.version}" for x in pkgvers),
1782 )
1784 issues = []
1785 # loop on the requested packages and versions
1786 for pkg in pkgvers:
1787 # skip removal requests
1788 if pkg.is_removal:
1789 continue
1791 suite = pkg.suite
1793 assert pkg.version is not None
1794 source = suite.sources.get(pkg.package)
1795 if source is None: 1795 ↛ 1796line 1795 didn't jump to line 1796 because the condition on line 1795 was never true
1796 issues.append(f"Source {pkg.package} has no version in {suite.name}")
1797 elif apt_pkg.version_compare(source.version, pkg.version) != 0: 1797 ↛ 1798line 1797 didn't jump to line 1798 because the condition on line 1797 was never true
1798 issues.append(
1799 f"Version mismatch, {pkg.package} {pkg.version} != {source.version}"
1800 )
1801 if issues: 1801 ↛ 1802line 1801 didn't jump to line 1802 because the condition on line 1801 was never true
1802 output_logger.warning("%s: Not using hint", ", ".join(issues))
1803 return False
1805 self.do_all(hinttype, pkgvers)
1806 return True
1808 def get_auto_hinter_hints(
1809 self, upgrade_me: list[MigrationItem]
1810 ) -> tuple[list[frozenset[MigrationItem]], list[frozenset[MigrationItem]]]:
1811 """Auto-generate "easy" hints.
1813 This method attempts to generate "easy" hints for sets of packages which
1814 must migrate together. Beginning with a package which does not depend on
1815 any other package (in terms of excuses), a list of dependencies and
1816 reverse dependencies is recursively created.
1818 Once all such lists have been generated, any which are subsets of other
1819 lists are ignored in favour of the larger lists. The remaining lists are
1820 then attempted in turn as "easy" hints.
1822 We also try to auto hint circular dependencies analyzing the update
1823 excuses relationships. If they build a circular dependency, which we already
1824 know as not-working with the standard do_all algorithm, try to `easy` them.
1825 """
1826 self.logger.info("> Processing hints from the auto hinter")
1828 sources_t = self.suite_info.target_suite.sources
1829 excuses = self.excuses
1831 def excuse_still_valid(excuse: "Excuse") -> bool:
1832 source = excuse.source
1833 assert isinstance(excuse.item, MigrationItem)
1834 arch = excuse.item.architecture
1835 # TODO for binNMUs, this check is always ok, even if the item
1836 # migrated already
1837 valid = (
1838 arch != "source"
1839 or source not in sources_t
1840 or sources_t[source].version != excuse.ver[1]
1841 )
1842 # TODO migrated items should be removed from upgrade_me, so this
1843 # should not happen
1844 if not valid: 1844 ↛ 1845line 1844 didn't jump to line 1845 because the condition on line 1844 was never true
1845 raise AssertionError(f"excuse no longer valid {excuse.item}")
1846 return valid
1848 # consider only excuses which are valid candidates and still relevant.
1849 valid_excuses = frozenset(
1850 e.name
1851 for e in excuses.values()
1852 if e.item in upgrade_me and excuse_still_valid(e)
1853 )
1854 excuses_deps = {
1855 name: valid_excuses.intersection(excuse.get_deps())
1856 for name, excuse in excuses.items()
1857 if name in valid_excuses
1858 }
1859 excuses_rdeps = defaultdict(set)
1860 for name, deps in excuses_deps.items():
1861 for dep in deps:
1862 excuses_rdeps[dep].add(name)
1864 # loop on them
1865 candidates = []
1866 mincands = []
1867 seen_hints = set()
1868 for e in valid_excuses:
1869 excuse = excuses[e]
1870 if not excuse.get_deps():
1871 assert isinstance(excuse.item, MigrationItem)
1872 items = [excuse.item]
1873 orig_size = 1
1874 looped = False
1875 seen_items = set()
1876 seen_items.update(items)
1878 for item in items:
1879 assert isinstance(item, MigrationItem)
1880 # excuses which depend on "item" or are depended on by it
1881 new_items = {
1882 excuses[x].item
1883 for x in chain(
1884 excuses_deps[item.name], excuses_rdeps[item.name]
1885 )
1886 }
1887 new_items -= seen_items
1888 items.extend(new_items)
1889 seen_items.update(new_items)
1891 if not looped and len(items) > 1:
1892 orig_size = len(items)
1893 h = frozenset(seen_items)
1894 if h not in seen_hints: 1894 ↛ 1897line 1894 didn't jump to line 1897 because the condition on line 1894 was always true
1895 mincands.append(h)
1896 seen_hints.add(h)
1897 looped = True
1898 if len(items) != orig_size: 1898 ↛ 1899line 1898 didn't jump to line 1899 because the condition on line 1898 was never true
1899 h = frozenset(seen_items)
1900 if h != mincands[-1] and h not in seen_hints:
1901 candidates.append(h)
1902 seen_hints.add(h)
1903 return (candidates, mincands)
1905 def run_auto_hinter(self) -> None:
1906 for lst in self.get_auto_hinter_hints(self.upgrade_me):
1907 for hint in lst:
1908 self.do_hint("easy", "autohinter", sorted(hint))
1910 def nuninst_arch_report(self, nuninst: dict[str, set[str]], arch: str) -> None:
1911 """Print a report of uninstallable packages for one architecture."""
1912 all = defaultdict(set)
1913 binaries_t = self.suite_info.target_suite.binaries
1914 for p in nuninst[arch]:
1915 pkg = binaries_t[arch][p]
1916 all[(pkg.source, pkg.source_version)].add(p)
1918 print(f"* {arch}")
1920 for (src, ver), pkgs in sorted(all.items()):
1921 print(" {} ({}): {}".format(src, ver, " ".join(sorted(pkgs))))
1923 print()
1925 def _remove_archall_faux_packages(self) -> None:
1926 """Remove faux packages added for the excuses phase
1928 To prevent binary packages from going missing while they are listed by
1929 their source package we add bin:faux packages during reading in the
1930 Sources. They are used during the excuses phase to prevent packages
1931 from becoming candidates. However, they interfere in complex ways
1932 during the installability phase, so instead of having all code during
1933 migration be aware of this excuses phase implementation detail, let's
1934 remove them again.
1936 """
1937 if not self.options.archall_inconsistency_allowed:
1938 all_binaries = self.all_binaries
1939 faux_a = {x for x in all_binaries.keys() if x.architecture == "faux"}
1940 for pkg_a in faux_a:
1941 del all_binaries[pkg_a]
1943 for suite in self.suite_info._suites.values():
1944 for arch in suite.binaries.keys():
1945 binaries = suite.binaries[arch]
1946 faux_b = {
1947 x for x in binaries if binaries[x].pkg_id.architecture == "faux"
1948 }
1949 for pkg_b in faux_b:
1950 del binaries[pkg_b]
1951 sources = suite.sources
1952 for src in sources.keys():
1953 faux_s = {
1954 x for x in sources[src].binaries if x.architecture == "faux"
1955 }
1956 sources[src].binaries -= faux_s
1958 def main(self) -> None:
1959 """Main method
1961 This is the entry point for the class: it includes the list of calls
1962 for the member methods which will produce the output files.
1963 """
1964 # if running in --print-uninst mode, quit
1965 if self.options.print_uninst: 1965 ↛ 1966line 1965 didn't jump to line 1966 because the condition on line 1965 was never true
1966 return
1967 # if no actions are provided, build the excuses and sort them
1968 elif not self.options.actions: 1968 ↛ 1972line 1968 didn't jump to line 1972 because the condition on line 1968 was always true
1969 self.write_excuses()
1970 # otherwise, use the actions provided by the command line
1971 else:
1972 self.upgrade_me = self.options.actions.split()
1974 self._remove_archall_faux_packages()
1976 if self.options.compute_migrations or self.options.hint_tester:
1977 if self.options.dry_run: 1977 ↛ 1978line 1977 didn't jump to line 1978 because the condition on line 1977 was never true
1978 self.logger.info(
1979 "Upgrade output not (also) written to a separate file"
1980 " as this is a dry-run."
1981 )
1982 elif hasattr(self.options, "upgrade_output"): 1982 ↛ 1992line 1982 didn't jump to line 1992 because the condition on line 1982 was always true
1983 upgrade_output = getattr(self.options, "upgrade_output")
1984 file_handler = logging.FileHandler(
1985 upgrade_output, mode="w", encoding="utf-8"
1986 )
1987 output_formatter = logging.Formatter("%(message)s")
1988 file_handler.setFormatter(output_formatter)
1989 self.output_logger.addHandler(file_handler)
1990 self.logger.info("Logging upgrade output to %s", upgrade_output)
1991 else:
1992 self.logger.info(
1993 "Upgrade output not (also) written to a separate file"
1994 " as the UPGRADE_OUTPUT configuration is not provided."
1995 )
1997 # run the hint tester
1998 if self.options.hint_tester: 1998 ↛ 1999line 1998 didn't jump to line 1999 because the condition on line 1998 was never true
1999 self.hint_tester()
2000 # run the upgrade test
2001 else:
2002 self.upgrade_testing()
2004 self.logger.info("> Stats from the installability tester")
2005 for stat in self._inst_tester.stats.stats():
2006 self.logger.info("> %s", stat)
2007 else:
2008 self.logger.info("Migration computation skipped as requested.")
2009 if not self.options.dry_run: 2009 ↛ 2011line 2009 didn't jump to line 2011 because the condition on line 2009 was always true
2010 self._policy_engine.save_state(self)
2011 logging.shutdown()
2014if __name__ == "__main__": 2014 ↛ 2015line 2014 didn't jump to line 2015 because the condition on line 2014 was never true
2015 Britney().main()