Coverage for britney2/utils.py: 92%
495 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-07-30 07:06 +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 Callable,
31 Container,
32 Iterable,
33 Iterator,
34 Mapping,
35 MutableSet,
36)
37from datetime import UTC, datetime
38from enum import Enum, StrEnum
39from functools import partial
40from itertools import chain, filterfalse
41from typing import (
42 IO,
43 TYPE_CHECKING,
44 Any,
45 Literal,
46 Protocol,
47 TypeVar,
48 Union,
49 cast,
50 overload,
51)
53import apt_pkg
54import yaml
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__)
139# iter_except is from the "itertools" recipe
140def iter_except(
141 func: Callable[[], _T],
142 exception: type[BaseException] | tuple[type[BaseException], ...],
143 first: Any = None,
144) -> Iterator[_T]: # pragma: no cover - itertools recipe function
145 """Call a function repeatedly until an exception is raised.
147 Converts a call-until-exception interface to an iterator interface.
148 Like __builtin__.iter(func, sentinel) but uses an exception instead
149 of a sentinel to end the loop.
151 Examples:
152 bsddbiter = iter_except(db.next, bsddb.error, db.first)
153 heapiter = iter_except(functools.partial(heappop, h), IndexError)
154 dictiter = iter_except(d.popitem, KeyError)
155 dequeiter = iter_except(d.popleft, IndexError)
156 queueiter = iter_except(q.get_nowait, Queue.Empty)
157 setiter = iter_except(s.pop, KeyError)
159 """
160 try:
161 if first is not None:
162 yield first()
163 while 1:
164 yield func()
165 except exception:
166 pass
169def log_and_format_old_libraries(
170 logger: logging.Logger, libs: list["MigrationItem"]
171) -> None:
172 """Format and log old libraries in a table (no header)"""
173 libraries: dict[str, list[str]] = {}
174 for i in libs:
175 pkg = i.package
176 if pkg in libraries:
177 libraries[pkg].append(i.architecture)
178 else:
179 libraries[pkg] = [i.architecture]
181 for lib in sorted(libraries):
182 logger.info(" %s: %s", lib, " ".join(libraries[lib]))
185def compute_reverse_tree(
186 pkg_universe: "BinaryPackageUniverse", affected: set[BinaryPackageId]
187) -> None:
188 """Calculate the full dependency tree for a set of packages
190 This method returns the full dependency tree for a given set of
191 packages. The first argument is an instance of the BinaryPackageUniverse
192 and the second argument are a set of BinaryPackageId.
194 The set of affected packages will be updated in place and must
195 therefore be mutable.
196 """
197 remain = list(affected)
198 while remain:
199 pkg_id = remain.pop()
200 new_pkg_ids = pkg_universe.reverse_dependencies_of(pkg_id) - affected
201 affected.update(new_pkg_ids)
202 remain.extend(new_pkg_ids)
205def add_transitive_dependencies_flatten(
206 pkg_universe: "BinaryPackageUniverse", initial_set: MutableSet[BinaryPackageId]
207) -> None:
208 """Find and include all transitive dependencies
210 This method updates the initial_set parameter to include all transitive
211 dependencies. The first argument is an instance of the BinaryPackageUniverse
212 and the second argument are a set of BinaryPackageId.
214 The set of initial packages will be updated in place and must
215 therefore be mutable.
216 """
217 remain = list(initial_set)
218 while remain:
219 pkg_id = remain.pop()
220 new_pkg_ids = {
221 x
222 for x in chain.from_iterable(pkg_universe.dependencies_of(pkg_id))
223 if x not in initial_set
224 }
225 initial_set |= new_pkg_ids
226 remain.extend(new_pkg_ids)
229def write_nuninst(filename: str, nuninst: dict[str, set[str]]) -> None:
230 """Write the non-installable report
232 Write the non-installable report derived from "nuninst" to the
233 file denoted by "filename".
234 """
235 with open(filename, "w", encoding="utf-8") as f:
236 # Having two fields with (almost) identical dates seems a bit
237 # redundant.
238 f.write(
239 "Built on: "
240 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
241 + "\n"
242 )
243 f.write(
244 "Last update: "
245 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
246 + "\n\n"
247 )
248 for k in nuninst:
249 f.write("{}: {}\n".format(k, " ".join(nuninst[k])))
252def read_nuninst(filename: str, architectures: set[str]) -> dict[str, set[str]]:
253 """Read the non-installable report
255 Read the non-installable report from the file denoted by
256 "filename" and return it. Only architectures in "architectures"
257 will be included in the report.
258 """
259 nuninst: dict[str, set[str]] = {}
260 with open(filename, encoding="utf-8") as f:
261 for r in f:
262 if ":" not in r:
263 continue
264 arch, packages = r.strip().split(":", 1)
265 if arch.split("+", 1)[0] in architectures:
266 nuninst[arch] = set(packages.split())
267 return nuninst
270def newly_uninst(
271 nuold: dict[str, set[str]], nunew: dict[str, set[str]]
272) -> dict[str, list[str]]:
273 """Return a nuninst statistic with only new uninstallable packages
275 This method subtracts the uninstallable packages of the statistic
276 "nunew" from the statistic "nuold".
278 It returns a dictionary with the architectures as keys and the list
279 of uninstallable packages as values. If there are no regressions
280 on a given architecture, then the architecture will be omitted in
281 the result. Accordingly, if none of the architectures have
282 regressions an empty directory is returned.
283 """
284 res: dict[str, list[str]] = {}
285 for arch in ifilter_only(nunew, nuold):
286 arch_nuninst = [x for x in nunew[arch] if x not in nuold[arch]]
287 # Leave res empty if there are no newly uninst packages
288 if arch_nuninst:
289 res[arch] = arch_nuninst
290 return res
293def format_and_log_uninst(
294 logger: logging.Logger,
295 architectures: Iterable[str],
296 nuninst: Mapping[str, Iterable[str]],
297 *,
298 loglevel: int = logging.INFO,
299) -> None:
300 """Emits the uninstallable packages to the log
302 An example of the output string is:
303 * i386: broken-pkg1, broken-pkg2
305 Note that if there is no uninstallable packages, then nothing is emitted.
306 """
307 for arch in architectures:
308 if arch in nuninst and nuninst[arch]:
309 msg = " * {}: {}".format(arch, ", ".join(sorted(nuninst[arch])))
310 logger.log(loglevel, msg)
313class Sorted(Protocol):
314 def __call__( 314 ↛ exitline 314 didn't jump to the function exit
315 self,
316 iterable: Iterable["SupportsRichComparisonT"],
317 /,
318 *,
319 key: None = None,
320 reverse: bool = False,
321 ) -> list["SupportsRichComparisonT"]: ...
324def write_heidi(
325 filename: str,
326 target_suite: TargetSuite,
327 *,
328 outofsync_arches: frozenset[str] = frozenset(),
329 sorted: Sorted = sorted,
330) -> None:
331 """Write the output HeidiResult
333 This method write the output for Heidi, which contains all the
334 binary packages and the source packages in the form:
336 <pkg-name> <pkg-version> <pkg-architecture> <pkg-section>
337 <src-name> <src-version> source <src-section>
339 The file is written as "filename" using the sources and packages
340 from the "target_suite" parameter.
342 outofsync_arches: If given, it is a set of architectures marked
343 as "out of sync". The output file may exclude some out of date
344 arch:all packages for those architectures to reduce the noise.
346 The "X=X" parameters are optimizations to avoid "load global" in
347 the loops.
348 """
349 sources_t = target_suite.sources
350 packages_t = target_suite.binaries
352 with open(filename, "w", encoding="ascii") as f:
354 # write binary packages
355 for arch in sorted(packages_t):
356 binaries = packages_t[arch]
357 for pkg_name in sorted(binaries):
358 pkg = binaries[pkg_name]
359 pkgv = pkg.version
360 pkgarch = pkg.architecture or "all"
361 pkgsec = pkg.section or "faux"
362 if pkgsec == "faux" or pkgsec.endswith("/faux"):
363 # Faux package; not really a part of testing
364 continue
365 if ( 365 ↛ 377line 365 didn't jump to line 377
366 pkg.source_version
367 and pkgarch == "all"
368 and pkg.source_version != sources_t[pkg.source].version
369 and arch in outofsync_arches
370 ):
371 # when architectures are marked as "outofsync", their binary
372 # versions may be lower than those of the associated
373 # source package in testing. the binary package list for
374 # such architectures will include arch:all packages
375 # matching those older versions, but we only want the
376 # newer arch:all in testing
377 continue
378 f.write(f"{pkg_name} {pkgv} {pkgarch} {pkgsec}\n")
380 # write sources
381 for src_name in sorted(sources_t):
382 src = sources_t[src_name]
383 srcv = src.version
384 srcsec = src.section or "unknown"
385 if srcsec == "faux" or srcsec.endswith("/faux"):
386 # Faux package; not really a part of testing
387 continue
388 f.write(f"{src_name} {srcv} source {srcsec}\n")
391def write_heidi_delta(filename: str, all_selected: list["MigrationItem"]) -> None:
392 """Write the output delta
394 This method writes the packages to be upgraded, in the form:
395 <src-name> <src-version>
396 or (if the source is to be removed):
397 -<src-name> <src-version>
399 The order corresponds to that shown in update_output.
400 """
401 with open(filename, "w", encoding="ascii") as fd:
403 fd.write("#HeidiDelta\n")
405 for item in all_selected:
406 prefix = ""
408 if item.is_removal:
409 prefix = "-"
411 if item.architecture == "source":
412 fd.write(f"{prefix}{item.package} {item.version}\n")
413 else:
414 fd.write(
415 "%s%s %s %s\n"
416 % (prefix, item.package, item.version, item.architecture)
417 )
420class Opener(Protocol):
421 def __call__( 421 ↛ exitline 421 didn't jump to the function exit
422 self, file: str, mode: Literal["wt"], encoding: Literal["utf-8"]
423 ) -> IO[Any]: ...
426class ExcusesOutputFormat(Enum):
427 YAML = 0
428 LEGACY_HTML = 1
431def write_excuses(
432 excuses: dict[str, "Excuse"] | dict[PackageId, "Excuse"],
433 dest_file: str,
434 output_format: ExcusesOutputFormat = ExcusesOutputFormat.YAML,
435) -> None:
436 """Write the excuses to dest_file
438 Writes a list of excuses in a specified output_format to the
439 path denoted by dest_file. The output_format can either be "yaml"
440 or "legacy-html".
441 """
442 excuselist = sorted(excuses.values(), key=lambda x: x.sortkey())
443 if output_format is ExcusesOutputFormat.YAML:
444 # use custom representer to avoid creation of the full list with all excuse data before starting the serialization
445 def represent_yaml_excuse(dumper: yaml.Dumper, data: Excuse) -> yaml.Node:
446 return dumper.represent_data(data.excusedata(excuses))
448 yaml.add_representer(Excuse, represent_yaml_excuse)
449 yaml.Dumper.add_multi_representer(
450 StrEnum, yaml.representer.Representer.represent_str
451 )
453 os.makedirs(os.path.dirname(dest_file), exist_ok=True)
454 opener: Opener = open # type: ignore[assignment]
455 if dest_file.endswith(".xz"): 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 import lzma
458 opener = lzma.open # type: ignore[assignment]
459 elif dest_file.endswith(".gz"): 459 ↛ 460line 459 didn't jump to line 460 because the condition on line 459 was never true
460 import gzip
462 opener = gzip.open # type: ignore[assignment]
463 with opener(f"{dest_file}.new", "wt", encoding="utf-8") as f:
464 excusesdata = {
465 "sources": excuselist,
466 "generated-date": datetime.now(UTC),
467 }
468 yaml.dump(
469 excusesdata, stream=f, default_flow_style=False, allow_unicode=True
470 )
471 os.replace(f"{dest_file}.new", dest_file)
472 elif output_format is ExcusesOutputFormat.LEGACY_HTML:
473 with open(f"{dest_file}.new", "w", encoding="utf-8") as f:
474 f.write(
475 '<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/REC-html40/strict.dtd">\n'
476 )
477 f.write("<html><head><title>excuses...</title>")
478 f.write(
479 '<meta http-equiv="Content-Type" content="text/html;charset=utf-8"></head><body>\n'
480 )
481 f.write(
482 "<p>Generated: "
483 + time.strftime("%Y.%m.%d %H:%M:%S %z", time.gmtime(time.time()))
484 + "</p>\n"
485 )
486 f.write("<ul>\n")
487 for e in excuselist:
488 f.write("<li>%s" % e.html(excuses))
489 f.write("</ul></body></html>\n")
490 os.replace(f"{dest_file}.new", dest_file)
491 else: # pragma: no cover
492 raise ValueError('Output format must be either "YAML or "LEGACY_HTML"')
495def old_libraries(
496 mi_factory: "MigrationItemFactory",
497 suite_info: Suites,
498 outofsync_arches: Iterable[str] = frozenset(),
499) -> list["MigrationItem"]:
500 """Detect old libraries left in the target suite for smooth transitions
502 This method detects old libraries which are in the target suite but no
503 longer built from the source package: they are still there because
504 other packages still depend on them, but they should be removed as
505 soon as possible.
507 For "outofsync" architectures, outdated binaries are allowed to be in
508 the target suite, so they are only added to the removal list if they
509 are no longer in the (primary) source suite.
510 """
511 sources_t = suite_info.target_suite.sources
512 binaries_t = suite_info.target_suite.binaries
513 binaries_s = suite_info.primary_source_suite.binaries
514 removals = []
515 for arch, binaries in binaries_t.items():
516 for pkg_name, pkg in binaries.items():
517 if sources_t[pkg.source].version != pkg.source_version and (
518 arch not in outofsync_arches or pkg_name not in binaries_s[arch]
519 ):
520 removals.append(mi_factory.generate_removal_for_cruft_item(pkg.pkg_id))
521 return removals
524def is_nuninst_asgood_generous(
525 constraints: dict[str, list[str]],
526 allow_uninst: dict[str, set[str | None]],
527 architectures: list[str],
528 old: dict[str, set[str]],
529 new: dict[str, set[str]],
530 break_arches: set[str] = cast(set[str], frozenset()),
531) -> bool:
532 """Compares the nuninst counters and constraints to see if they improved
534 Given a list of architectures, the previous and the current nuninst
535 counters, this function determines if the current nuninst counter
536 is better than the previous one. Optionally it also accepts a set
537 of "break_arches", the nuninst counter for any architecture listed
538 in this set are completely ignored.
540 If the nuninst counters are equal or better, then the constraints
541 are checked for regressions (ignoring break_arches).
543 Returns True if the new nuninst counter is better than the
544 previous and there are no constraint regressions (ignoring Break-archs).
545 Returns False otherwise.
547 """
548 diff = 0
549 for arch in architectures:
550 if arch in break_arches:
551 continue
552 diff = diff + (
553 len(new[arch] - allow_uninst[arch]) - len(old[arch] - allow_uninst[arch])
554 )
555 if diff > 0:
556 return False
557 must_be_installable = constraints["keep-installable"]
558 for arch in architectures:
559 if arch in break_arches:
560 continue
561 regression = new[arch] - old[arch]
562 if not regression.isdisjoint(must_be_installable): 562 ↛ 563line 562 didn't jump to line 563 because the condition on line 562 was never true
563 return False
564 return True
567def clone_nuninst(
568 nuninst: dict[str, set[str]],
569 *,
570 packages_s: dict[str, dict[str, BinaryPackage]] | None = None,
571 architectures: Iterable[str] | None = None,
572) -> dict[str, set[str]]:
573 """Completely or Selectively deep clone nuninst
575 Given nuninst table, the package table for a given suite and
576 a list of architectures, this function will clone the nuninst
577 table. Only the listed architectures will be deep cloned -
578 the rest will only be shallow cloned. When packages_s is given,
579 packages not listed in packages_s will be pruned from the clone
580 (if packages_s is omitted, the per architecture nuninst is cloned
581 as-is)
582 """
583 clone = nuninst.copy()
584 if architectures is None: 584 ↛ 585line 584 didn't jump to line 585 because the condition on line 584 was never true
585 return clone
586 if packages_s is not None:
587 for arch in architectures:
588 clone[arch] = {x for x in nuninst[arch] if x in packages_s[arch]}
589 clone[arch + "+all"] = {
590 x for x in nuninst[arch + "+all"] if x in packages_s[arch]
591 }
592 else:
593 for arch in architectures:
594 clone[arch] = set(nuninst[arch])
595 clone[arch + "+all"] = set(nuninst[arch + "+all"])
596 return clone
599def test_installability(
600 target_suite: TargetSuite,
601 pkg_name: str,
602 pkg_id: BinaryPackageId,
603 broken: set[str],
604 nuninst_arch: set[str] | None,
605) -> None:
606 """Test for installability of a package on an architecture
608 (pkg_name, pkg_version, pkg_arch) is the package to check.
610 broken is the set of broken packages. If p changes
611 installability (e.g. goes from uninstallable to installable),
612 broken will be updated accordingly.
614 If nuninst_arch is not None then it also updated in the same
615 way as broken is.
616 """
617 if not target_suite.is_installable(pkg_id):
618 # if pkg_name not in broken: regression else: already broken
619 broken.add(pkg_name)
620 if nuninst_arch is not None:
621 nuninst_arch.add(pkg_name)
622 else:
623 # if pkg_name in broken: # improvement else: already not broken
624 broken.discard(pkg_name)
625 if nuninst_arch is not None:
626 nuninst_arch.discard(pkg_name)
629def check_installability(
630 target_suite: TargetSuite,
631 binaries: dict[str, dict[str, BinaryPackage]],
632 arch: str,
633 updates: set[BinaryPackageId],
634 check_archall: bool,
635 nuninst: dict[str, set[str]],
636) -> None:
637 broken = nuninst[arch + "+all"]
638 packages_t_a = binaries[arch]
640 for pkg_id in (x for x in updates if x.architecture == arch):
641 name, version, parch = pkg_id.package_name, pkg_id.version, pkg_id.architecture
642 if name not in packages_t_a:
643 continue
644 pkgdata = packages_t_a[name]
645 if version != pkgdata.version:
646 # Not the version in testing right now, ignore
647 continue
648 actual_arch = pkgdata.architecture
649 nuninst_arch = None
650 # only check arch:all packages if requested
651 if check_archall or actual_arch != "all":
652 nuninst_arch = nuninst[parch]
653 elif actual_arch == "all": 653 ↛ 655line 653 didn't jump to line 655 because the condition on line 653 was always true
654 nuninst[parch].discard(name)
655 test_installability(target_suite, name, pkg_id, broken, nuninst_arch)
658def possibly_compressed(
659 path: str, *, permitted_compressions: list[str] | None = None
660) -> str:
661 """Find and select a (possibly compressed) variant of a path
663 If the given path exists, it will be returned
665 :param path: The base path.
666 :param permitted_compressions: Alternative extensions to look for. Defaults to "gz" and "xz".
667 :return: The path given possibly with one of the permitted extensions.
668 :raises FileNotFoundError: if the path is not found
669 """
670 if os.path.exists(path): 670 ↛ 672line 670 didn't jump to line 672 because the condition on line 670 was always true
671 return path
672 if permitted_compressions is None:
673 permitted_compressions = ["gz", "xz"]
674 for ext in permitted_compressions:
675 cpath = f"{path}.{ext}"
676 if os.path.exists(cpath):
677 return cpath
678 raise FileNotFoundError(
679 errno.ENOENT, os.strerror(errno.ENOENT), path
680 ) # pragma: no cover
683def create_provides_map(
684 packages: dict[str, BinaryPackage],
685) -> dict[str, set[tuple[str, str]]]:
686 """Create a provides map from a map binary package names and their BinaryPackage objects
688 :param packages: A dict mapping binary package names to their BinaryPackage object
689 :return: A provides map
690 """
691 # create provides
692 provides = defaultdict(set)
694 for pkg, dpkg in packages.items():
695 if dpkg.provides is None:
696 continue
697 # register virtual packages and real packages that provide
698 # them
699 for provided_pkg, provided_version, _ in dpkg.provides:
700 provides[provided_pkg].add((pkg, provided_version))
702 return provides
705def read_release_file(suite_dir: str) -> "TagSection[str]":
706 """Parses a given "Release" file
708 :param suite_dir: The directory to the suite
709 :return: A dict of the first (and only) paragraph in an Release file
710 """
711 release_file = os.path.join(suite_dir, "Release")
712 with open(release_file) as fd:
713 tag_file = iter(apt_pkg.TagFile(fd))
714 result = next(tag_file)
715 if next(tag_file, None) is not None: # pragma: no cover
716 raise TypeError("%s has more than one paragraph" % release_file)
717 return result
720def read_sources_file(
721 filename: str,
722 sources: dict[str, SourcePackage] | None = None,
723 add_faux: bool = True,
724 sources_target_suite: dict[str, SourcePackage] | None = None,
725 intern: Callable[[str], str] = sys.intern,
726) -> dict[str, SourcePackage]:
727 """Parse a single Sources file into a hash
729 Parse a single Sources file into a dict mapping a source package
730 name to a SourcePackage object. If there are multiple source
731 packages with the same version, then highest versioned source
732 package (that is not marked as "Extra-Source-Only") is the
733 version kept in the dict.
735 :param filename: Path to the Sources file. Can be compressed by any algorithm supported by apt_pkg.TagFile
736 :param sources: Optional dict to add the packages to. If given, this is also the value returned.
737 :param add_faux: Add a faux arch:all binary for each source that claims it has arch:all
738 :param sources_target_suite: SourcPackages loaded from the target suite for memory optimizations
739 :param intern: Internal optimisation / implementation detail to avoid python's "LOAD_GLOBAL" instruction in a loop
740 :return: mapping from names to a source package
741 """
742 if sources is None:
743 sources = {}
744 if sources_target_suite is None:
745 sources_target_suite = {}
747 tag_file = apt_pkg.TagFile(filename)
748 get_field = tag_file.section.get
749 step = tag_file.step
751 while step():
752 if get_field("Extra-Source-Only", "no") == "yes":
753 # Ignore sources only referenced by Built-Using
754 continue
755 pkg = get_field("Package")
756 ver = get_field("Version")
757 # There may be multiple versions of the source package
758 # (in unstable) if some architectures have out-of-date
759 # binaries. We only ever consider the source with the
760 # largest version for migration.
761 if pkg in sources and apt_pkg.version_compare(sources[pkg].version, ver) > 0:
762 continue
763 maint = get_field("Maintainer")
764 if maint: 764 ↛ 766line 764 didn't jump to line 766 because the condition on line 764 was always true
765 maint = intern(maint.strip())
766 section = get_field("Section")
767 if section: 767 ↛ 770line 767 didn't jump to line 770 because the condition on line 767 was always true
768 section = intern(section.strip())
769 build_deps_arch: str | None
770 build_deps_arch = ", ".join(
771 x
772 for x in (get_field("Build-Depends"), get_field("Build-Depends-Arch"))
773 if x is not None
774 )
775 if build_deps_arch != "":
776 build_deps_arch = sys.intern(build_deps_arch)
777 else:
778 build_deps_arch = None
779 build_deps_indep = get_field("Build-Depends-Indep")
780 if build_deps_indep is not None:
781 build_deps_indep = sys.intern(build_deps_indep)
783 # Adding arch:all packages to the list of binaries already to be able
784 # to check for them later. Helps mitigate bug 887060 and is the
785 # (partial?) answer to bug 1064428.
786 binaries: set[BinaryPackageId] = set()
787 if add_faux and "all" in get_field("Architecture", "").split():
788 # the value "faux" in arch:faux is used elsewhere, so keep in sync
789 pkg_id = BinaryPackageId(f"{pkg}-faux", intern("0~~~~"), intern("faux"))
790 binaries.add(pkg_id)
792 pkg = intern(pkg)
793 ver = intern(ver)
794 sources[pkg] = srcpkg = SourcePackage(
795 pkg,
796 ver,
797 section,
798 binaries,
799 maint,
800 False,
801 build_deps_arch,
802 build_deps_indep,
803 get_field("Testsuite", "").split(),
804 get_field("Testsuite-Triggers", "").replace(",", "").split(),
805 )
807 if (
808 srcpkg_target := sources_target_suite.get(pkg, None)
809 ) is not None and srcpkg_target.version == ver:
810 # If the source package exists and the version is the same, reuse the already stored data.
811 # Note that the binaries field may be different if cruft packages are involved.
812 srcpkg.build_deps_arch = srcpkg_target.build_deps_arch
813 srcpkg.build_deps_indep = srcpkg_target.build_deps_indep
814 srcpkg.testsuite = srcpkg_target.testsuite
815 srcpkg.testsuite_triggers = srcpkg_target.testsuite_triggers
816 return sources
819def _check_and_update_packages(
820 packages: list[BinaryPackage],
821 package: BinaryPackage,
822 archqual: str | None,
823 build_depends: bool,
824) -> None:
825 """Helper for get_dependency_solvers
827 This method updates the list of packages with a given package if that
828 package is a valid (Build-)Depends.
830 :param packages: which packages are to be updated
831 :param archqual: Architecture qualifier
832 :param build_depends: If True, check if the "package" parameter is valid as a build-dependency.
833 """
835 # See also bug #971739 and #1059929
836 if archqual is None:
837 packages.append(package)
838 elif archqual == "native" and build_depends:
839 # Multi-arch handling for build-dependencies
840 # - :native is ok always
841 packages.append(package)
842 elif archqual == "any" and package.multi_arch is MultiArch.ALLOWED:
843 # Multi-arch handling for both build-dependencies and regular dependencies
844 # - :any is ok iff the target has "M-A: allowed"
845 packages.append(package)
848class GetDependencySolversProto(Protocol):
849 def __call__( 849 ↛ exitline 849 didn't jump to the function exit
850 self,
851 block: Iterable[tuple[str, str, str]],
852 binaries_s_a: dict[str, BinaryPackage],
853 provides_s_a: dict[str, set[tuple[str, str]]],
854 *,
855 build_depends: bool = False,
856 ) -> list[BinaryPackage]: ...
859def get_dependency_solvers(
860 block: Iterable[tuple[str, str, str]],
861 binaries_s_a: dict[str, BinaryPackage],
862 provides_s_a: dict[str, set[tuple[str, str]]],
863 *,
864 build_depends: bool = False,
865) -> list[BinaryPackage]:
866 """Find the packages which satisfy a dependency block
868 This method returns the list of packages which satisfy a dependency
869 block (as returned by apt_pkg.parse_depends) in a package table
870 for a given suite and architecture (a la self.binaries[suite][arch])
872 It can also handle build-dependency relations if the named parameter
873 "build_depends" is set to True. In this case, block should be based
874 on the return value from apt_pkg.parse_src_depends.
876 :param block: The dependency block as parsed by apt_pkg.parse_depends (or apt_pkg.parse_src_depends
877 if the "build_depends" is True)
878 :param binaries_s_a: Mapping of package names to the relevant BinaryPackage
879 :param provides_s_a: Mapping of package names to their providers (as generated by parse_provides)
880 :param build_depends: If True, treat the "block" parameter as a build-dependency relation rather than
881 a regular dependency relation.
882 :return: package names solving the relation
883 """
884 packages: list[BinaryPackage] = []
886 # for every package, version and operation in the block
887 for name, version, op in block:
888 if ":" in name:
889 name, archqual = name.split(":", 1)
890 else:
891 archqual = None
893 # look for the package in unstable
894 if (package := binaries_s_a.get(name)) is not None:
895 # check the versioned dependency and architecture qualifier
896 # (if present)
897 if (op == "" and version == "") or apt_pkg.check_dep(
898 package.version, op, version
899 ):
900 _check_and_update_packages(packages, package, archqual, build_depends)
902 # look for the package in the virtual packages list and loop on them
903 if (provides := provides_s_a.get(name)) is not None:
904 for prov, prov_version in provides:
905 assert prov in binaries_s_a
906 package = binaries_s_a[prov]
907 # See Policy Manual §7.5
908 if (op == "" and version == "") or (
909 prov_version != "" and apt_pkg.check_dep(prov_version, op, version)
910 ):
911 _check_and_update_packages(
912 packages, package, archqual, build_depends
913 )
915 return packages
918def invalidate_excuses(
919 excuses: dict[str, "Excuse"],
920 valid: set[str],
921 invalid: set[str],
922 invalidated: set[str],
923) -> None:
924 """Invalidate impossible excuses
926 This method invalidates the impossible excuses, which depend
927 on invalid excuses. The two parameters contains the sets of
928 `valid' and `invalid' excuses.
929 """
930 # make a list of all packages (source and binary) that are present in the
931 # excuses we have
932 excuses_packages: dict[PackageId | BinaryPackageId, set[str]] = defaultdict(set)
933 for exc in excuses.values():
934 for arch in exc.packages:
935 for pkg_arch_id in exc.packages[arch]:
936 # note that the same package can be in multiple excuses
937 # eg. when unstable and TPU have the same packages
938 excuses_packages[pkg_arch_id].add(exc.name)
940 # create dependencies between excuses based on packages
941 excuses_rdeps = defaultdict(set)
942 for exc in excuses.values():
943 # Note that excuses_rdeps is only populated by dependencies generated
944 # based on packages below. There are currently no dependencies between
945 # excuses that are added directly, so this is ok.
947 for pkg_dep in exc.depends_packages:
948 # set of excuses, each of which can satisfy this specific
949 # dependency
950 # if there is a dependency on a package for which no
951 # excuses exist (e.g. a cruft binary), the set will
952 # contain an ImpossibleDependencyState
953 dep_exc: set[str | DependencyState] = set()
954 for pkg_dep_id in cast(set[BinaryPackageId], pkg_dep.deps):
955 pkg_excuses = excuses_packages[pkg_dep_id]
956 # if the dependency isn't found, we get an empty set
957 if pkg_excuses == frozenset():
958 imp_dep = ImpossibleDependencyState(
959 PolicyVerdict.REJECTED_PERMANENTLY, pkg_dep_id.name
960 )
961 dep_exc.add(imp_dep)
963 else:
964 dep_exc |= pkg_excuses
965 for e in pkg_excuses:
966 excuses_rdeps[e].add(exc.name)
967 if not exc.add_dependency(dep_exc, pkg_dep.spec):
968 valid.discard(exc.name)
969 invalid.add(exc.name)
971 # loop on the invalid excuses
972 # Convert invalid to a list for deterministic results
973 invalid2 = sorted(invalid)
974 for ename in iter_except(invalid2.pop, IndexError):
975 invalidated.add(ename)
976 # if there is no reverse dependency, skip the item
977 if ename not in excuses_rdeps:
978 continue
980 rdep_verdict = PolicyVerdict.REJECTED_WAITING_FOR_ANOTHER_ITEM
981 if excuses[ename].policy_verdict.is_blocked:
982 rdep_verdict = PolicyVerdict.REJECTED_BLOCKED_BY_ANOTHER_ITEM
984 # loop on the reverse dependencies
985 for x in sorted(excuses_rdeps[ename]):
986 exc = excuses[x]
987 # if the item is valid and it is not marked as `forced', then we
988 # invalidate this specific dependency
989 if x in valid and not exc.forced:
990 # mark this specific dependency as invalid
991 still_valid = exc.invalidate_dependency(ename, rdep_verdict)
993 # if there are no alternatives left for this dependency,
994 # invalidate the excuse
995 if not still_valid:
996 valid.discard(x)
997 invalid2.append(x)
1000def compile_nuninst(
1001 target_suite: TargetSuite, architectures: list[str], nobreakall_arches: list[str]
1002) -> dict[str, set[str]]:
1003 """Compile a nuninst dict from the current testing
1005 :param target_suite: The target suite
1006 :param architectures: Which architectures to check
1007 :param nobreakall_arches: Which architectures where arch:all packages must be installable
1008 """
1009 nuninst: dict[str, set[str]] = {}
1010 binaries_t = target_suite.binaries
1012 # for all the architectures
1013 for arch in architectures:
1014 # if it is in the nobreakall ones, check arch-independent packages too
1015 check_archall = arch in nobreakall_arches
1017 # check all the packages for this architecture
1018 nuninst[arch] = set()
1019 packages_t_a = binaries_t[arch]
1020 for pkg_name, pkg_data in packages_t_a.items():
1021 r = target_suite.is_installable(pkg_data.pkg_id)
1022 if not r:
1023 nuninst[arch].add(pkg_name)
1025 # if they are not required, remove architecture-independent packages
1026 nuninst[arch + "+all"] = nuninst[arch].copy()
1027 if not check_archall:
1028 for pkg_name in nuninst[arch + "+all"]:
1029 pkg_data = packages_t_a[pkg_name]
1030 if pkg_data.architecture == "all":
1031 nuninst[arch].remove(pkg_name)
1033 return nuninst
1036def is_smooth_update_allowed(
1037 binary: BinaryPackage, smooth_updates: list[str], hints: "HintCollection"
1038) -> bool:
1039 if "ALL" in smooth_updates: 1039 ↛ 1040line 1039 didn't jump to line 1040 because the condition on line 1039 was never true
1040 return True
1041 section = binary.section.split("/")[-1]
1042 if section in smooth_updates:
1043 return True
1044 # note that this needs to match the source version *IN TESTING*
1045 return hints.has_hint(
1046 "allow-smooth-update", package=binary.source, version=binary.source_version
1047 )
1050def find_smooth_updateable_binaries(
1051 binaries_to_check: list[BinaryPackageId],
1052 source_data: SourcePackage,
1053 pkg_universe: "BinaryPackageUniverse",
1054 target_suite: TargetSuite,
1055 binaries_t: dict[str, dict[str, BinaryPackage]],
1056 binaries_s: dict[str, dict[str, BinaryPackage]],
1057 removals: set[BinaryPackageId] | frozenset[BinaryPackageId],
1058 smooth_updates: list[str],
1059 hints: "HintCollection",
1060) -> set[BinaryPackageId]:
1061 check: set[BinaryPackageId] = set()
1062 smoothbins: set[BinaryPackageId] = set()
1064 binaries_to_check_set = set(binaries_to_check)
1065 for check_pkg_id in binaries_to_check:
1066 binary, parch = check_pkg_id.package_name, check_pkg_id.architecture
1068 cruftbins: set[BinaryPackageId] = set()
1070 # Not a candidate for smooth up date (newer non-cruft version in unstable)
1071 if binary in binaries_s[parch]:
1072 if binaries_s[parch][binary].source_version == source_data.version:
1073 continue
1074 cruftbins.add(binaries_s[parch][binary].pkg_id)
1076 # Maybe a candidate (cruft or removed binary): check if config allows us to smooth update it.
1077 if is_smooth_update_allowed(binaries_t[parch][binary], smooth_updates, hints):
1078 # if the package has reverse-dependencies which are
1079 # built from other sources, it's a valid candidate for
1080 # a smooth update. if not, it may still be a valid
1081 # candidate if one if its r-deps is itself a candidate,
1082 # so note it for checking later
1083 #
1084 # We ignore all binaries listed in "removals" as we
1085 # assume they will leave at the same time as the
1086 # given package.
1087 rdeps = {
1088 x
1089 for x in pkg_universe.reverse_dependencies_of(check_pkg_id)
1090 if x not in removals and x not in binaries_to_check_set
1091 }
1093 smooth_update_it = False
1094 if target_suite.any_of_these_are_in_the_suite(rdeps):
1095 for rdep in rdeps:
1096 # each dependency clause has a set of possible
1097 # alternatives that can satisfy that dependency.
1098 # if any of them is outside the set of smoothbins, the
1099 # dependency can be satisfied even if this binary was
1100 # removed, so there is no need to keep it around for a
1101 # smooth update
1102 # if not, only this binary can satisfy the dependency, so
1103 # we should keep it around until the rdep is no longer in
1104 # testing
1105 for dep_clause in pkg_universe.dependencies_of(rdep):
1106 # filter out cruft binaries from unstable, because
1107 # they will not be added to the set of packages that
1108 # will be migrated
1109 if all(
1110 x in smoothbins or x == check_pkg_id
1111 for x in dep_clause
1112 if x not in cruftbins
1113 ):
1114 smoothbins.add(check_pkg_id)
1115 smooth_update_it = True
1116 break
1118 if not smooth_update_it:
1119 check.add(check_pkg_id)
1121 # check whether we should perform a smooth update for
1122 # packages which are candidates but do not have r-deps
1123 # outside of the current source
1124 while 1:
1125 found_any = False
1126 for candidate_pkg_id in check:
1127 rdeps = pkg_universe.reverse_dependencies_of(candidate_pkg_id)
1128 if not rdeps.isdisjoint(smoothbins):
1129 smoothbins.add(candidate_pkg_id)
1130 found_any = True
1131 if not found_any:
1132 break
1133 check = {x for x in check if x not in smoothbins}
1135 return smoothbins
1138def find_newer_binaries(
1139 suite_info: Suites, pkg: BinaryPackage, add_source_for_dropped_bin: bool = False
1140) -> Iterator[tuple[PackageId, Suite]]:
1141 """
1142 Find newer binaries for pkg in any of the source suites.
1144 :param pkg: BinaryPackage (is assumed to be in the target suite)
1146 :param add_source_for_dropped_bin: If True, newer versions of the
1147 source of pkg will be added if they don't have the binary pkg
1149 :return: the newer binaries (or sources) and their suites
1150 """
1151 source = pkg.source
1152 for suite in suite_info:
1153 if suite.suite_class is SuiteClass.TARGET_SUITE:
1154 continue
1156 suite_binaries_on_arch = suite.binaries.get(pkg.pkg_id.architecture)
1157 if not suite_binaries_on_arch: 1157 ↛ 1158line 1157 didn't jump to line 1158 because the condition on line 1157 was never true
1158 continue
1160 newerbin = None
1161 if pkg.pkg_id.package_name in suite_binaries_on_arch:
1162 newerbin = suite_binaries_on_arch[pkg.pkg_id.package_name]
1163 if suite.is_cruft(newerbin):
1164 # We pretend the cruft binary doesn't exist.
1165 # We handle this as if the source didn't have the binary
1166 # (see below)
1167 newerbin = None
1168 elif apt_pkg.version_compare(newerbin.version, pkg.version) <= 0:
1169 continue
1170 else:
1171 if source not in suite.sources:
1172 # bin and source not in suite: no newer version
1173 continue
1175 if not newerbin:
1176 if not add_source_for_dropped_bin: 1176 ↛ 1177line 1176 didn't jump to line 1177 because the condition on line 1176 was never true
1177 continue
1178 # We only get here if there is a newer version of the source,
1179 # which doesn't have the binary anymore (either it doesn't
1180 # exist, or it's cruft and we pretend it doesn't exist).
1181 # Add the new source instead.
1182 nsrc = suite.sources[source]
1183 n_id = PackageId(source, nsrc.version, "source")
1184 overs = pkg.source_version
1185 if apt_pkg.version_compare(nsrc.version, overs) <= 0:
1186 continue
1187 else:
1188 n_id = newerbin.pkg_id
1190 yield (n_id, suite)
1193def parse_provides(
1194 provides_raw: str,
1195 pkg_id: BinaryPackageId | None = None,
1196 logger: logging.Logger | None = None,
1197) -> list[tuple[str, str, str]]:
1198 parts = apt_pkg.parse_depends(provides_raw, False)
1199 nprov = []
1200 for or_clause in parts:
1201 if len(or_clause) != 1: # pragma: no cover
1202 if logger is not None:
1203 msg = "Ignoring invalid provides in %s: Alternatives [%s]"
1204 logger.warning(msg, pkg_id, or_clause)
1205 continue
1206 for part in or_clause:
1207 provided, provided_version, op = part
1208 if op != "" and op != "=": # pragma: no cover
1209 if logger is not None:
1210 msg = "Ignoring invalid provides in %s: %s (%s %s)"
1211 logger.warning(msg, pkg_id, provided, op, provided_version)
1212 continue
1213 provided = sys.intern(provided)
1214 provided_version = sys.intern(provided_version)
1215 part = (provided, provided_version, sys.intern(op))
1216 nprov.append(part)
1217 return nprov
1220def parse_builtusing(
1221 builtusing_raw: str,
1222 pkg_id: BinaryPackageId | None = None,
1223 logger: logging.Logger | None = None,
1224) -> list[tuple[str, str]]:
1225 parts = apt_pkg.parse_depends(builtusing_raw, False)
1226 nbu = []
1227 for or_clause in parts:
1228 if len(or_clause) != 1: # pragma: no cover
1229 if logger is not None:
1230 msg = "Ignoring invalid builtusing in %s: Alternatives [%s]"
1231 logger.warning(msg, pkg_id, or_clause)
1232 continue
1233 for part in or_clause:
1234 bu, bu_version, op = part
1235 if op != "=": # pragma: no cover
1236 if logger is not None:
1237 msg = "Ignoring invalid builtusing in %s: %s (%s %s)"
1238 logger.warning(msg, pkg_id, bu, op, bu_version)
1239 continue
1240 bu = sys.intern(bu)
1241 bu_version = sys.intern(bu_version)
1242 nbu.append((bu, bu_version))
1243 return nbu
1246def parse_option(
1247 options: "optparse.Values",
1248 option_name: str,
1249 default: Any | None = None,
1250 to_bool: bool = False,
1251 to_int: bool = False,
1252 day_to_sec: bool = False,
1253) -> None:
1254 """Ensure the option exist and has a sane value
1256 :param options: dict with options
1258 :param option_name: string with the name of the option
1260 :param default: the default value for the option
1262 :param to_int: convert the input to int (defaults to sys.maxsize)
1264 :param to_bool: convert the input to bool
1266 :param day_to_sec: convert the input from days to seconds (implies to_int=True)
1267 """
1268 value = getattr(options, option_name, default)
1270 # Option was provided with no value (or default is '') so pick up the default
1271 if value == "":
1272 value = default
1274 if (to_int or day_to_sec) and value in (None, ""):
1275 value = sys.maxsize
1277 if day_to_sec:
1278 value = int(float(value) * 24 * 60 * 60) # type: ignore[arg-type]
1280 if to_int:
1281 value = int(value) # type: ignore[arg-type]
1283 if to_bool:
1284 if value and (
1285 isinstance(value, bool) or value.lower() in ("yes", "y", "true", "t", "1")
1286 ):
1287 value = True
1288 else:
1289 value = False
1291 setattr(options, option_name, value)
1294def filter_out_faux_gen(
1295 binaries: Iterable[BinaryPackageId],
1296) -> Iterator[BinaryPackageId]:
1297 """Generator for packages without faux packages"""
1299 for pkg in binaries:
1300 if not pkg.package_name.endswith("-faux-build-depends"):
1301 yield pkg
1304def filter_out_faux(binaries: Iterable[BinaryPackageId]) -> set[BinaryPackageId]:
1305 """Returns a set without faux packages"""
1307 return {pkg for pkg in filter_out_faux_gen(binaries)}
1310def binaries_from_source_version(
1311 source_data: SourcePackage, suite_info: Suites
1312) -> tuple[set[BinaryPackageId], str]:
1313 """Returns a set of real bid with only packages from this source version"""
1315 binaries = source_data.binaries.copy()
1316 # We don't know from which suite the source version comes
1317 for suite in suite_info.source_suites: 1317 ↛ 1328line 1317 didn't jump to line 1328 because the loop on line 1317 didn't complete
1318 # But if it's there, we assume it will have all the associated binaries
1319 if source_data.source in suite.sources: 1319 ↛ 1317line 1319 didn't jump to line 1317 because the condition on line 1319 was always true
1320 for bid in binaries.copy():
1321 if (
1322 suite.all_binaries_in_suite[bid].source_version
1323 != source_data.version
1324 ):
1325 binaries.remove(bid)
1326 break
1328 return filter_out_faux(binaries), suite.name
1331def get_component(section: str) -> str:
1332 """Returns the component based on the section"""
1334 # horrible hard-coding, but currently, we don't keep track of the component
1335 # when loading the packages files, but let's centralize it here
1336 component = "main"
1337 if "/" in section:
1338 component = section.split("/")[0]
1339 return component