Coverage for britney2/inputs/suiteloader.py: 92%
291 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +0000
1import logging
2import optparse
3import os
4import sys
5from abc import abstractmethod
6from collections.abc import Callable, Iterable, Iterator
7from itertools import chain
8from typing import Any, Literal, TypeVar, overload
10import apt_pkg
12from britney2 import (
13 BinaryPackage,
14 BinaryPackageId,
15 MultiArch,
16 SourcePackage,
17 Suite,
18 SuiteClass,
19 Suites,
20 TargetSuite,
21)
22from britney2.utils import (
23 create_provides_map,
24 parse_builtusing,
25 parse_provides,
26 possibly_compressed,
27 read_release_file,
28 read_sources_file,
29)
32class MissingRequiredConfigurationError(RuntimeError):
33 pass
36_T = TypeVar("_T")
39class SuiteContentLoader:
40 def __init__(self, base_config: optparse.Values) -> None:
41 self._base_config = base_config
42 self._architectures: list[str] = SuiteContentLoader.config_str_as_list(
43 base_config.architectures
44 )
45 self._nobreakall_arches: list[str] = SuiteContentLoader.config_str_as_list(
46 base_config.nobreakall_arches, []
47 )
48 self._outofsync_arches: list[str] = SuiteContentLoader.config_str_as_list(
49 base_config.outofsync_arches, []
50 )
51 self._break_arches: list[str] = SuiteContentLoader.config_str_as_list(
52 base_config.break_arches, []
53 )
54 self._new_arches: list[str] = SuiteContentLoader.config_str_as_list(
55 base_config.new_arches, []
56 )
57 self._components: list[str] = []
58 self._all_binaries: dict[BinaryPackageId, BinaryPackage] = {}
59 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__))
60 self.logger = logging.getLogger(logger_name)
62 @overload
63 @staticmethod
64 def config_str_as_list(value: Literal[None], default_value: _T) -> _T: ... 64 ↛ exitline 64 didn't return from function 'config_str_as_list' because
66 @overload
67 @staticmethod
68 def config_str_as_list(value: str, default_value: Any) -> list[str]: ... 68 ↛ exitline 68 didn't return from function 'config_str_as_list' because
70 @overload
71 @staticmethod
72 def config_str_as_list(value: Any, default_value: Any | None = None) -> Any: ... 72 ↛ exitline 72 didn't return from function 'config_str_as_list' because
74 @staticmethod
75 def config_str_as_list(value: Any, default_value: Any | None = None) -> Any:
76 if value is None:
77 return default_value
78 if isinstance(value, str): 78 ↛ 80line 78 didn't jump to line 80 because the condition on line 78 was always true
79 return value.split()
80 return value
82 @property
83 def architectures(self) -> list[str]:
84 return self._architectures
86 @property
87 def nobreakall_arches(self) -> list[str]:
88 return self._nobreakall_arches
90 @property
91 def outofsync_arches(self) -> list[str]:
92 return self._outofsync_arches
94 @property
95 def break_arches(self) -> list[str]:
96 return self._break_arches
98 @property
99 def new_arches(self) -> list[str]:
100 return self._new_arches
102 @property
103 def components(self) -> list[str]:
104 return self._components
106 def all_binaries(self) -> dict[BinaryPackageId, BinaryPackage]:
107 return self._all_binaries
109 @abstractmethod
110 def load_suites(self) -> Suites: # pragma: no cover
111 pass
114def _strip_alternatives_from_depends(
115 deps: str | None, architecture: str = ""
116) -> Iterator[str]:
117 if deps is None:
118 return
120 for block in apt_pkg.parse_src_depends(deps, architecture=architecture):
121 # Like the buildds, we don't care about alternatives
122 first_block = block[0]
123 if first_block[1] != "":
124 # The extra space in the middle is a workaround for
125 # an apt_pkg bug in bookworm
126 yield f"{first_block[0]} ({first_block[2]} {first_block[1]})"
127 else:
128 yield f"{first_block[0]}"
131class DebMirrorLikeSuiteContentLoader(SuiteContentLoader):
132 CHECK_FIELDS = (
133 "source",
134 "source_version",
135 "architecture",
136 "multi_arch",
137 "depends",
138 "conflicts",
139 "provides",
140 )
142 def load_suites(self) -> Suites:
143 suites = []
144 missing_config_msg = (
145 "Configuration %s is not set in the config (and cannot be auto-detected)"
146 )
147 target_suite = None
148 for suitename in ("testing", "unstable", "pu", "tpu"):
149 suffix = suitename if suitename in ("pu", "tpu") else ""
150 if hasattr(self._base_config, suitename):
151 suite_path = getattr(self._base_config, suitename)
152 suite_class = SuiteClass.TARGET_SUITE
153 if suitename != "testing":
154 suite_class = (
155 SuiteClass.ADDITIONAL_SOURCE_SUITE
156 if suffix
157 else SuiteClass.PRIMARY_SOURCE_SUITE
158 )
159 suites.append(
160 Suite(
161 suite_class, suitename, suite_path, suite_short_name=suffix
162 )
163 )
164 else:
165 target_suite = TargetSuite(
166 suite_class, suitename, suite_path, suite_short_name=suffix
167 )
168 else:
169 if suitename in ("testing", "unstable"): # pragma: no cover
170 self.logger.error(missing_config_msg, suitename.upper())
171 raise MissingRequiredConfigurationError(
172 missing_config_msg % suitename.upper()
173 )
174 self.logger.info(
175 "Optional suite %s is not defined (config option: %s) ",
176 suitename,
177 suitename.upper(),
178 )
180 assert target_suite is not None, "Logic regression, this should be impossible."
182 self._check_release_file(target_suite, missing_config_msg)
183 self._setup_architectures()
185 # read the source and binary packages for the involved distributions. Notes:
186 # - Load testing last as some live-data tests have more complete information in
187 # unstable
188 # - Load all sources before any of the binaries.
189 for suite in chain((target_suite,), suites):
190 sources = self._read_sources(
191 suite.path, None if suite is target_suite else target_suite.sources
192 )
193 self._update_suite_name(suite)
194 suite.sources = sources
195 (suite.binaries, suite.provides_table) = self._read_binaries(
196 suite, self._architectures
197 )
198 self._fixup_faux_arch_all_binaries(suite)
199 if self._base_config.be_strict_with_build_deps: 199 ↛ 189line 199 didn't jump to line 189 because the condition on line 199 was always true
200 self._add_build_dep_faux_binaries(suite)
202 return Suites(target_suite, suites)
204 def _fixup_faux_arch_all_binaries(self, suite: Suite) -> None:
205 """remove faux arch:all binary if a real arch:all binary is available
207 We don't know for which architectures bin/$something must be available
208 except for arch:all, which should be available in each arch. The
209 information that a source builds an arch:all binary is available during
210 the loading of the sources, but we have to pick an order in which to
211 load the files and the Sources is loaded before the Packages are
212 read. Hence we fake an arch:all binary during source loading, but it
213 shouldn't be there in the final list if real arch:all binaries are
214 present in the Packages file.
216 Also, if we keep the fake binary, it should be added to the lists of
217 known binaries in the suite, otherwise britney2 trips later on.
219 """
221 all_binaries = self._all_binaries
222 binaries = suite.binaries
223 faux_arches = (
224 set(self.architectures)
225 - set(self.break_arches)
226 - set(self.outofsync_arches)
227 - set(self.new_arches)
228 )
230 for srcpkg in suite.sources.values():
231 faux = {x for x in srcpkg.binaries if x.architecture == "faux"}
232 if faux and any(
233 x
234 for x in (srcpkg.binaries - faux)
235 if all_binaries[x].architecture == "all"
236 ):
237 srcpkg.binaries -= faux
239 # Calculate again because we may have changed the set
240 faux = {x for x in srcpkg.binaries if x.architecture == "faux"}
241 for binpkg_id in faux:
242 bin_data = BinaryPackage(
243 "faux",
244 srcpkg.source,
245 srcpkg.version,
246 "all",
247 MultiArch.NO,
248 None,
249 None,
250 None,
251 False,
252 binpkg_id,
253 None,
254 )
255 for arch_all in faux_arches:
256 binaries[arch_all][binpkg_id.package_name] = bin_data
257 all_binaries[binpkg_id] = bin_data
258 suite.binaries = binaries
260 def _add_build_dep_faux_binaries(self, suite: Suite) -> None:
261 """Add faux packages that keep track of build depends
263 To ensure that Build-Depends are fully protected against inappropriate
264 removal or upgrade, we add faux packages to source packages containing
265 the Build-Depends as Depends.
266 """
268 all_binaries = self._all_binaries
269 for src_name, src_pkg in suite.sources.items():
270 # TODO: something with arch, which one?
271 archall = self._nobreakall_arches[0]
272 bd_pid = BinaryPackageId(
273 f"{src_name}-faux-build-depends", sys.intern(src_pkg.version), archall
274 )
275 deps = ",".join(
276 chain(
277 _strip_alternatives_from_depends(src_pkg.build_deps_arch, archall),
278 _strip_alternatives_from_depends(src_pkg.build_deps_indep, archall),
279 )
280 )
281 if deps == "":
282 continue
283 dpkg = BinaryPackage(
284 "faux",
285 src_name,
286 src_pkg.version,
287 archall,
288 MultiArch.NO,
289 deps,
290 None,
291 None,
292 False,
293 bd_pid,
294 None,
295 )
296 suite.binaries.setdefault(archall, {})[bd_pid.package_name] = dpkg
297 src_pkg.binaries.add(bd_pid)
298 all_binaries[bd_pid] = dpkg
300 def _setup_architectures(self) -> None:
301 allarches = self._architectures
302 # Re-order the architectures such as that the most important architectures are listed first
303 # (this is to make the log easier to read as most important architectures will be listed
304 # first)
305 arches = [x for x in allarches if x in self._nobreakall_arches]
306 arches += [
307 x for x in allarches if x not in arches and x not in self._outofsync_arches
308 ]
309 arches += [
310 x for x in allarches if x not in arches and x not in self._break_arches
311 ]
312 arches += [
313 x for x in allarches if x not in arches and x not in self._new_arches
314 ]
315 arches += [x for x in allarches if x not in arches]
317 # Intern architectures for efficiency; items in this list will be used for lookups and
318 # building items/keys - by intern strings we reduce memory (considerably).
319 self._architectures = [sys.intern(arch) for arch in allarches]
320 assert "all" not in self._architectures, "all not allowed in architectures"
322 def _get_suite_name(
323 self, suite: Suite, release_file: "apt_pkg.TagSection[str]"
324 ) -> tuple[str, str]:
325 name = None
326 codename = None
327 if "Suite" in release_file: 327 ↛ 329line 327 didn't jump to line 329 because the condition on line 327 was always true
328 name = release_file["Suite"]
329 if "Codename" in release_file:
330 codename = release_file["Codename"]
332 if name is None: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 name = codename
334 elif codename is None:
335 codename = name
337 if name is None: 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 self.logger.warning(
339 'Either of the fields "Suite" or "Codename" '
340 + "should be present in a release file."
341 )
342 self.logger.error(
343 "Release file for suite %s is missing both the "
344 + '"Suite" and the "Codename" fields.',
345 suite.name,
346 )
347 raise KeyError("Suite")
349 assert codename is not None # required for type checking
350 return (name, codename)
352 def _update_suite_name(self, suite: Suite) -> None:
353 try:
354 release_file = read_release_file(suite.path)
355 except FileNotFoundError:
356 self.logger.info(
357 "The %s suite does not have a Release file, unable to update the name",
358 suite.name,
359 )
360 release_file = None
362 if release_file is not None:
363 (suite.name, suite.codename) = self._get_suite_name(suite, release_file)
364 self.logger.info("Using suite name from Release file: %s", suite.name)
365 self.logger.debug(
366 "Using suite codename from Release file: %s", suite.codename
367 )
369 def _check_release_file(self, target_suite: Suite, missing_config_msg: str) -> None:
370 try:
371 release_file = read_release_file(target_suite.path)
372 self.logger.info(
373 "Found a Release file in %s - using that for defaults",
374 target_suite.name,
375 )
376 except FileNotFoundError:
377 self.logger.info(
378 "The %s suite does not have a Release file.", target_suite.name
379 )
380 release_file = None
382 if release_file is not None:
383 self._components = release_file["Components"].split()
384 self.logger.info(
385 "Using components listed in Release file: %s",
386 " ".join(self._components),
387 )
389 if self._architectures is None:
390 if release_file is None: # pragma: no cover
391 self.logger.error(
392 "No configured architectures and there is no release file in the %s suite.",
393 target_suite.name,
394 )
395 self.logger.error(
396 'Please check if there is a "Release" file in %s', target_suite.path
397 )
398 self.logger.error(
399 'or if the config file contains a non-empty "ARCHITECTURES" field'
400 )
401 raise MissingRequiredConfigurationError(
402 missing_config_msg % "ARCHITECTURES"
403 )
404 self._architectures = sorted(
405 x for x in release_file["Architectures"].split() if x != "all"
406 )
407 self.logger.info(
408 "Using architectures listed in Release file: %s",
409 " ".join(self._architectures),
410 )
412 def _read_sources(
413 self, basedir: str, sources_target_suite: dict[str, SourcePackage] | None = None
414 ) -> dict[str, SourcePackage]:
415 """Read the list of source packages from the specified directory
417 The source packages are read from the `Sources' file within the
418 directory specified as `basedir' parameter. Considering the
419 large amount of memory needed, not all the fields are loaded
420 in memory. The available fields are Version, Maintainer and Section.
422 The method returns a list where every item represents a source
423 package as a dictionary.
424 """
426 if self._components:
427 sources: dict[str, SourcePackage] = {}
428 for component in self._components:
429 filename = os.path.join(basedir, component, "source", "Sources")
430 try:
431 filename = possibly_compressed(filename)
432 except FileNotFoundError:
433 if component == "non-free-firmware":
434 self.logger.info("Skipping %s as it doesn't exist", filename)
435 continue
436 raise
437 self.logger.info("Loading source packages from %s", filename)
438 read_sources_file(
439 filename,
440 sources,
441 not self._base_config.archall_inconsistency_allowed,
442 sources_target_suite,
443 )
444 else:
445 filename = os.path.join(basedir, "Sources")
446 self.logger.info("Loading source packages from %s", filename)
447 sources = read_sources_file(
448 filename,
449 None,
450 not self._base_config.archall_inconsistency_allowed,
451 sources_target_suite,
452 )
454 return sources
456 @staticmethod
457 def merge_fields(
458 get_field: Callable[[str], str | None],
459 *field_names: str,
460 separator: str = ", ",
461 ) -> str | None:
462 """Merge two or more fields (filtering out empty fields; returning None if all are empty)"""
463 return separator.join(filter(None, (get_field(x) for x in field_names))) or None
465 def _read_packages_file(
466 self,
467 filename: str,
468 arch: str,
469 srcdist: dict[str, SourcePackage],
470 packages: dict[str, BinaryPackage] | None = None,
471 ) -> dict[str, BinaryPackage]:
472 self.logger.info("Loading binary packages from %s", filename)
474 if packages is None:
475 packages = {}
477 all_binaries = self._all_binaries
479 tag_file = apt_pkg.TagFile(filename)
480 get_field = tag_file.section.get
481 step = tag_file.step
482 intern = sys.intern
484 while step():
485 pkg = get_field("Package")
486 version = get_field("Version")
488 # There may be multiple versions of any arch:all packages
489 # (in unstable) if some architectures have out-of-date
490 # binaries. We only ever consider the package with the
491 # largest version for migration.
492 pkg = intern(pkg)
493 version = intern(version)
494 pkg_id = BinaryPackageId(pkg, version, arch)
496 if (old_pkg_data := packages.get(pkg)) is not None:
497 if apt_pkg.version_compare(old_pkg_data.version, version) > 0:
498 continue
499 old_pkg_id = old_pkg_data.pkg_id
500 old_src_binaries = srcdist[old_pkg_data.source].binaries
501 old_src_binaries.remove(old_pkg_id)
502 # This may seem weird at first glance, but the current code rely
503 # on this behaviour to avoid issues like #709460. Admittedly it
504 # is a special case, but Britney will attempt to remove the
505 # arch:all packages without this. Even then, this particular
506 # stop-gap relies on the packages files being sorted by name
507 # and the version, so it is not particularly resilient.
508 if pkg_id not in old_src_binaries: 508 ↛ 514line 508 didn't jump to line 514 because the condition on line 508 was always true
509 old_src_binaries.add(pkg_id)
511 # Merge Pre-Depends with Depends and Conflicts with
512 # Breaks. Britney is not interested in the "finer
513 # semantic differences" of these fields anyway.
514 deps = DebMirrorLikeSuiteContentLoader.merge_fields(
515 get_field, "Pre-Depends", "Depends"
516 )
517 conflicts = DebMirrorLikeSuiteContentLoader.merge_fields(
518 get_field, "Conflicts", "Breaks"
519 )
521 ess = get_field("Essential", "no") == "yes"
523 source = pkg
524 source_version = version
525 # retrieve the name and the version of the source package
526 source_raw = get_field("Source")
527 if source_raw:
528 source = source_raw.split(" ")[0]
529 if "(" in source_raw:
530 source_version = source_raw[
531 source_raw.find("(") + 1 : source_raw.find(")")
532 ]
534 provides_raw = get_field("Provides")
535 if provides_raw:
536 provides = parse_provides(
537 provides_raw, pkg_id=pkg_id, logger=self.logger
538 )
539 else:
540 provides = None
542 raw_arch = get_field("Architecture")
543 if raw_arch not in {"all", arch}: # pragma: no cover
544 raise AssertionError(
545 f"{pkg_id!s} has wrong architecture ({raw_arch}) - should be either {arch} or all"
546 )
548 builtusing_raw = get_field("Built-Using")
549 if builtusing_raw:
550 builtusing = parse_builtusing(
551 builtusing_raw, pkg_id=pkg_id, logger=self.logger
552 )
553 else:
554 builtusing = None
556 dpkg = BinaryPackage(
557 get_field("Section"),
558 source,
559 source_version,
560 raw_arch,
561 MultiArch.from_str(get_field("Multi-Arch")),
562 deps,
563 conflicts,
564 provides,
565 ess,
566 pkg_id,
567 builtusing,
568 )
570 # if the source package is available in the distribution, then register this binary package
571 if (source_pkg := srcdist.get(source)) is not None:
572 # There may be multiple versions of any arch:all packages
573 # (in unstable) if some architectures have out-of-date
574 # binaries. We only want to include the package in the
575 # source -> binary mapping once. It doesn't matter which
576 # of the versions we include as only the package name and
577 # architecture are recorded.
578 source_pkg.binaries.add(pkg_id)
579 # if the source package doesn't exist, create a fake one
580 else:
581 srcdist[source] = SourcePackage(
582 source,
583 source_version,
584 "faux",
585 {pkg_id},
586 None,
587 True,
588 )
590 # add the resulting dictionary to the package list
591 if (old_binary := all_binaries.get(pkg_id)) is not None:
592 # If the binary package is the same in across suites, we reuse
593 # existing BinaryPackage instances to reduce memory usage.
594 self._merge_pkg_entries(pkg, arch, old_binary, dpkg)
595 packages[pkg] = old_binary
596 else:
597 packages[pkg] = dpkg
598 all_binaries[pkg_id] = dpkg
600 return packages
602 def _read_binaries(
603 self, suite: Suite, architectures: Iterable[str]
604 ) -> tuple[
605 dict[str, dict[str, BinaryPackage]], dict[str, dict[str, set[tuple[str, str]]]]
606 ]:
607 """Read the list of binary packages from the specified directory
609 This method reads all the binary packages for a given suite.
611 If the "components" config parameter is set, the directory should
612 be the "suite" directory of a local mirror (i.e. the one containing
613 the "Release" file). Otherwise, Britney will read the packages
614 information from all the "Packages_${arch}" files referenced by
615 the "architectures" parameter.
617 Considering the
618 large amount of memory needed, not all the fields are loaded
619 in memory. The available fields are Version, Source, Multi-Arch,
620 Depends, Conflicts, Provides and Architecture.
622 The `Provides' field is used to populate the virtual packages list.
624 The method returns a tuple of two dicts with architecture as key and
625 another dict as value. The value dicts of the first dict map
626 from binary package name to "BinaryPackage" objects; the other second
627 value dicts map a package name to the packages providing them.
628 """
629 binaries: dict[str, dict[str, BinaryPackage]] = {}
630 provides_table: dict[str, dict[str, set[tuple[str, str]]]] = {}
631 basedir = suite.path
633 if self._components:
634 release_file = read_release_file(basedir)
635 listed_archs = set(release_file["Architectures"].split())
636 for arch in architectures:
637 packages: dict[str, BinaryPackage] = {}
638 if arch not in listed_archs: 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true
639 self.logger.info(
640 "Skipping arch %s for %s: It is not listed in the Release file",
641 arch,
642 suite.name,
643 )
644 binaries[arch] = {}
645 provides_table[arch] = {}
646 continue
647 for component in self._components:
648 binary_dir = f"binary-{arch}"
649 filename = os.path.join(basedir, component, binary_dir, "Packages")
650 try:
651 filename = possibly_compressed(filename)
652 except FileNotFoundError:
653 if component == "non-free-firmware":
654 self.logger.info(
655 "Skipping %s as it doesn't exist", filename
656 )
657 continue
658 raise
659 udeb_filename = os.path.join(
660 basedir, component, "debian-installer", binary_dir, "Packages"
661 )
662 # We assume the udeb Packages file is present if the
663 # regular one is present
664 udeb_filename = possibly_compressed(udeb_filename)
665 self._read_packages_file(filename, arch, suite.sources, packages)
666 self._read_packages_file(
667 udeb_filename, arch, suite.sources, packages
668 )
669 # create provides
670 provides = create_provides_map(packages)
671 binaries[arch] = packages
672 provides_table[arch] = provides
673 else:
674 for arch in architectures:
675 filename = os.path.join(basedir, f"Packages_{arch}")
676 packages = self._read_packages_file(filename, arch, suite.sources)
677 provides = create_provides_map(packages)
678 binaries[arch] = packages
679 provides_table[arch] = provides
681 return (binaries, provides_table)
683 def _merge_pkg_entries(
684 self,
685 package: str,
686 parch: str,
687 pkg_entry1: BinaryPackage,
688 pkg_entry2: BinaryPackage,
689 ) -> None:
690 bad = []
691 for f in self.CHECK_FIELDS:
692 v1 = getattr(pkg_entry1, f)
693 v2 = getattr(pkg_entry2, f)
694 if v1 != v2: # pragma: no cover
695 bad.append((f, v1, v2))
697 if bad: # pragma: no cover
698 self.logger.error(
699 "Mismatch found %s %s %s differs", package, pkg_entry1.version, parch
700 )
701 for f, v1, v2 in bad:
702 self.logger.info(" ... %s %s != %s", f, v1, v2)
703 raise ValueError("Inconsistent / Unsupported data set")
705 # Merge ESSENTIAL if necessary
706 assert pkg_entry1.is_essential or not pkg_entry2.is_essential