Coverage for britney2/utils.py: 92%
480 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +0000
1# Refactored parts from britney.py, which is/was:
2# Copyright (C) 2001-2008 Anthony Towns <ajt@debian.org>
3# Andreas Barth <aba@debian.org>
4# Fabio Tranchitella <kobold@debian.org>
5# Copyright (C) 2010-2012 Adam D. Barratt <adsb@debian.org>
6# Copyright (C) 2012 Niels Thykier <niels@thykier.net>
7#
8# New portions
9# Copyright (C) 2013 Adam D. Barratt <adsb@debian.org>
11# This program is free software; you can redistribute it and/or modify
12# it under the terms of the GNU General Public License as published by
13# the Free Software Foundation; either version 2 of the License, or
14# (at your option) any later version.
16# This program is distributed in the hope that it will be useful,
17# but WITHOUT ANY WARRANTY; without even the implied warranty of
18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19# GNU General Public License for more details.
22import errno
23import logging
24import optparse
25import os
26import sys
27import time
28from collections import defaultdict
29from collections.abc import (
30 Container,
31 Iterable,
32 Iterator,
33 Mapping,
34 MutableSet,
35)
36from datetime import UTC, datetime
37from enum import Enum, StrEnum
38from functools import partial
39from itertools import chain, filterfalse
40from typing import (
41 IO,
42 TYPE_CHECKING,
43 Any,
44 Literal,
45 Protocol,
46 TypeVar,
47 Union,
48 cast,
49 overload,
50)
52import apt_pkg
53import yaml
54from more_itertools import iter_except
56from britney2 import (
57 BinaryPackage,
58 BinaryPackageId,
59 MultiArch,
60 PackageId,
61 SourcePackage,
62 Suite,
63 SuiteClass,
64 Suites,
65 TargetSuite,
66)
67from britney2.excuse import Excuse
68from britney2.excusedeps import DependencyState, ImpossibleDependencyState
69from britney2.policies import PolicyVerdict
71if TYPE_CHECKING: 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was never true
73 from _typeshed import SupportsRichComparisonT
74 from apt_pkg import TagSection
76 from .hints import HintCollection
77 from .installability.universe import BinaryPackageUniverse
78 from .migrationitem import MigrationItem, MigrationItemFactory
80_T = TypeVar("_T")
83class MigrationConstraintException(Exception):
84 pass
87@overload
88def ifilter_except( 88 ↛ exitline 88 didn't jump to the function exit
89 container: Container[_T], iterable: Literal[None] = None
90) -> "partial[filterfalse[_T]]": ...
93@overload
94def ifilter_except( 94 ↛ exitline 94 didn't jump to the function exit
95 container: Container[_T], iterable: Iterable[_T]
96) -> "filterfalse[_T]": ...
99def ifilter_except(
100 container: Container[_T], iterable: Iterable[_T] | None = None
101) -> Union["filterfalse[_T]", "partial[filterfalse[_T]]"]:
102 """Filter out elements in container
104 If given an iterable it returns a filtered iterator, otherwise it
105 returns a function to generate filtered iterators. The latter is
106 useful if the same filter has to be (re-)used on multiple
107 iterators that are not known on beforehand.
108 """
109 if iterable is not None: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 return filterfalse(container.__contains__, iterable)
111 return cast(
112 "partial[filterfalse[_T]]", partial(filterfalse, container.__contains__)
113 )
116@overload
117def ifilter_only(container: Container[_T], iterable: Iterable[_T]) -> "filter[_T]": ... 117 ↛ exitline 117 didn't return from function 'ifilter_only' because
120@overload
121def ifilter_only(container: Container[_T], iterable: None) -> "partial[filter[_T]]": ... 121 ↛ exitline 121 didn't return from function 'ifilter_only' because
124def ifilter_only(
125 container: Container[_T], iterable: Iterable[_T] | None = None
126) -> Union["filter[_T]", "partial[filter[_T]]"]:
127 """Filter out elements in which are not in container
129 If given an iterable it returns a filtered iterator, otherwise it
130 returns a function to generate filtered iterators. The latter is
131 useful if the same filter has to be (re-)used on multiple
132 iterators that are not known on beforehand.
133 """
134 if iterable is not None: 134 ↛ 136line 134 didn't jump to line 136 because the condition on line 134 was always true
135 return filter(container.__contains__, iterable)
136 return partial(filter, container.__contains__)
139def log_and_format_old_libraries(
140 logger: logging.Logger, libs: list["MigrationItem"]
141) -> None:
142 """Format and log old libraries in a table (no header)"""
143 libraries: dict[str, list[str]] = {}
144 for i in libs:
145 pkg = i.package
146 if pkg in libraries:
147 libraries[pkg].append(i.architecture)
148 else:
149 libraries[pkg] = [i.architecture]
151 for lib in sorted(libraries):
152 logger.info(" %s: %s", lib, " ".join(libraries[lib]))
155def compute_reverse_tree(
156 pkg_universe: "BinaryPackageUniverse", affected: set[BinaryPackageId]
157) -> None:
158 """Calculate the full dependency tree for a set of packages
160 This method returns the full dependency tree for a given set of
161 packages. The first argument is an instance of the BinaryPackageUniverse
162 and the second argument are a set of BinaryPackageId.
164 The set of affected packages will be updated in place and must
165 therefore be mutable.
166 """
167 remain = list(affected)
168 while remain:
169 pkg_id = remain.pop()
170 new_pkg_ids = pkg_universe.reverse_dependencies_of(pkg_id) - affected
171 affected.update(new_pkg_ids)
172 remain.extend(new_pkg_ids)
175def add_transitive_dependencies_flatten(
176 pkg_universe: "BinaryPackageUniverse", initial_set: MutableSet[BinaryPackageId]
177) -> None:
178 """Find and include all transitive dependencies
180 This method updates the initial_set parameter to include all transitive
181 dependencies. The first argument is an instance of the BinaryPackageUniverse
182 and the second argument are a set of BinaryPackageId.
184 The set of initial packages will be updated in place and must
185 therefore be mutable.
186 """
187 remain = list(initial_set)
188 while remain:
189 pkg_id = remain.pop()
190 new_pkg_ids = {
191 x
192 for x in chain.from_iterable(pkg_universe.dependencies_of(pkg_id))
193 if x not in initial_set
194 }
195 initial_set |= new_pkg_ids
196 remain.extend(new_pkg_ids)
199def write_nuninst(filename: str, nuninst: dict[str, set[str]]) -> None:
200 """Write the non-installable report
202 Write the non-installable report derived from "nuninst" to the
203 file denoted by "filename".
204 """
205 with open(filename, "w", encoding="utf-8") as f:
206 # Having two fields with (almost) identical dates seems a bit
207 # redundant.
208 f.write(
209 "Built on: "
210 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
211 + "\n"
212 )
213 f.write(
214 "Last update: "
215 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
216 + "\n\n"
217 )
218 for k in nuninst:
219 f.write("{}: {}\n".format(k, " ".join(nuninst[k])))
222def read_nuninst(filename: str, architectures: set[str]) -> dict[str, set[str]]:
223 """Read the non-installable report
225 Read the non-installable report from the file denoted by
226 "filename" and return it. Only architectures in "architectures"
227 will be included in the report.
228 """
229 nuninst: dict[str, set[str]] = {}
230 with open(filename, encoding="utf-8") as f:
231 for r in f:
232 if ":" not in r:
233 continue
234 arch, packages = r.strip().split(":", 1)
235 if arch.split("+", 1)[0] in architectures:
236 nuninst[arch] = set(packages.split())
237 return nuninst
240def newly_uninst(
241 nuold: dict[str, set[str]], nunew: dict[str, set[str]]
242) -> dict[str, list[str]]:
243 """Return a nuninst statistic with only new uninstallable packages
245 This method subtracts the uninstallable packages of the statistic
246 "nunew" from the statistic "nuold".
248 It returns a dictionary with the architectures as keys and the list
249 of uninstallable packages as values. If there are no regressions
250 on a given architecture, then the architecture will be omitted in
251 the result. Accordingly, if none of the architectures have
252 regressions an empty directory is returned.
253 """
254 res: dict[str, list[str]] = {}
255 for arch in ifilter_only(nunew, nuold):
256 arch_nuninst = [x for x in nunew[arch] if x not in nuold[arch]]
257 # Leave res empty if there are no newly uninst packages
258 if arch_nuninst:
259 res[arch] = arch_nuninst
260 return res
263def format_and_log_uninst(
264 logger: logging.Logger,
265 architectures: Iterable[str],
266 nuninst: Mapping[str, Iterable[str]],
267 *,
268 loglevel: int = logging.INFO,
269) -> None:
270 """Emits the uninstallable packages to the log
272 An example of the output string is:
273 * i386: broken-pkg1, broken-pkg2
275 Note that if there is no uninstallable packages, then nothing is emitted.
276 """
277 for arch in architectures:
278 if arch in nuninst and nuninst[arch]:
279 msg = " * {}: {}".format(arch, ", ".join(sorted(nuninst[arch])))
280 logger.log(loglevel, msg)
283class Sorted(Protocol):
284 def __call__( 284 ↛ exitline 284 didn't jump to the function exit
285 self,
286 iterable: Iterable["SupportsRichComparisonT"],
287 /,
288 *,
289 key: None = None,
290 reverse: bool = False,
291 ) -> list["SupportsRichComparisonT"]: ...
294def write_heidi(
295 filename: str,
296 target_suite: TargetSuite,
297 *,
298 outofsync_arches: frozenset[str] = frozenset(),
299 sorted: Sorted = sorted,
300) -> None:
301 """Write the output HeidiResult
303 This method write the output for Heidi, which contains all the
304 binary packages and the source packages in the form:
306 <pkg-name> <pkg-version> <pkg-architecture> <pkg-section>
307 <src-name> <src-version> source <src-section>
309 The file is written as "filename" using the sources and packages
310 from the "target_suite" parameter.
312 outofsync_arches: If given, it is a set of architectures marked
313 as "out of sync". The output file may exclude some out of date
314 arch:all packages for those architectures to reduce the noise.
316 The "X=X" parameters are optimizations to avoid "load global" in
317 the loops.
318 """
319 sources_t = target_suite.sources
320 packages_t = target_suite.binaries
322 with open(filename, "w", encoding="ascii") as f:
324 # write binary packages
325 for arch in sorted(packages_t):
326 binaries = packages_t[arch]
327 for pkg_name in sorted(binaries):
328 pkg = binaries[pkg_name]
329 pkgv = pkg.version
330 pkgarch = pkg.architecture or "all"
331 pkgsec = pkg.section or "faux"
332 if pkgsec == "faux" or pkgsec.endswith("/faux"):
333 # Faux package; not really a part of testing
334 continue
335 if ( 335 ↛ 347line 335 didn't jump to line 347
336 pkg.source_version
337 and pkgarch == "all"
338 and pkg.source_version != sources_t[pkg.source].version
339 and arch in outofsync_arches
340 ):
341 # when architectures are marked as "outofsync", their binary
342 # versions may be lower than those of the associated
343 # source package in testing. the binary package list for
344 # such architectures will include arch:all packages
345 # matching those older versions, but we only want the
346 # newer arch:all in testing
347 continue
348 f.write(f"{pkg_name} {pkgv} {pkgarch} {pkgsec}\n")
350 # write sources
351 for src_name in sorted(sources_t):
352 src = sources_t[src_name]
353 srcv = src.version
354 srcsec = src.section or "unknown"
355 if srcsec == "faux" or srcsec.endswith("/faux"):
356 # Faux package; not really a part of testing
357 continue
358 f.write(f"{src_name} {srcv} source {srcsec}\n")
361def write_heidi_delta(filename: str, all_selected: list["MigrationItem"]) -> None:
362 """Write the output delta
364 This method writes the packages to be upgraded, in the form:
365 <src-name> <src-version>
366 or (if the source is to be removed):
367 -<src-name> <src-version>
369 The order corresponds to that shown in update_output.
370 """
371 with open(filename, "w", encoding="ascii") as fd:
373 fd.write("#HeidiDelta\n")
375 for item in all_selected:
376 prefix = ""
378 if item.is_removal:
379 prefix = "-"
381 if item.architecture == "source":
382 fd.write(f"{prefix}{item.package} {item.version}\n")
383 else:
384 fd.write(f"{prefix}{item.package} {item.version} {item.architecture}\n")
387class Opener(Protocol):
388 def __call__( 388 ↛ exitline 388 didn't jump to the function exit
389 self, file: str, mode: Literal["wt"], encoding: Literal["utf-8"]
390 ) -> IO[Any]: ...
393class ExcusesOutputFormat(Enum):
394 YAML = 0
395 LEGACY_HTML = 1
398def write_excuses(
399 excuses: dict[str, "Excuse"] | dict[PackageId, "Excuse"],
400 dest_file: str,
401 output_format: ExcusesOutputFormat = ExcusesOutputFormat.YAML,
402) -> None:
403 """Write the excuses to dest_file
405 Writes a list of excuses in a specified output_format to the
406 path denoted by dest_file. The output_format can either be "yaml"
407 or "legacy-html".
408 """
409 excuselist = sorted(excuses.values(), key=lambda x: x.sortkey())
410 if output_format is ExcusesOutputFormat.YAML:
411 # use custom representer to avoid creation of the full list with all excuse data before starting the serialization
412 def represent_yaml_excuse(dumper: yaml.Dumper, data: Excuse) -> yaml.Node:
413 return dumper.represent_data(data.excusedata(excuses))
415 yaml.add_representer(Excuse, represent_yaml_excuse)
416 yaml.Dumper.add_multi_representer(
417 StrEnum, yaml.representer.Representer.represent_str
418 )
420 os.makedirs(os.path.dirname(dest_file), exist_ok=True)
421 opener: Opener = open # type: ignore[assignment]
422 if dest_file.endswith(".xz"): 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 import lzma
425 opener = lzma.open # type: ignore[assignment]
426 elif dest_file.endswith(".gz"): 426 ↛ 427line 426 didn't jump to line 427 because the condition on line 426 was never true
427 import gzip
429 opener = gzip.open # type: ignore[assignment]
430 with opener(f"{dest_file}.new", "wt", encoding="utf-8") as f:
431 excusesdata = {
432 "sources": excuselist,
433 "generated-date": datetime.now(UTC),
434 }
435 yaml.dump(
436 excusesdata, stream=f, default_flow_style=False, allow_unicode=True
437 )
438 os.replace(f"{dest_file}.new", dest_file)
439 elif output_format is ExcusesOutputFormat.LEGACY_HTML:
440 with open(f"{dest_file}.new", "w", encoding="utf-8") as f:
441 f.write(
442 '<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/REC-html40/strict.dtd">\n'
443 )
444 f.write("<html><head><title>excuses...</title>")
445 f.write(
446 '<meta http-equiv="Content-Type" content="text/html;charset=utf-8"></head><body>\n'
447 )
448 f.write(
449 "<p>Generated: "
450 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
451 + "</p>\n"
452 )
453 f.write("<ul>\n")
454 for e in excuselist:
455 f.write("<li>%s" % e.html(excuses))
456 f.write("</ul></body></html>\n")
457 os.replace(f"{dest_file}.new", dest_file)
458 else: # pragma: no cover
459 raise ValueError('Output format must be either "YAML or "LEGACY_HTML"')
462def old_libraries(
463 mi_factory: "MigrationItemFactory",
464 suite_info: Suites,
465 outofsync_arches: Iterable[str] = frozenset(),
466) -> list["MigrationItem"]:
467 """Detect old libraries left in the target suite for smooth transitions
469 This method detects old libraries which are in the target suite but no
470 longer built from the source package: they are still there because
471 other packages still depend on them, but they should be removed as
472 soon as possible.
474 For "outofsync" architectures, outdated binaries are allowed to be in
475 the target suite, so they are only added to the removal list if they
476 are no longer in the (primary) source suite.
477 """
478 sources_t = suite_info.target_suite.sources
479 binaries_t = suite_info.target_suite.binaries
480 binaries_s = suite_info.primary_source_suite.binaries
481 removals = []
482 for arch, binaries in binaries_t.items():
483 for pkg_name, pkg in binaries.items():
484 if sources_t[pkg.source].version != pkg.source_version and (
485 arch not in outofsync_arches or pkg_name not in binaries_s[arch]
486 ):
487 removals.append(mi_factory.generate_removal_for_cruft_item(pkg.pkg_id))
488 return removals
491def is_nuninst_asgood_generous(
492 constraints: dict[str, list[str]],
493 allow_uninst: dict[str, set[str | None]],
494 architectures: list[str],
495 old: dict[str, set[str]],
496 new: dict[str, set[str]],
497 break_arches: set[str] = cast(set[str], frozenset()),
498) -> bool:
499 """Compares the nuninst counters and constraints to see if they improved
501 Given a list of architectures, the previous and the current nuninst
502 counters, this function determines if the current nuninst counter
503 is better than the previous one. Optionally it also accepts a set
504 of "break_arches", the nuninst counter for any architecture listed
505 in this set are completely ignored.
507 If the nuninst counters are equal or better, then the constraints
508 are checked for regressions (ignoring break_arches).
510 Returns True if the new nuninst counter is better than the
511 previous and there are no constraint regressions (ignoring Break-archs).
512 Returns False otherwise.
514 """
515 diff = 0
516 for arch in architectures:
517 if arch in break_arches:
518 continue
519 diff = diff + (
520 len(new[arch] - allow_uninst[arch]) - len(old[arch] - allow_uninst[arch])
521 )
522 if diff > 0:
523 return False
524 must_be_installable = constraints["keep-installable"]
525 for arch in architectures:
526 if arch in break_arches:
527 continue
528 regression = new[arch] - old[arch]
529 if not regression.isdisjoint(must_be_installable): 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true
530 return False
531 return True
534def clone_nuninst(
535 nuninst: dict[str, set[str]],
536 *,
537 packages_s: dict[str, dict[str, BinaryPackage]] | None = None,
538 architectures: Iterable[str] | None = None,
539) -> dict[str, set[str]]:
540 """Completely or Selectively deep clone nuninst
542 Given nuninst table, the package table for a given suite and
543 a list of architectures, this function will clone the nuninst
544 table. Only the listed architectures will be deep cloned -
545 the rest will only be shallow cloned. When packages_s is given,
546 packages not listed in packages_s will be pruned from the clone
547 (if packages_s is omitted, the per architecture nuninst is cloned
548 as-is)
549 """
550 clone = nuninst.copy()
551 if architectures is None: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true
552 return clone
553 if packages_s is not None:
554 for arch in architectures:
555 clone[arch] = {x for x in nuninst[arch] if x in packages_s[arch]}
556 clone[arch + "+all"] = {
557 x for x in nuninst[arch + "+all"] if x in packages_s[arch]
558 }
559 else:
560 for arch in architectures:
561 clone[arch] = set(nuninst[arch])
562 clone[arch + "+all"] = set(nuninst[arch + "+all"])
563 return clone
566def test_installability(
567 target_suite: TargetSuite,
568 pkg_name: str,
569 pkg_id: BinaryPackageId,
570 broken: set[str],
571 nuninst_arch: set[str] | None,
572) -> None:
573 """Test for installability of a package on an architecture
575 (pkg_name, pkg_version, pkg_arch) is the package to check.
577 broken is the set of broken packages. If p changes
578 installability (e.g. goes from uninstallable to installable),
579 broken will be updated accordingly.
581 If nuninst_arch is not None then it also updated in the same
582 way as broken is.
583 """
584 if not target_suite.is_installable(pkg_id):
585 # if pkg_name not in broken: regression else: already broken
586 broken.add(pkg_name)
587 if nuninst_arch is not None:
588 nuninst_arch.add(pkg_name)
589 else:
590 # if pkg_name in broken: # improvement else: already not broken
591 broken.discard(pkg_name)
592 if nuninst_arch is not None:
593 nuninst_arch.discard(pkg_name)
596def check_installability(
597 target_suite: TargetSuite,
598 binaries: dict[str, dict[str, BinaryPackage]],
599 arch: str,
600 updates: set[BinaryPackageId],
601 check_archall: bool,
602 nuninst: dict[str, set[str]],
603) -> None:
604 broken = nuninst[arch + "+all"]
605 packages_t_a = binaries[arch]
607 for pkg_id in (x for x in updates if x.architecture == arch):
608 name, version, parch = pkg_id.package_name, pkg_id.version, pkg_id.architecture
609 pkgdata = packages_t_a.get(name)
610 if pkgdata is None:
611 continue
612 if version != pkgdata.version:
613 # Not the version in testing right now, ignore
614 continue
615 actual_arch = pkgdata.architecture
616 nuninst_arch = None
617 # only check arch:all packages if requested
618 if check_archall or actual_arch != "all":
619 nuninst_arch = nuninst[parch]
620 elif actual_arch == "all": 620 ↛ 622line 620 didn't jump to line 622 because the condition on line 620 was always true
621 nuninst[parch].discard(name)
622 test_installability(target_suite, name, pkg_id, broken, nuninst_arch)
625def possibly_compressed(
626 path: str, *, permitted_compressions: list[str] | None = None
627) -> str:
628 """Find and select a (possibly compressed) variant of a path
630 If the given path exists, it will be returned
632 :param path: The base path.
633 :param permitted_compressions: Alternative extensions to look for. Defaults to "gz" and "xz".
634 :return: The path given possibly with one of the permitted extensions.
635 :raises FileNotFoundError: if the path is not found
636 """
637 if os.path.exists(path): 637 ↛ 639line 637 didn't jump to line 639 because the condition on line 637 was always true
638 return path
639 if permitted_compressions is None:
640 permitted_compressions = ["gz", "xz"]
641 for ext in permitted_compressions:
642 cpath = f"{path}.{ext}"
643 if os.path.exists(cpath):
644 return cpath
645 raise FileNotFoundError(
646 errno.ENOENT, os.strerror(errno.ENOENT), path
647 ) # pragma: no cover
650def create_provides_map(
651 packages: dict[str, BinaryPackage],
652) -> dict[str, set[tuple[str, str]]]:
653 """Create a provides map from a map binary package names and their BinaryPackage objects
655 :param packages: A dict mapping binary package names to their BinaryPackage object
656 :return: A provides map
657 """
658 # create provides
659 provides = defaultdict(set)
661 for pkg, dpkg in packages.items():
662 if dpkg.provides is None:
663 continue
664 # register virtual packages and real packages that provide
665 # them
666 for provided_pkg, provided_version in dpkg.provides:
667 provides[provided_pkg].add((pkg, provided_version))
669 return provides
672def read_release_file(suite_dir: str) -> "TagSection[str]":
673 """Parses a given "Release" file
675 :param suite_dir: The directory to the suite
676 :return: A dict of the first (and only) paragraph in an Release file
677 """
678 release_file = os.path.join(suite_dir, "Release")
679 with open(release_file) as fd:
680 tag_file = iter(apt_pkg.TagFile(fd))
681 result = next(tag_file)
682 if next(tag_file, None) is not None: # pragma: no cover
683 raise TypeError(f"{release_file} has more than one paragraph")
684 return result
687def read_sources_file(
688 filename: str,
689 sources: dict[str, SourcePackage] | None = None,
690 add_faux: bool = True,
691 sources_target_suite: dict[str, SourcePackage] | None = None,
692) -> dict[str, SourcePackage]:
693 """Parse a single Sources file into a hash
695 Parse a single Sources file into a dict mapping a source package
696 name to a SourcePackage object. If there are multiple source
697 packages with the same version, then highest versioned source
698 package (that is not marked as "Extra-Source-Only") is the
699 version kept in the dict.
701 :param filename: Path to the Sources file. Can be compressed by any algorithm supported by apt_pkg.TagFile
702 :param sources: Optional dict to add the packages to. If given, this is also the value returned.
703 :param add_faux: Add a faux arch:all binary for each source that claims it has arch:all
704 :param sources_target_suite: SourcPackages loaded from the target suite for memory optimizations
705 :return: mapping from names to a source package
706 """
707 if sources is None:
708 sources = {}
709 if sources_target_suite is None:
710 sources_target_suite = {}
712 tag_file = apt_pkg.TagFile(filename)
713 get_field = tag_file.section.get
714 step = tag_file.step
715 intern = sys.intern
717 while step():
718 if get_field("Extra-Source-Only", "no") == "yes":
719 # Ignore sources only referenced by Built-Using
720 continue
721 # intern pkg for faster lookups in sources
722 pkg = intern(get_field("Package"))
723 ver = get_field("Version")
724 # There may be multiple versions of the source package
725 # (in unstable) if some architectures have out-of-date
726 # binaries. We only ever consider the source with the
727 # largest version for migration.
728 if (other_source := sources.get(pkg)) is not None and apt_pkg.version_compare(
729 other_source.version, ver
730 ) > 0:
731 continue
732 maint = get_field("Maintainer")
733 if maint is not None: 733 ↛ 735line 733 didn't jump to line 735 because the condition on line 733 was always true
734 maint = maint.strip()
735 section = get_field("Section").strip()
736 build_deps_arch = (
737 ", ".join(
738 x
739 for x in (get_field("Build-Depends"), get_field("Build-Depends-Arch"))
740 if x is not None
741 )
742 or None
743 )
744 build_deps_indep = get_field("Build-Depends-Indep")
746 # Adding arch:all packages to the list of binaries already to be able
747 # to check for them later. Helps mitigate bug 887060 and is the
748 # (partial?) answer to bug 1064428.
749 binaries: set[BinaryPackageId] = set()
750 if add_faux and "all" in get_field("Architecture", "").split():
751 # the value "faux" in arch:faux is used elsewhere, so keep in sync
752 pkg_id = BinaryPackageId(
753 intern(f"{pkg}-faux"), intern("0~~~~"), intern("faux")
754 )
755 binaries.add(pkg_id)
757 sources[pkg] = SourcePackage(
758 pkg,
759 ver,
760 section,
761 binaries,
762 maint,
763 False,
764 build_deps_arch,
765 build_deps_indep,
766 get_field("Testsuite", "").split() or None,
767 get_field("Testsuite-Triggers", "").replace(",", "").split() or None,
768 )
770 return sources
773def _check_packages(
774 package: BinaryPackage,
775 archqual: str | None,
776 build_depends: bool,
777) -> bool:
778 """Helper for get_dependency_solvers
780 This method checks if a given package is valid as (Build-)Depends.
782 :param packages: which packages are to be updated
783 :param archqual: Architecture qualifier
784 :param build_depends: If True, check if the "package" parameter is valid as a build-dependency.
785 """
787 # See also bug #971739 and #1059929
788 if archqual is None:
789 return True
790 elif archqual == "native" and build_depends:
791 # Multi-arch handling for build-dependencies
792 # - :native is ok always
793 return True
794 elif archqual == "any" and package.multi_arch is MultiArch.ALLOWED:
795 # Multi-arch handling for both build-dependencies and regular dependencies
796 # - :any is ok iff the target has "M-A: allowed"
797 return True
798 return False
801def get_dependency_solvers(
802 block: Iterable[tuple[str, str, str]],
803 binaries_s_a: dict[str, BinaryPackage],
804 provides_s_a: dict[str, set[tuple[str, str]]],
805 *,
806 build_depends: bool = False,
807) -> Iterator[BinaryPackage]:
808 """Find the packages which satisfy a dependency block
810 This method returns the list of packages which satisfy a dependency
811 block (as returned by apt_pkg.parse_depends) in a package table
812 for a given suite and architecture (a la self.binaries[suite][arch])
814 It can also handle build-dependency relations if the named parameter
815 "build_depends" is set to True. In this case, block should be based
816 on the return value from apt_pkg.parse_src_depends.
818 :param block: The dependency block as parsed by apt_pkg.parse_depends (or apt_pkg.parse_src_depends
819 if the "build_depends" is True)
820 :param binaries_s_a: Mapping of package names to the relevant BinaryPackage
821 :param provides_s_a: Mapping of package names to their providers (as generated by parse_provides)
822 :param build_depends: If True, treat the "block" parameter as a build-dependency relation rather than
823 a regular dependency relation.
824 :return: package names solving the relation
825 """
827 # for every package, version and operation in the block
828 for name, version, op in block:
829 if ":" in name:
830 name, archqual = name.split(":", 1)
831 else:
832 archqual = None
834 # look for the package in unstable
835 if (package := binaries_s_a.get(name)) is not None:
836 # check the versioned dependency and architecture qualifier
837 # (if present)
838 if (
839 (op == "" and version == "")
840 or apt_pkg.check_dep(package.version, op, version)
841 ) and _check_packages(package, archqual, build_depends):
842 yield package
843 # look for the package in the virtual packages list and loop on them
844 if (provides := provides_s_a.get(name)) is not None:
845 for prov, prov_version in provides:
846 assert prov in binaries_s_a
847 package = binaries_s_a[prov]
848 # See Policy Manual §7.5
849 if (
850 (op == "" and version == "")
851 or (
852 prov_version != ""
853 and apt_pkg.check_dep(prov_version, op, version)
854 )
855 ) and _check_packages(package, archqual, build_depends):
856 yield package
859def invalidate_excuses(
860 excuses: dict[str, "Excuse"],
861 valid: set[str],
862 invalid: set[str],
863 invalidated: set[str],
864) -> None:
865 """Invalidate impossible excuses
867 This method invalidates the impossible excuses, which depend
868 on invalid excuses. The two parameters contains the sets of
869 `valid' and `invalid' excuses.
870 """
871 # make a list of all packages (source and binary) that are present in the
872 # excuses we have
873 excuses_packages: dict[PackageId | BinaryPackageId, set[str]] = defaultdict(set)
874 for exc in excuses.values():
875 for arch in exc.packages:
876 for pkg_arch_id in exc.packages[arch]:
877 # note that the same package can be in multiple excuses
878 # eg. when unstable and TPU have the same packages
879 excuses_packages[pkg_arch_id].add(exc.name)
881 # create dependencies between excuses based on packages
882 excuses_rdeps = defaultdict(set)
883 for exc in excuses.values():
884 # Note that excuses_rdeps is only populated by dependencies generated
885 # based on packages below. There are currently no dependencies between
886 # excuses that are added directly, so this is ok.
888 for pkg_dep in exc.depends_packages:
889 # set of excuses, each of which can satisfy this specific
890 # dependency
891 # if there is a dependency on a package for which no
892 # excuses exist (e.g. a cruft binary), the set will
893 # contain an ImpossibleDependencyState
894 dep_exc: set[str | DependencyState] = set()
895 for pkg_dep_id in cast(set[BinaryPackageId], pkg_dep.deps):
896 pkg_excuses = excuses_packages[pkg_dep_id]
897 # if the dependency isn't found, we get an empty set
898 if pkg_excuses == frozenset():
899 imp_dep = ImpossibleDependencyState(
900 PolicyVerdict.REJECTED_PERMANENTLY, pkg_dep_id.name
901 )
902 dep_exc.add(imp_dep)
904 else:
905 dep_exc |= pkg_excuses
906 for e in pkg_excuses:
907 excuses_rdeps[e].add(exc.name)
908 if not exc.add_dependency(dep_exc, pkg_dep.spec):
909 valid.discard(exc.name)
910 invalid.add(exc.name)
912 # loop on the invalid excuses
913 # Convert invalid to a list for deterministic results
914 invalid2 = sorted(invalid)
915 for ename in iter_except(invalid2.pop, IndexError):
916 invalidated.add(ename)
917 # if there is no reverse dependency, skip the item
918 if ename not in excuses_rdeps:
919 continue
921 rdep_verdict = PolicyVerdict.REJECTED_WAITING_FOR_ANOTHER_ITEM
922 if excuses[ename].policy_verdict.is_blocked:
923 rdep_verdict = PolicyVerdict.REJECTED_BLOCKED_BY_ANOTHER_ITEM
925 # loop on the reverse dependencies
926 for x in sorted(excuses_rdeps[ename]):
927 exc = excuses[x]
928 # if the item is valid and it is not marked as `forced', then we
929 # invalidate this specific dependency
930 if x in valid and not exc.forced:
931 # mark this specific dependency as invalid
932 still_valid = exc.invalidate_dependency(ename, rdep_verdict)
934 # if there are no alternatives left for this dependency,
935 # invalidate the excuse
936 if not still_valid:
937 valid.discard(x)
938 invalid2.append(x)
941def compile_nuninst(
942 target_suite: TargetSuite, architectures: list[str], nobreakall_arches: list[str]
943) -> dict[str, set[str]]:
944 """Compile a nuninst dict from the current testing
946 :param target_suite: The target suite
947 :param architectures: Which architectures to check
948 :param nobreakall_arches: Which architectures where arch:all packages must be installable
949 """
950 nuninst: dict[str, set[str]] = {}
951 binaries_t = target_suite.binaries
953 # for all the architectures
954 for arch in architectures:
955 # if it is in the nobreakall ones, check arch-independent packages too
956 check_archall = arch in nobreakall_arches
958 # check all the packages for this architecture
959 nuninst[arch] = set()
960 packages_t_a = binaries_t[arch]
961 for pkg_name, pkg_data in packages_t_a.items():
962 r = target_suite.is_installable(pkg_data.pkg_id)
963 if not r:
964 nuninst[arch].add(pkg_name)
966 # if they are not required, remove architecture-independent packages
967 nuninst[arch + "+all"] = nuninst[arch].copy()
968 if not check_archall:
969 for pkg_name in nuninst[arch + "+all"]:
970 pkg_data = packages_t_a[pkg_name]
971 if pkg_data.architecture == "all":
972 nuninst[arch].remove(pkg_name)
974 return nuninst
977def is_smooth_update_allowed(
978 binary: BinaryPackage, smooth_updates: list[str], hints: "HintCollection"
979) -> bool:
980 if "ALL" in smooth_updates: 980 ↛ 981line 980 didn't jump to line 981 because the condition on line 980 was never true
981 return True
982 section = binary.section.split("/")[-1]
983 if section in smooth_updates:
984 return True
985 # note that this needs to match the source version *IN TESTING*
986 return hints.has_hint(
987 "allow-smooth-update", package=binary.source, version=binary.source_version
988 )
991def find_smooth_updateable_binaries(
992 binaries_to_check: list[BinaryPackageId],
993 source_data: SourcePackage,
994 pkg_universe: "BinaryPackageUniverse",
995 target_suite: TargetSuite,
996 binaries_t: dict[str, dict[str, BinaryPackage]],
997 binaries_s: dict[str, dict[str, BinaryPackage]],
998 removals: set[BinaryPackageId] | None,
999 smooth_updates: list[str],
1000 hints: "HintCollection",
1001) -> set[BinaryPackageId]:
1002 check: set[BinaryPackageId] = set()
1003 smoothbins: set[BinaryPackageId] = set()
1005 # see note below in is_smooth_update_allowed branch
1006 rdeps_to_skip = set(binaries_to_check)
1007 if removals is not None:
1008 rdeps_to_skip.update(removals)
1010 for check_pkg_id in binaries_to_check:
1011 binary, parch = check_pkg_id.package_name, check_pkg_id.architecture
1013 cruftbins: set[BinaryPackageId] = set()
1015 # Not a candidate for smooth up date (newer non-cruft version in unstable)
1016 if (pkg := binaries_s[parch].get(binary)) is not None:
1017 if pkg.source_version == source_data.version:
1018 continue
1019 cruftbins.add(pkg.pkg_id)
1021 # Maybe a candidate (cruft or removed binary): check if config allows us to smooth update it.
1022 if is_smooth_update_allowed(binaries_t[parch][binary], smooth_updates, hints):
1023 # if the package has reverse-dependencies which are
1024 # built from other sources, it's a valid candidate for
1025 # a smooth update. if not, it may still be a valid
1026 # candidate if one if its r-deps is itself a candidate,
1027 # so note it for checking later
1028 #
1029 # We ignore all binaries listed in "removals" as we
1030 # assume they will leave at the same time as the
1031 # given package.
1032 rdeps = pkg_universe.reverse_dependencies_of(check_pkg_id) - rdeps_to_skip
1034 smooth_update_it = False
1035 if target_suite.any_of_these_are_in_the_suite(rdeps):
1036 for rdep in rdeps:
1037 # each dependency clause has a set of possible
1038 # alternatives that can satisfy that dependency.
1039 # if any of them is outside the set of smoothbins, the
1040 # dependency can be satisfied even if this binary was
1041 # removed, so there is no need to keep it around for a
1042 # smooth update
1043 # if not, only this binary can satisfy the dependency, so
1044 # we should keep it around until the rdep is no longer in
1045 # testing
1046 for dep_clause in pkg_universe.dependencies_of(rdep):
1047 # filter out cruft binaries from unstable, because
1048 # they will not be added to the set of packages that
1049 # will be migrated
1050 if all(
1051 x in smoothbins or x == check_pkg_id
1052 for x in dep_clause
1053 if x not in cruftbins
1054 ):
1055 smoothbins.add(check_pkg_id)
1056 smooth_update_it = True
1057 break
1059 if not smooth_update_it:
1060 check.add(check_pkg_id)
1062 # check whether we should perform a smooth update for
1063 # packages which are candidates but do not have r-deps
1064 # outside of the current source
1065 while 1:
1066 found_any = False
1067 for candidate_pkg_id in check:
1068 rdeps = pkg_universe.reverse_dependencies_of(candidate_pkg_id)
1069 if not rdeps.isdisjoint(smoothbins):
1070 smoothbins.add(candidate_pkg_id)
1071 found_any = True
1072 if not found_any:
1073 break
1074 check.difference_update(smoothbins)
1076 return smoothbins
1079def find_newer_binaries(
1080 suite_info: Suites, pkg: BinaryPackage, add_source_for_dropped_bin: bool = False
1081) -> Iterator[tuple[PackageId, Suite]]:
1082 """
1083 Find newer binaries for pkg in any of the source suites.
1085 :param pkg: BinaryPackage (is assumed to be in the target suite)
1087 :param add_source_for_dropped_bin: If True, newer versions of the
1088 source of pkg will be added if they don't have the binary pkg
1090 :return: the newer binaries (or sources) and their suites
1091 """
1092 source = pkg.source
1093 for suite in suite_info:
1094 if suite.suite_class is SuiteClass.TARGET_SUITE:
1095 continue
1097 suite_binaries_on_arch = suite.binaries.get(pkg.pkg_id.architecture)
1098 if not suite_binaries_on_arch: 1098 ↛ 1099line 1098 didn't jump to line 1099 because the condition on line 1098 was never true
1099 continue
1101 newerbin = suite_binaries_on_arch.get(pkg.pkg_id.package_name)
1102 if newerbin is not None:
1103 if suite.is_cruft(newerbin):
1104 # We pretend the cruft binary doesn't exist.
1105 # We handle this as if the source didn't have the binary
1106 # (see below)
1107 newerbin = None
1108 elif apt_pkg.version_compare(newerbin.version, pkg.version) <= 0:
1109 continue
1110 else:
1111 if source not in suite.sources:
1112 # bin and source not in suite: no newer version
1113 continue
1115 if newerbin is None:
1116 if not add_source_for_dropped_bin: 1116 ↛ 1117line 1116 didn't jump to line 1117 because the condition on line 1116 was never true
1117 continue
1118 # We only get here if there is a newer version of the source,
1119 # which doesn't have the binary anymore (either it doesn't
1120 # exist, or it's cruft and we pretend it doesn't exist).
1121 # Add the new source instead.
1122 nsrc = suite.sources[source]
1123 n_id = PackageId(source, nsrc.version, "source")
1124 overs = pkg.source_version
1125 if apt_pkg.version_compare(nsrc.version, overs) <= 0:
1126 continue
1127 else:
1128 n_id = newerbin.pkg_id
1130 yield (n_id, suite)
1133def parse_provides(
1134 provides_raw: str,
1135 pkg_id: BinaryPackageId | None = None,
1136 logger: logging.Logger | None = None,
1137) -> list[tuple[str, str]]:
1138 parts = apt_pkg.parse_depends(provides_raw, False)
1139 nprov = []
1140 for or_clause in parts:
1141 if len(or_clause) != 1: # pragma: no cover
1142 if logger is not None:
1143 msg = "Ignoring invalid provides in %s: Alternatives [%s]"
1144 logger.warning(msg, pkg_id, or_clause)
1145 continue
1146 for part in or_clause:
1147 provided, provided_version, op = part
1148 if op != "" and op != "=": # pragma: no cover
1149 if logger is not None:
1150 msg = "Ignoring invalid provides in %s: %s (%s %s)"
1151 logger.warning(msg, pkg_id, provided, op, provided_version)
1152 continue
1153 provided = sys.intern(provided)
1154 provided_version = sys.intern(provided_version)
1155 nprov.append((provided, provided_version))
1156 return nprov
1159def parse_builtusing(
1160 builtusing_raw: str,
1161 pkg_id: BinaryPackageId | None = None,
1162 logger: logging.Logger | None = None,
1163) -> list[tuple[str, str]]:
1164 parts = apt_pkg.parse_depends(builtusing_raw, False)
1165 nbu = []
1166 for or_clause in parts:
1167 if len(or_clause) != 1: # pragma: no cover
1168 if logger is not None:
1169 msg = "Ignoring invalid builtusing in %s: Alternatives [%s]"
1170 logger.warning(msg, pkg_id, or_clause)
1171 continue
1172 for part in or_clause:
1173 bu, bu_version, op = part
1174 if op != "=": # pragma: no cover
1175 if logger is not None:
1176 msg = "Ignoring invalid builtusing in %s: %s (%s %s)"
1177 logger.warning(msg, pkg_id, bu, op, bu_version)
1178 continue
1179 bu = sys.intern(bu)
1180 bu_version = sys.intern(bu_version)
1181 nbu.append((bu, bu_version))
1182 return nbu
1185def parse_option(
1186 options: "optparse.Values",
1187 option_name: str,
1188 default: Any | None = None,
1189 to_bool: bool = False,
1190 to_int: bool = False,
1191 day_to_sec: bool = False,
1192) -> None:
1193 """Ensure the option exist and has a sane value
1195 :param options: dict with options
1197 :param option_name: string with the name of the option
1199 :param default: the default value for the option
1201 :param to_int: convert the input to int (defaults to sys.maxsize)
1203 :param to_bool: convert the input to bool
1205 :param day_to_sec: convert the input from days to seconds (implies to_int=True)
1206 """
1207 value = getattr(options, option_name, default)
1209 # Option was provided with no value (or default is '') so pick up the default
1210 if value == "":
1211 value = default
1213 if (to_int or day_to_sec) and value in (None, ""):
1214 value = sys.maxsize
1216 if day_to_sec:
1217 value = int(float(value) * 24 * 60 * 60) # type: ignore[arg-type]
1219 if to_int:
1220 value = int(value) # type: ignore[arg-type]
1222 if to_bool:
1223 if value and (
1224 isinstance(value, bool) or value.lower() in ("yes", "y", "true", "t", "1")
1225 ):
1226 value = True
1227 else:
1228 value = False
1230 setattr(options, option_name, value)
1233def filter_out_faux_gen(
1234 binaries: Iterable[BinaryPackageId],
1235) -> Iterator[BinaryPackageId]:
1236 """Generator for packages without faux packages"""
1238 for pkg in binaries:
1239 if not pkg.package_name.endswith("-faux-build-depends"):
1240 yield pkg
1243def filter_out_faux(binaries: Iterable[BinaryPackageId]) -> set[BinaryPackageId]:
1244 """Returns a set without faux packages"""
1246 return {pkg for pkg in filter_out_faux_gen(binaries)}
1249def binaries_from_source_version(
1250 source_data: SourcePackage, suite_info: Suites
1251) -> tuple[set[BinaryPackageId], str]:
1252 """Returns a set of real bid with only packages from this source version"""
1254 binaries = source_data.binaries.copy()
1255 # We don't know from which suite the source version comes
1256 for suite in suite_info.source_suites: 1256 ↛ 1267line 1256 didn't jump to line 1267 because the loop on line 1256 didn't complete
1257 # But if it's there, we assume it will have all the associated binaries
1258 if source_data.source in suite.sources: 1258 ↛ 1256line 1258 didn't jump to line 1256 because the condition on line 1258 was always true
1259 for bid in binaries.copy():
1260 if (
1261 suite.all_binaries_in_suite[bid].source_version
1262 != source_data.version
1263 ):
1264 binaries.remove(bid)
1265 break
1267 return filter_out_faux(binaries), suite.name
1270def get_component(section: str) -> str:
1271 """Returns the component based on the section"""
1273 # horrible hard-coding, but currently, we don't keep track of the component
1274 # when loading the packages files, but let's centralize it here
1275 component = "main"
1276 if "/" in section:
1277 component = section.split("/")[0]
1278 return component