Coverage for src/debputy/highlevel_manifest.py: 63%
904 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-22 10:58 +0000
1import dataclasses
2import functools
3import os
4import textwrap
5import typing
6from contextlib import suppress
7from dataclasses import dataclass, field
8from typing import (
9 Any,
10 TypeVar,
11 Generic,
12 cast,
13)
14from collections.abc import Iterable, Mapping, Sequence, Callable
16from debian.debian_support import DpkgArchTable
18from debputy.dh.debhelper_emulation import (
19 dhe_dbgsym_root_dir,
20 assert_no_dbgsym_migration,
21 read_dbgsym_file,
22)
23from ._deb_options_profiles import DebBuildOptionsAndProfiles
24from ._manifest_constants import *
25from .architecture_support import DpkgArchitectureBuildProcessValuesTable
26from .builtin_manifest_rules import builtin_mode_normalization_rules
27from .exceptions import (
28 DebputySubstitutionError,
29 DebputyRuntimeErrorWithPreamble,
30)
31from .filesystem_scan import (
32 InMemoryVirtualRootDir,
33 OSFSROOverlay,
34 FSControlRootDir,
35 InMemoryVirtualPathBase,
36 InMemoryOverlayFSRootDirectory,
37)
38from .installations import (
39 InstallRule,
40 SourcePathMatcher,
41 PathAlreadyInstalledOrDiscardedError,
42 NoMatchForInstallPatternError,
43 InstallRuleContext,
44 BinaryPackageInstallRuleContext,
45 InstallSearchDirContext,
46 SearchDir,
47)
48from .intermediate_manifest import TarMember, PathType, IntermediateManifest
49from .maintscript_snippet import (
50 DpkgMaintscriptHelperCommand,
51 PackageMaintscriptSnippetContainer,
52)
53from .manifest_conditions import ConditionContext
54from .manifest_parser.base_types import (
55 FileSystemMatchRule,
56 FileSystemExactMatchRule,
57 BuildEnvironments,
58)
59from .manifest_parser.util import AttributePath
60from .packager_provided_files import PackagerProvidedFile
61from .packages import BinaryPackage, SourcePackage
62from .plugin.api.feature_set import PluginProvidedFeatureSet
63from .plugin.api.impl import BinaryCtrlAccessorProviderCreator
64from .plugin.api.impl_types import (
65 PackageProcessingContextProvider,
66 PackageDataTable,
67)
68from .plugin.api.spec import (
69 FlushableSubstvars,
70 VirtualPath,
71 DebputyIntegrationMode,
72 INTEGRATION_MODE_DH_DEBPUTY_RRR,
73 INTEGRATION_MODE_FULL,
74)
75from debputy.plugins.debputy.binary_package_rules import ServiceRule
76from debputy.plugins.debputy.build_system_rules import BuildRule
77from .plugin.plugin_state import run_in_context_of_plugin
78from .substitution import Substitution
79from .transformation_rules import (
80 TransformationRule,
81 ModeNormalizationTransformationRule,
82 NormalizeShebangLineTransformation,
83)
84from .util import (
85 _error,
86 _warn,
87 debian_policy_normalize_symlink_target,
88 generated_content_dir,
89 _info,
90 assume_not_none,
91)
92from .yaml import MANIFEST_YAML
93from .yaml.compat import CommentedMap, CommentedSeq
96def tar_path(p: VirtualPath) -> str:
97 path = p.path
98 if p.is_dir:
99 return path + "/"
100 return path
103def tar_owner_info(path: InMemoryVirtualPathBase) -> tuple[str, int, str, int]:
104 owner = path._owner # noqa
105 group = path._group # noqa
106 return (
107 owner.entity_name,
108 owner.entity_id,
109 group.entity_name,
110 group.entity_id,
111 )
114class PathNotCoveredByInstallRulesError(DebputyRuntimeErrorWithPreamble):
116 @property
117 def unmatched_paths(self) -> Sequence[VirtualPath]:
118 return self.args[1]
120 @property
121 def search_dir(self) -> VirtualPath:
122 return self.args[2]
124 def render_preamble(self) -> None:
125 _warn(
126 f"The following paths were present in {self.search_dir.fs_path}, but not installed (nor explicitly discarded)."
127 )
128 _warn("")
129 for entry in self.unmatched_paths:
130 desc = _describe_missing_path(entry)
131 _warn(f" * {desc}")
132 _warn("")
135@dataclass(slots=True)
136class DbgsymInfo:
137 binary_package: BinaryPackage
138 dbgsym_fs_root: InMemoryVirtualPathBase
139 _dbgsym_root_fs: str | None
140 dbgsym_ids: list[str]
141 run_dwz: bool
143 @property
144 def dbgsym_root_dir(self) -> str:
145 root_dir = self._dbgsym_root_fs
146 if root_dir is None:
147 root_dir = generated_content_dir(
148 package=self.binary_package,
149 subdir_key="dbgsym-fs-root",
150 )
151 self._dbgsym_root_fs = root_dir
152 return root_dir
154 @property
155 def dbgsym_ctrl_dir(self) -> FSControlRootDir:
156 return FSControlRootDir.create_root_dir(
157 os.path.join(self.dbgsym_root_dir, "DEBIAN")
158 )
161@dataclass(slots=True, frozen=True)
162class BinaryPackageData:
163 source_package: SourcePackage
164 binary_package: BinaryPackage
165 binary_staging_root_dir: str
166 fs_root: InMemoryVirtualPathBase
167 substvars: FlushableSubstvars
168 package_metadata_context: PackageProcessingContextProvider
169 ctrl_creator: BinaryCtrlAccessorProviderCreator
170 dbgsym_info: DbgsymInfo
172 @property
173 def control_output_dir(self) -> FSControlRootDir:
174 return FSControlRootDir.create_root_dir(
175 generated_content_dir(
176 package=self.binary_package,
177 subdir_key="DEBIAN",
178 )
179 )
182@dataclass(slots=True)
183class PackageTransformationDefinition:
184 binary_package: BinaryPackage
185 substitution: Substitution
186 is_auto_generated_package: bool
187 binary_version: str | None = None
188 search_dirs: list[FileSystemExactMatchRule] | None = None
189 dpkg_maintscript_helper_snippets: list[DpkgMaintscriptHelperCommand] = field(
190 default_factory=list
191 )
192 maintscript_snippets: PackageMaintscriptSnippetContainer = field(
193 default_factory=PackageMaintscriptSnippetContainer
194 )
195 transformations: list[TransformationRule] = field(default_factory=list)
196 reserved_packager_provided_files: dict[str, list[PackagerProvidedFile]] = field(
197 default_factory=dict
198 )
199 install_rules: list[InstallRule] = field(default_factory=list)
200 requested_service_rules: list[ServiceRule] = field(default_factory=list)
203def _path_to_tar_member(
204 path: InMemoryVirtualPathBase,
205 clamp_mtime_to: int,
206) -> TarMember:
207 mtime = float(clamp_mtime_to)
208 owner, uid, group, gid = tar_owner_info(path)
209 mode = path.mode
211 if path.has_fs_path:
212 mtime = min(mtime, path.mtime)
214 if path.is_dir:
215 path_type = PathType.DIRECTORY
216 elif path.is_file:
217 # TODO: someday we will need to deal with hardlinks and it might appear here.
218 path_type = PathType.FILE
219 elif path.is_symlink: 219 ↛ 239line 219 didn't jump to line 239 because the condition on line 219 was always true
220 # Special-case that we resolve immediately (since we need to normalize the target anyway)
221 link_target = debian_policy_normalize_symlink_target(
222 path.path,
223 path.readlink(),
224 )
225 return TarMember.virtual_path(
226 tar_path(path),
227 PathType.SYMLINK,
228 mtime,
229 link_target=link_target,
230 # Force mode to be 0777 as that is the mode we see in the data.tar. In theory, tar lets you set
231 # it to whatever. However, for reproducibility, we have to be well-behaved - and that is 0777.
232 mode=0o0777,
233 owner=owner,
234 uid=uid,
235 group=group,
236 gid=gid,
237 )
238 else:
239 assert not path.is_symlink
240 raise AssertionError(
241 f"Unsupported file type: {path.path} - not a file, dir nor a symlink!"
242 )
244 if not path.has_fs_path:
245 assert not path.is_file
246 return TarMember.virtual_path(
247 tar_path(path),
248 path_type,
249 mtime,
250 mode=mode,
251 owner=owner,
252 uid=uid,
253 group=group,
254 gid=gid,
255 )
256 may_steal_fs_path = getattr(path, "_can_replace_inline", False)
257 return TarMember.from_file(
258 tar_path(path),
259 path.fs_path,
260 mode=mode,
261 uid=uid,
262 owner=owner,
263 gid=gid,
264 group=group,
265 path_type=path_type,
266 path_mtime=mtime,
267 clamp_mtime_to=clamp_mtime_to,
268 may_steal_fs_path=may_steal_fs_path,
269 )
272def _generate_intermediate_manifest(
273 fs_root: InMemoryVirtualPathBase,
274 clamp_mtime_to: int,
275) -> Iterable[TarMember]:
276 symlinks = []
277 for path in fs_root.all_paths():
278 tar_member = _path_to_tar_member(path, clamp_mtime_to)
279 if tar_member.path_type == PathType.SYMLINK:
280 symlinks.append(tar_member)
281 continue
282 yield tar_member
283 yield from symlinks
286ST = TypeVar("ST")
287T = TypeVar("T")
290class AbstractYAMLSubStore(Generic[ST]):
291 def __init__(
292 self,
293 parent_store: Any,
294 parent_key: int | str | None,
295 store: ST | None = None,
296 ) -> None:
297 if parent_store is not None and parent_key is not None:
298 try:
299 from_parent_store = parent_store[parent_key]
300 except (KeyError, IndexError):
301 from_parent_store = None
302 if ( 302 ↛ 307line 302 didn't jump to line 307 because the condition on line 302 was never true
303 store is not None
304 and from_parent_store is not None
305 and store is not parent_store
306 ):
307 raise ValueError(
308 "Store is provided but is not the one already in the parent store"
309 )
310 if store is None: 310 ↛ 312line 310 didn't jump to line 312 because the condition on line 310 was always true
311 store = from_parent_store
312 self._parent_store = parent_store
313 self._parent_key = parent_key
314 self._is_detached = (
315 parent_key is None or parent_store is None or parent_key not in parent_store
316 )
317 assert self._is_detached or store is not None
318 if store is None:
319 store = self._create_new_instance()
320 self._store: ST = store
322 def _create_new_instance(self) -> ST:
323 raise NotImplementedError
325 def create_definition_if_missing(self) -> None:
326 if self._is_detached:
327 self.create_definition()
329 def create_definition(self) -> None:
330 if not self._is_detached: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 raise RuntimeError("Definition is already present")
332 parent_store = self._parent_store
333 if parent_store is None: 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 raise RuntimeError(
335 f"Definition is not attached to any parent!? ({self.__class__.__name__})"
336 )
337 if isinstance(parent_store, list):
338 assert self._parent_key is None
339 self._parent_key = len(parent_store)
340 self._parent_store.append(self._store)
341 else:
342 parent_store[self._parent_key] = self._store
343 self._is_detached = False
345 def remove_definition(self) -> None:
346 self._ensure_attached()
347 del self._parent_store[self._parent_key]
348 if isinstance(self._parent_store, list):
349 self._parent_key = None
350 self._is_detached = True
352 def _ensure_attached(self) -> None:
353 if self._is_detached:
354 raise RuntimeError("The definition has been removed!")
357class AbstractYAMLListSubStore(Generic[T], AbstractYAMLSubStore[list[T]]):
358 def _create_new_instance(self) -> list[T]:
359 return CommentedSeq()
362class AbstractYAMLDictSubStore(Generic[T], AbstractYAMLSubStore[dict[str, T]]):
363 def _create_new_instance(self) -> dict[str, T]:
364 return CommentedMap()
367class MutableCondition:
368 @classmethod
369 def arch_matches(cls, arch_filter: str) -> CommentedMap:
370 return CommentedMap({MK_CONDITION_ARCH_MATCHES: arch_filter})
372 @classmethod
373 def build_profiles_matches(cls, build_profiles_matches: str) -> CommentedMap:
374 return CommentedMap(
375 {MK_CONDITION_BUILD_PROFILES_MATCHES: build_profiles_matches}
376 )
379class MutableYAMLSymlink(AbstractYAMLDictSubStore[Any]):
380 @classmethod
381 def new_symlink(
382 cls, link_path: str, link_target: str, condition: Any | None
383 ) -> "MutableYAMLSymlink":
384 inner = {
385 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_PATH: link_path,
386 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_TARGET: link_target,
387 }
388 content = {MK_TRANSFORMATIONS_CREATE_SYMLINK: inner}
389 if condition is not None: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 inner["when"] = condition
391 return cls(None, None, store=CommentedMap(content))
393 @property
394 def symlink_path(self) -> str:
395 return self._store[MK_TRANSFORMATIONS_CREATE_SYMLINK][
396 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_PATH
397 ]
399 @symlink_path.setter
400 def symlink_path(self, path: str) -> None:
401 self._store[MK_TRANSFORMATIONS_CREATE_SYMLINK][
402 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_PATH
403 ] = path
405 @property
406 def symlink_target(self) -> str | None:
407 return self._store[MK_TRANSFORMATIONS_CREATE_SYMLINK][
408 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_TARGET
409 ]
411 @symlink_target.setter
412 def symlink_target(self, target: str) -> None:
413 self._store[MK_TRANSFORMATIONS_CREATE_SYMLINK][
414 MK_TRANSFORMATIONS_CREATE_SYMLINK_LINK_TARGET
415 ] = target
418class MutableYAMLConffileManagementItem(AbstractYAMLDictSubStore[Any]):
419 @classmethod
420 def rm_conffile(
421 cls,
422 conffile: str,
423 prior_to_version: str | None,
424 owning_package: str | None,
425 ) -> "MutableYAMLConffileManagementItem":
426 r = cls(
427 None,
428 None,
429 store=CommentedMap(
430 {
431 MK_CONFFILE_MANAGEMENT_REMOVE: CommentedMap(
432 {MK_CONFFILE_MANAGEMENT_REMOVE_PATH: conffile}
433 )
434 }
435 ),
436 )
437 r.prior_to_version = prior_to_version
438 r.owning_package = owning_package
439 return r
441 @classmethod
442 def mv_conffile(
443 cls,
444 old_conffile: str,
445 new_conffile: str,
446 prior_to_version: str | None,
447 owning_package: str | None,
448 ) -> "MutableYAMLConffileManagementItem":
449 r = cls(
450 None,
451 None,
452 store=CommentedMap(
453 {
454 MK_CONFFILE_MANAGEMENT_RENAME: CommentedMap(
455 {
456 MK_CONFFILE_MANAGEMENT_RENAME_SOURCE: old_conffile,
457 MK_CONFFILE_MANAGEMENT_RENAME_TARGET: new_conffile,
458 }
459 )
460 }
461 ),
462 )
463 r.prior_to_version = prior_to_version
464 r.owning_package = owning_package
465 return r
467 @property
468 def _container(self) -> dict[str, Any]:
469 assert len(self._store) == 1
470 return next(iter(self._store.values()))
472 @property
473 def command(self) -> str:
474 assert len(self._store) == 1
475 return next(iter(self._store))
477 @property
478 def obsolete_conffile(self) -> str:
479 if self.command == MK_CONFFILE_MANAGEMENT_REMOVE:
480 return self._container[MK_CONFFILE_MANAGEMENT_REMOVE_PATH]
481 assert self.command == MK_CONFFILE_MANAGEMENT_RENAME
482 return self._container[MK_CONFFILE_MANAGEMENT_RENAME_SOURCE]
484 @obsolete_conffile.setter
485 def obsolete_conffile(self, value: str) -> None:
486 if self.command == MK_CONFFILE_MANAGEMENT_REMOVE:
487 self._container[MK_CONFFILE_MANAGEMENT_REMOVE_PATH] = value
488 else:
489 assert self.command == MK_CONFFILE_MANAGEMENT_RENAME
490 self._container[MK_CONFFILE_MANAGEMENT_RENAME_SOURCE] = value
492 @property
493 def new_conffile(self) -> str:
494 if self.command != MK_CONFFILE_MANAGEMENT_RENAME:
495 raise TypeError(
496 f"The new_conffile attribute is only applicable to command {MK_CONFFILE_MANAGEMENT_RENAME}."
497 f" This is a {self.command}"
498 )
499 return self._container[MK_CONFFILE_MANAGEMENT_RENAME_TARGET]
501 @new_conffile.setter
502 def new_conffile(self, value: str) -> None:
503 if self.command != MK_CONFFILE_MANAGEMENT_RENAME:
504 raise TypeError(
505 f"The new_conffile attribute is only applicable to command {MK_CONFFILE_MANAGEMENT_RENAME}."
506 f" This is a {self.command}"
507 )
508 self._container[MK_CONFFILE_MANAGEMENT_RENAME_TARGET] = value
510 @property
511 def prior_to_version(self) -> str | None:
512 return self._container.get(MK_CONFFILE_MANAGEMENT_X_PRIOR_TO_VERSION)
514 @prior_to_version.setter
515 def prior_to_version(self, value: str | None) -> None:
516 if value is None:
517 try:
518 del self._container[MK_CONFFILE_MANAGEMENT_X_PRIOR_TO_VERSION]
519 except KeyError:
520 pass
521 else:
522 self._container[MK_CONFFILE_MANAGEMENT_X_PRIOR_TO_VERSION] = value
524 @property
525 def owning_package(self) -> str | None:
526 return self._container[MK_CONFFILE_MANAGEMENT_X_PRIOR_TO_VERSION]
528 @owning_package.setter
529 def owning_package(self, value: str | None) -> None:
530 if value is None:
531 try:
532 del self._container[MK_CONFFILE_MANAGEMENT_X_OWNING_PACKAGE]
533 except KeyError:
534 pass
535 else:
536 self._container[MK_CONFFILE_MANAGEMENT_X_OWNING_PACKAGE] = value
539class MutableYAMLPackageDefinition(AbstractYAMLDictSubStore):
540 def _list_store(
541 self, key, *, create_if_absent: bool = False
542 ) -> list[dict[str, Any]] | None:
543 if self._is_detached or key not in self._store:
544 if create_if_absent: 544 ↛ 545line 544 didn't jump to line 545 because the condition on line 544 was never true
545 return None
546 self.create_definition_if_missing()
547 self._store[key] = []
548 return self._store[key]
550 def _insert_item(self, key: str, item: AbstractYAMLDictSubStore) -> None:
551 parent_store = self._list_store(key, create_if_absent=True)
552 assert parent_store is not None
553 if not item._is_detached or ( 553 ↛ 556line 553 didn't jump to line 556 because the condition on line 553 was never true
554 item._parent_store is not None and item._parent_store is not parent_store
555 ):
556 raise RuntimeError(
557 "Item is already attached or associated with a different container"
558 )
559 item._parent_store = parent_store
560 item.create_definition()
562 def add_symlink(self, symlink: MutableYAMLSymlink) -> None:
563 self._insert_item(MK_TRANSFORMATIONS, symlink)
565 def symlinks(self) -> Iterable[MutableYAMLSymlink]:
566 store = self._list_store(MK_TRANSFORMATIONS)
567 if store is None: 567 ↛ 568line 567 didn't jump to line 568 because the condition on line 567 was never true
568 return
569 for i in range(len(store)): 569 ↛ 570line 569 didn't jump to line 570 because the loop on line 569 never started
570 d = store[i]
571 if d and isinstance(d, dict) and len(d) == 1 and "symlink" in d:
572 yield MutableYAMLSymlink(store, i)
574 def conffile_management_items(self) -> Iterable[MutableYAMLConffileManagementItem]:
575 store = self._list_store(MK_CONFFILE_MANAGEMENT)
576 if store is None: 576 ↛ 577line 576 didn't jump to line 577 because the condition on line 576 was never true
577 return
578 yield from (
579 MutableYAMLConffileManagementItem(store, i) for i in range(len(store))
580 )
582 def add_conffile_management(
583 self, conffile_management_item: MutableYAMLConffileManagementItem
584 ) -> None:
585 self._insert_item(MK_CONFFILE_MANAGEMENT, conffile_management_item)
588class AbstractMutableYAMLInstallRule(AbstractYAMLDictSubStore):
589 @property
590 def _container(self) -> dict[str, Any]:
591 assert len(self._store) == 1
592 return next(iter(self._store.values()))
594 @property
595 def into(self) -> list[str] | None:
596 v = self._container[MK_INSTALLATIONS_INSTALL_INTO]
597 if v is None:
598 return None
599 if isinstance(v, str):
600 return [v]
601 return v
603 @into.setter
604 def into(self, new_value: str | list[str] | None) -> None:
605 if new_value is None: 605 ↛ 609line 605 didn't jump to line 609 because the condition on line 605 was always true
606 with suppress(KeyError):
607 del self._container[MK_INSTALLATIONS_INSTALL_INTO]
608 return
609 if isinstance(new_value, str):
610 self._container[MK_INSTALLATIONS_INSTALL_INTO] = new_value
611 return
612 new_list = CommentedSeq(new_value)
613 self._container[MK_INSTALLATIONS_INSTALL_INTO] = new_list
615 @property
616 def when(self) -> str | Mapping[str, Any] | None:
617 return self._container[MK_CONDITION_WHEN]
619 @when.setter
620 def when(self, new_value: str | Mapping[str, Any] | None) -> None:
621 if new_value is None: 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true
622 with suppress(KeyError):
623 del self._container[MK_CONDITION_WHEN]
624 return
625 if isinstance(new_value, str): 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 self._container[MK_CONDITION_WHEN] = new_value
627 return
628 new_map = CommentedMap(new_value)
629 self._container[MK_CONDITION_WHEN] = new_map
631 @classmethod
632 def install_dest(
633 cls,
634 sources: str | list[str],
635 into: str | list[str] | None,
636 *,
637 dest_dir: str | None = None,
638 when: str | Mapping[str, Any] | None = None,
639 ) -> "MutableYAMLInstallRuleInstall":
640 k = MK_INSTALLATIONS_INSTALL_SOURCES
641 if isinstance(sources, str):
642 k = MK_INSTALLATIONS_INSTALL_SOURCE
643 r = MutableYAMLInstallRuleInstall(
644 None,
645 None,
646 store=CommentedMap(
647 {
648 MK_INSTALLATIONS_INSTALL: CommentedMap(
649 {
650 k: sources,
651 }
652 )
653 }
654 ),
655 )
656 r.dest_dir = dest_dir
657 r.into = into
658 if when is not None:
659 r.when = when
660 return r
662 @classmethod
663 def multi_dest_install(
664 cls,
665 sources: str | list[str],
666 dest_dirs: Sequence[str],
667 into: str | list[str] | None,
668 *,
669 when: str | Mapping[str, Any] | None = None,
670 ) -> "MutableYAMLInstallRuleInstall":
671 k = MK_INSTALLATIONS_INSTALL_SOURCES
672 if isinstance(sources, str): 672 ↛ 674line 672 didn't jump to line 674 because the condition on line 672 was always true
673 k = MK_INSTALLATIONS_INSTALL_SOURCE
674 r = MutableYAMLInstallRuleInstall(
675 None,
676 None,
677 store=CommentedMap(
678 {
679 MK_INSTALLATIONS_MULTI_DEST_INSTALL: CommentedMap(
680 {
681 k: sources,
682 "dest-dirs": dest_dirs,
683 }
684 )
685 }
686 ),
687 )
688 r.into = into
689 if when is not None: 689 ↛ 690line 689 didn't jump to line 690 because the condition on line 689 was never true
690 r.when = when
691 return r
693 @classmethod
694 def install_as(
695 cls,
696 source: str,
697 install_as: str,
698 into: str | list[str] | None,
699 when: str | Mapping[str, Any] | None = None,
700 ) -> "MutableYAMLInstallRuleInstall":
701 r = MutableYAMLInstallRuleInstall(
702 None,
703 None,
704 store=CommentedMap(
705 {
706 MK_INSTALLATIONS_INSTALL: CommentedMap(
707 {
708 MK_INSTALLATIONS_INSTALL_SOURCE: source,
709 MK_INSTALLATIONS_INSTALL_AS: install_as,
710 }
711 )
712 }
713 ),
714 )
715 r.into = into
716 if when is not None: 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 r.when = when
718 return r
720 @classmethod
721 def install_doc_as(
722 cls,
723 source: str,
724 install_as: str,
725 into: str | list[str] | None,
726 when: str | Mapping[str, Any] | None = None,
727 ) -> "MutableYAMLInstallRuleInstall":
728 r = MutableYAMLInstallRuleInstall(
729 None,
730 None,
731 store=CommentedMap(
732 {
733 MK_INSTALLATIONS_INSTALL_DOCS: CommentedMap(
734 {
735 MK_INSTALLATIONS_INSTALL_SOURCE: source,
736 MK_INSTALLATIONS_INSTALL_AS: install_as,
737 }
738 )
739 }
740 ),
741 )
742 r.into = into
743 if when is not None:
744 r.when = when
745 return r
747 @classmethod
748 def install_docs(
749 cls,
750 sources: str | list[str],
751 into: str | list[str] | None,
752 *,
753 dest_dir: str | None = None,
754 when: str | Mapping[str, Any] | None = None,
755 ) -> "MutableYAMLInstallRuleInstall":
756 k = MK_INSTALLATIONS_INSTALL_SOURCES
757 if isinstance(sources, str):
758 k = MK_INSTALLATIONS_INSTALL_SOURCE
759 r = MutableYAMLInstallRuleInstall(
760 None,
761 None,
762 store=CommentedMap(
763 {
764 MK_INSTALLATIONS_INSTALL_DOCS: CommentedMap(
765 {
766 k: sources,
767 }
768 )
769 }
770 ),
771 )
772 r.into = into
773 r.dest_dir = dest_dir
774 if when is not None:
775 r.when = when
776 return r
778 @classmethod
779 def install_examples(
780 cls,
781 sources: str | list[str],
782 into: str | list[str] | None,
783 when: str | Mapping[str, Any] | None = None,
784 ) -> "MutableYAMLInstallRuleInstallExamples":
785 k = MK_INSTALLATIONS_INSTALL_SOURCES
786 if isinstance(sources, str):
787 k = MK_INSTALLATIONS_INSTALL_SOURCE
788 r = MutableYAMLInstallRuleInstallExamples(
789 None,
790 None,
791 store=CommentedMap(
792 {
793 MK_INSTALLATIONS_INSTALL_EXAMPLES: CommentedMap(
794 {
795 k: sources,
796 }
797 )
798 }
799 ),
800 )
801 r.into = into
802 if when is not None: 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true
803 r.when = when
804 return r
806 @classmethod
807 def install_man(
808 cls,
809 sources: str | list[str],
810 into: str | list[str] | None,
811 language: str | None,
812 when: str | Mapping[str, Any] | None = None,
813 ) -> "MutableYAMLInstallRuleMan":
814 k = MK_INSTALLATIONS_INSTALL_SOURCES
815 if isinstance(sources, str): 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true
816 k = MK_INSTALLATIONS_INSTALL_SOURCE
817 r = MutableYAMLInstallRuleMan(
818 None,
819 None,
820 store=CommentedMap(
821 {
822 MK_INSTALLATIONS_INSTALL_MAN: CommentedMap(
823 {
824 k: sources,
825 }
826 )
827 }
828 ),
829 )
830 r.language = language
831 r.into = into
832 if when is not None: 832 ↛ 833line 832 didn't jump to line 833 because the condition on line 832 was never true
833 r.when = when
834 return r
836 @classmethod
837 def discard(
838 cls,
839 sources: str | list[str],
840 ) -> "MutableYAMLInstallRuleDiscard":
841 return MutableYAMLInstallRuleDiscard(
842 None,
843 None,
844 store=CommentedMap({MK_INSTALLATIONS_DISCARD: sources}),
845 )
848class MutableYAMLInstallRuleInstallExamples(AbstractMutableYAMLInstallRule):
849 pass
852class MutableYAMLInstallRuleMan(AbstractMutableYAMLInstallRule):
853 @property
854 def language(self) -> str | None:
855 return self._container[MK_INSTALLATIONS_INSTALL_MAN_LANGUAGE]
857 @language.setter
858 def language(self, new_value: str | None) -> None:
859 if new_value is not None:
860 self._container[MK_INSTALLATIONS_INSTALL_MAN_LANGUAGE] = new_value
861 return
862 with suppress(KeyError):
863 del self._container[MK_INSTALLATIONS_INSTALL_MAN_LANGUAGE]
866class MutableYAMLInstallRuleDiscard(AbstractMutableYAMLInstallRule):
867 pass
870class MutableYAMLInstallRuleInstall(AbstractMutableYAMLInstallRule):
871 @property
872 def sources(self) -> list[str]:
873 v = self._container[MK_INSTALLATIONS_INSTALL_SOURCES]
874 if isinstance(v, str):
875 return [v]
876 return v
878 @sources.setter
879 def sources(self, new_value: str | list[str]) -> None:
880 if isinstance(new_value, str):
881 self._container[MK_INSTALLATIONS_INSTALL_SOURCES] = new_value
882 return
883 new_list = CommentedSeq(new_value)
884 self._container[MK_INSTALLATIONS_INSTALL_SOURCES] = new_list
886 @property
887 def dest_dir(self) -> str | None:
888 return self._container.get(MK_INSTALLATIONS_INSTALL_DEST_DIR)
890 @dest_dir.setter
891 def dest_dir(self, new_value: str | None) -> None:
892 if new_value is not None and self.dest_as is not None: 892 ↛ 893line 892 didn't jump to line 893 because the condition on line 892 was never true
893 raise ValueError(
894 f'Cannot both have a "{MK_INSTALLATIONS_INSTALL_DEST_DIR}" and'
895 f' "{MK_INSTALLATIONS_INSTALL_AS}"'
896 )
897 if new_value is not None:
898 self._container[MK_INSTALLATIONS_INSTALL_DEST_DIR] = new_value
899 else:
900 with suppress(KeyError):
901 del self._container[MK_INSTALLATIONS_INSTALL_DEST_DIR]
903 @property
904 def dest_as(self) -> str | None:
905 return self._container.get(MK_INSTALLATIONS_INSTALL_AS)
907 @dest_as.setter
908 def dest_as(self, new_value: str | None) -> None:
909 if new_value is not None:
910 if self.dest_dir is not None:
911 raise ValueError(
912 f'Cannot both have a "{MK_INSTALLATIONS_INSTALL_DEST_DIR}" and'
913 f' "{MK_INSTALLATIONS_INSTALL_AS}"'
914 )
916 sources = self._container[MK_INSTALLATIONS_INSTALL_SOURCES]
917 if isinstance(sources, list):
918 if len(sources) != 1:
919 raise ValueError(
920 f'Cannot have "{MK_INSTALLATIONS_INSTALL_AS}" when'
921 f' "{MK_INSTALLATIONS_INSTALL_SOURCES}" is not exactly one item'
922 )
923 self.sources = sources[0]
924 self._container[MK_INSTALLATIONS_INSTALL_AS] = new_value
925 else:
926 with suppress(KeyError):
927 del self._container[MK_INSTALLATIONS_INSTALL_AS]
930class MutableYAMLInstallationsDefinition(AbstractYAMLListSubStore[Any]):
931 def append(self, install_rule: AbstractMutableYAMLInstallRule) -> None:
932 parent_store = self._store
933 if not install_rule._is_detached or ( 933 ↛ 937line 933 didn't jump to line 937 because the condition on line 933 was never true
934 install_rule._parent_store is not None
935 and install_rule._parent_store is not parent_store
936 ):
937 raise RuntimeError(
938 "Item is already attached or associated with a different container"
939 )
940 self.create_definition_if_missing()
941 install_rule._parent_store = parent_store
942 install_rule.create_definition()
944 def extend(self, install_rules: Iterable[AbstractMutableYAMLInstallRule]) -> None:
945 parent_store = self._store
946 for install_rule in install_rules:
947 if not install_rule._is_detached or ( 947 ↛ 951line 947 didn't jump to line 951 because the condition on line 947 was never true
948 install_rule._parent_store is not None
949 and install_rule._parent_store is not parent_store
950 ):
951 raise RuntimeError(
952 "Item is already attached or associated with a different container"
953 )
954 self.create_definition_if_missing()
955 install_rule._parent_store = parent_store
956 install_rule.create_definition()
959class MutableYAMLManifestVariables(AbstractYAMLDictSubStore):
960 @property
961 def variables(self) -> dict[str, Any]:
962 return self._store
964 def __setitem__(self, key: str, value: Any) -> None:
965 self._store[key] = value
966 self.create_definition_if_missing()
969class MutableYAMLManifestDefinitions(AbstractYAMLDictSubStore):
970 def manifest_variables(
971 self, *, create_if_absent: bool = True
972 ) -> MutableYAMLManifestVariables:
973 d = MutableYAMLManifestVariables(self._store, MK_MANIFEST_VARIABLES)
974 if create_if_absent: 974 ↛ 975line 974 didn't jump to line 975 because the condition on line 974 was never true
975 d.create_definition_if_missing()
976 return d
979class MutableYAMLRemoveDuringCleanDefinitions(AbstractYAMLListSubStore[str]):
980 def append(self, rule: str) -> None:
981 self.create_definition_if_missing()
982 self._store.append(rule)
984 def __len__(self) -> int:
985 return len(self._store)
987 def extend(self, rules: Iterable[str]) -> None:
988 it = iter(rules)
989 try:
990 first_rule = next(it)
991 except StopIteration:
992 return
993 self.create_definition_if_missing()
994 self._store.append(first_rule)
995 self._store.extend(it)
998class MutableYAMLManifest:
999 def __init__(self, store: Any) -> None:
1000 self._store = store
1002 @classmethod
1003 def empty_manifest(cls) -> "MutableYAMLManifest":
1004 return cls(CommentedMap({MK_MANIFEST_VERSION: DEFAULT_MANIFEST_VERSION}))
1006 @property
1007 def manifest_version(self) -> str:
1008 return self._store[MK_MANIFEST_VERSION]
1010 @manifest_version.setter
1011 def manifest_version(self, version: str) -> None:
1012 if version not in SUPPORTED_MANIFEST_VERSIONS:
1013 raise ValueError("Unsupported version")
1014 self._store[MK_MANIFEST_VERSION] = version
1016 def remove_during_clean(
1017 self,
1018 *,
1019 create_if_absent: bool = True,
1020 ) -> MutableYAMLRemoveDuringCleanDefinitions:
1021 d = MutableYAMLRemoveDuringCleanDefinitions(
1022 self._store, MK_MANIFEST_REMOVE_DURING_CLEAN
1023 )
1024 if create_if_absent: 1024 ↛ 1025line 1024 didn't jump to line 1025 because the condition on line 1024 was never true
1025 d.create_definition_if_missing()
1026 return d
1028 def installations(
1029 self,
1030 *,
1031 create_if_absent: bool = True,
1032 ) -> MutableYAMLInstallationsDefinition:
1033 d = MutableYAMLInstallationsDefinition(self._store, MK_INSTALLATIONS)
1034 if create_if_absent: 1034 ↛ 1035line 1034 didn't jump to line 1035 because the condition on line 1034 was never true
1035 d.create_definition_if_missing()
1036 return d
1038 def manifest_definitions(
1039 self,
1040 *,
1041 create_if_absent: bool = True,
1042 ) -> MutableYAMLManifestDefinitions:
1043 d = MutableYAMLManifestDefinitions(self._store, MK_MANIFEST_DEFINITIONS)
1044 if create_if_absent: 1044 ↛ 1045line 1044 didn't jump to line 1045 because the condition on line 1044 was never true
1045 d.create_definition_if_missing()
1046 return d
1048 def package(
1049 self, name: str, *, create_if_absent: bool = True
1050 ) -> MutableYAMLPackageDefinition:
1051 if MK_PACKAGES not in self._store: 1051 ↛ 1053line 1051 didn't jump to line 1053 because the condition on line 1051 was always true
1052 self._store[MK_PACKAGES] = CommentedMap()
1053 packages_store = self._store[MK_PACKAGES]
1054 package = packages_store.get(name)
1055 if package is None: 1055 ↛ 1062line 1055 didn't jump to line 1062 because the condition on line 1055 was always true
1056 if not create_if_absent: 1056 ↛ 1057line 1056 didn't jump to line 1057 because the condition on line 1056 was never true
1057 raise KeyError(name)
1058 assert packages_store is not None
1059 d = MutableYAMLPackageDefinition(packages_store, name)
1060 d.create_definition()
1061 else:
1062 d = MutableYAMLPackageDefinition(packages_store, name)
1063 return d
1065 def write_to(self, fd) -> None:
1066 MANIFEST_YAML.dump(self._store, fd)
1069def _describe_missing_path(entry: VirtualPath) -> str:
1070 if entry.is_dir:
1071 return f"{entry.fs_path}/ (empty directory; possible integration point)"
1072 if entry.is_symlink:
1073 target = os.readlink(entry.fs_path)
1074 return f"{entry.fs_path} (symlink; links to {target})"
1075 if entry.is_file:
1076 return f"{entry.fs_path} (file)"
1077 return f"{entry.fs_path} (other!? Probably not supported by debputy and may need a `remove`)"
1080def _detect_missing_installations(
1081 path_matcher: SourcePathMatcher,
1082 search_dir: VirtualPath,
1083) -> None:
1084 if not search_dir.is_dir: 1084 ↛ 1085line 1084 didn't jump to line 1085 because the condition on line 1084 was never true
1085 return
1086 missing = list(path_matcher.detect_missing(search_dir))
1087 if not missing:
1088 return
1090 excl = textwrap.dedent("""\
1091 - discard: "*"
1092 """)
1094 raise PathNotCoveredByInstallRulesError(
1095 "Please review the list and add either install rules or exclusions to `installations` in"
1096 " debian/debputy.manifest. If you do not need any of these paths, add the following to the"
1097 f" end of your 'installations`:\n\n{excl}\n",
1098 missing,
1099 search_dir,
1100 )
1103def _list_automatic_discard_rules(path_matcher: SourcePathMatcher) -> None:
1104 used_discard_rules = path_matcher.used_auto_discard_rules
1105 # Discard rules can match and then be overridden. In that case, they appear
1106 # but have 0 matches.
1107 if not sum((len(v) for v in used_discard_rules.values()), 0):
1108 return
1109 _info("The following automatic discard rules were triggered:")
1110 example_path: str | None = None
1111 for rule in sorted(used_discard_rules):
1112 for fs_path in sorted(used_discard_rules[rule]):
1113 if example_path is None: 1113 ↛ 1115line 1113 didn't jump to line 1115 because the condition on line 1113 was always true
1114 example_path = fs_path
1115 _info(f" * {rule} -> {fs_path}")
1116 assert example_path is not None
1117 _info("")
1118 _info(
1119 "Note that some of these may have been overruled. The overrule detection logic is not"
1120 )
1121 _info("100% reliable.")
1122 _info("")
1123 _info(
1124 "You can overrule an automatic discard rule by explicitly listing the path. As an example:"
1125 )
1126 _info(" installations:")
1127 _info(" - install:")
1128 _info(f" source: {example_path}")
1131def _install_everything_from_source_dir_if_present(
1132 dctrl_bin: BinaryPackage,
1133 substitution: Substitution,
1134 path_matcher: SourcePathMatcher,
1135 install_rule_context: InstallRuleContext,
1136 source_condition_context: ConditionContext,
1137 source_dir: VirtualPath,
1138 *,
1139 into_dir: VirtualPath | None = None,
1140) -> None:
1141 attribute_path = AttributePath.builtin_path()[f"installing {source_dir.fs_path}"]
1142 pkg_set = frozenset([dctrl_bin])
1143 install_rule = run_in_context_of_plugin(
1144 "debputy",
1145 InstallRule.install_dest,
1146 [FileSystemMatchRule.from_path_match("*", attribute_path, substitution)],
1147 None,
1148 pkg_set,
1149 f"Built-in; install everything from {source_dir.fs_path} into {dctrl_bin.name}",
1150 None,
1151 )
1152 pkg_search_dir: tuple[SearchDir] = (
1153 SearchDir(
1154 source_dir,
1155 pkg_set,
1156 ),
1157 )
1158 replacements = {
1159 "search_dirs": pkg_search_dir,
1160 }
1161 if into_dir is not None: 1161 ↛ 1162line 1161 didn't jump to line 1162 because the condition on line 1161 was never true
1162 binary_package_contexts = dict(install_rule_context.binary_package_contexts)
1163 updated = binary_package_contexts[dctrl_bin.name].replace(fs_root=into_dir)
1164 binary_package_contexts[dctrl_bin.name] = updated
1165 replacements["binary_package_contexts"] = binary_package_contexts
1167 fake_install_rule_context = install_rule_context.replace(**replacements)
1168 try:
1169 install_rule.perform_install(
1170 path_matcher,
1171 fake_install_rule_context,
1172 source_condition_context,
1173 )
1174 except (
1175 NoMatchForInstallPatternError,
1176 PathAlreadyInstalledOrDiscardedError,
1177 ):
1178 # Empty directory or everything excluded by default; ignore the error
1179 pass
1182def _add_build_install_dirs_to_per_package_search_dirs(
1183 build_system_install_dirs: Sequence[tuple[str, frozenset[BinaryPackage]]],
1184 per_package_search_dirs: dict[BinaryPackage, list[VirtualPath]],
1185 as_path: Callable[[str], VirtualPath],
1186) -> None:
1187 seen_pp_search_dirs: set[tuple[BinaryPackage, str]] = set()
1188 for dest_dir, for_packages in build_system_install_dirs:
1189 dest_path = as_path(dest_dir)
1190 for pkg in for_packages:
1191 seen_key = (pkg, dest_dir)
1192 if seen_key in seen_pp_search_dirs:
1193 continue
1194 seen_pp_search_dirs.add(seen_key)
1195 if pkg not in per_package_search_dirs:
1196 per_package_search_dirs[pkg] = [dest_path]
1197 else:
1198 per_package_search_dirs[pkg].append(dest_path)
1201class HighLevelManifest:
1202 def __init__(
1203 self,
1204 manifest_path: str,
1205 mutable_manifest: MutableYAMLManifest | None,
1206 remove_during_clean_rules: list[FileSystemMatchRule],
1207 install_rules: list[InstallRule] | None,
1208 source_package: SourcePackage,
1209 binary_packages: Mapping[str, BinaryPackage],
1210 substitution: Substitution,
1211 package_transformations: Mapping[str, PackageTransformationDefinition],
1212 dpkg_architecture_variables: DpkgArchitectureBuildProcessValuesTable,
1213 dpkg_arch_query_table: DpkgArchTable,
1214 build_env: DebBuildOptionsAndProfiles,
1215 build_environments: BuildEnvironments,
1216 build_rules: list[BuildRule] | None,
1217 value_table: Mapping[
1218 tuple[SourcePackage | BinaryPackage, type[Any]],
1219 Any,
1220 ],
1221 plugin_provided_feature_set: PluginProvidedFeatureSet,
1222 debian_dir: VirtualPath,
1223 ) -> None:
1224 self.manifest_path = manifest_path
1225 self.mutable_manifest = mutable_manifest
1226 self._remove_during_clean_rules: list[FileSystemMatchRule] = (
1227 remove_during_clean_rules
1228 )
1229 self._install_rules = install_rules
1230 self.source_package = source_package
1231 self._binary_packages = binary_packages
1232 self.substitution = substitution
1233 self.package_transformations = package_transformations
1234 self._dpkg_architecture_variables = dpkg_architecture_variables
1235 self.dpkg_arch_query_table = dpkg_arch_query_table
1236 self._build_env = build_env
1237 self._used_for: set[str] = set()
1238 self.build_environments = build_environments
1239 self.build_rules = build_rules
1240 self._value_table = value_table
1241 self._plugin_provided_feature_set = plugin_provided_feature_set
1242 self._debian_dir = debian_dir
1243 self._source_condition_context = ConditionContext(
1244 binary_package=None,
1245 substitution=self.substitution,
1246 deb_options_and_profiles=self._build_env,
1247 dpkg_architecture_variables=self._dpkg_architecture_variables,
1248 dpkg_arch_query_table=self.dpkg_arch_query_table,
1249 )
1251 def source_version(self, include_binnmu_version: bool = True) -> str:
1252 # TODO: There should an easier way to determine the source version; really.
1253 version_var = "{{DEB_VERSION}}"
1254 if not include_binnmu_version:
1255 version_var = "{{_DEBPUTY_INTERNAL_NON_BINNMU_SOURCE}}"
1256 try:
1257 return self.substitution.substitute(
1258 version_var, "internal (resolve version)"
1259 )
1260 except DebputySubstitutionError as e:
1261 raise AssertionError(f"Could not resolve {version_var}") from e
1263 @property
1264 def source_condition_context(self) -> ConditionContext:
1265 return self._source_condition_context
1267 @property
1268 def debian_dir(self) -> VirtualPath:
1269 return self._debian_dir
1271 @property
1272 def dpkg_architecture_variables(self) -> DpkgArchitectureBuildProcessValuesTable:
1273 return self._dpkg_architecture_variables
1275 @property
1276 def deb_options_and_profiles(self) -> DebBuildOptionsAndProfiles:
1277 return self._build_env
1279 @property
1280 def plugin_provided_feature_set(self) -> PluginProvidedFeatureSet:
1281 return self._plugin_provided_feature_set
1283 @property
1284 def remove_during_clean_rules(self) -> list[FileSystemMatchRule]:
1285 return self._remove_during_clean_rules
1287 @property
1288 def active_packages(self) -> Iterable[BinaryPackage]:
1289 yield from (p for p in self._binary_packages.values() if p.should_be_acted_on)
1291 @property
1292 def all_packages(self) -> Iterable[BinaryPackage]:
1293 yield from self._binary_packages.values()
1295 def manifest_configuration[T](
1296 self,
1297 context_package: SourcePackage | BinaryPackage,
1298 value_type: type[T],
1299 ) -> T | None:
1300 res = self._value_table.get((context_package, value_type))
1301 return typing.cast("T | None", res)
1303 def package_state_for(self, package: str) -> PackageTransformationDefinition:
1304 return self.package_transformations[package]
1306 def _detect_doc_main_package_for(self, package: BinaryPackage) -> BinaryPackage:
1307 name = package.name
1308 for doc_main_field in ("Doc-Main-Package", "X-Doc-Main-Package"):
1309 doc_main_package_name = package.fields.get(doc_main_field)
1310 if doc_main_package_name: 1310 ↛ 1311line 1310 didn't jump to line 1311 because the condition on line 1310 was never true
1311 main_package = self._binary_packages.get(doc_main_package_name)
1312 if main_package is None:
1313 _error(
1314 f"Invalid Doc-Main-Package for {name}: The package {doc_main_package_name!r} is not listed in d/control"
1315 )
1316 return main_package
1317 # If it is not a -doc package, then docs should be installed
1318 # under its own package name.
1319 if not name.endswith("-doc"): 1319 ↛ 1321line 1319 didn't jump to line 1321 because the condition on line 1319 was always true
1320 return package
1321 name = name[:-4]
1322 main_package = self._binary_packages.get(name)
1323 if main_package:
1324 return main_package
1325 if name.startswith("lib"):
1326 dev_pkg = self._binary_packages.get(f"{name}-dev")
1327 if dev_pkg:
1328 return dev_pkg
1330 # If we found no better match; default to the doc package itself.
1331 return package
1333 def perform_installations(
1334 self,
1335 integration_mode: DebputyIntegrationMode,
1336 build_system_install_dirs: Sequence[tuple[str, frozenset[BinaryPackage]]],
1337 *,
1338 install_request_context: InstallSearchDirContext | None = None,
1339 ) -> PackageDataTable:
1340 package_data_dict = {}
1341 package_data_table = PackageDataTable(package_data_dict)
1342 enable_manifest_installation_feature = (
1343 integration_mode != INTEGRATION_MODE_DH_DEBPUTY_RRR
1344 )
1346 if build_system_install_dirs: 1346 ↛ 1347line 1346 didn't jump to line 1347 because the condition on line 1346 was never true
1347 if integration_mode != INTEGRATION_MODE_FULL:
1348 raise ValueError(
1349 "The build_system_install_dirs parameter can only be used in full integration mode"
1350 )
1351 if install_request_context:
1352 raise ValueError(
1353 "The build_system_install_dirs parameter cannot be used with install_request_context"
1354 " (not implemented)"
1355 )
1357 if install_request_context is None: 1357 ↛ 1359line 1357 didn't jump to line 1359 because the condition on line 1357 was never true
1359 @functools.lru_cache(None)
1360 def _as_path(fs_path: str) -> VirtualPath:
1361 return OSFSROOverlay.create_root_dir(".", fs_path)
1363 dtmp_dir = _as_path("debian/tmp")
1364 dtmp_deb_dir: VirtualPath | None = _as_path("debian/tmp-deb")
1365 dtmp_udeb_dir: VirtualPath | None = _as_path("debian/tmp-udeb")
1366 if not os.path.isdir(assume_not_none(dtmp_dir).fs_path):
1367 dtmp_dir = None
1368 if not os.path.isdir(assume_not_none(dtmp_deb_dir).fs_path):
1369 dtmp_deb_dir = None
1370 if not os.path.isdir(assume_not_none(dtmp_udeb_dir).fs_path):
1371 dtmp_udeb_dir = None
1372 source_root_dir = _as_path(".")
1373 into = frozenset(self._binary_packages.values())
1374 per_package_search_dirs = {
1375 t.binary_package: [_as_path(f.match_rule.path) for f in t.search_dirs]
1376 for t in self.package_transformations.values()
1377 if t.search_dirs is not None
1378 }
1379 if integration_mode == INTEGRATION_MODE_FULL:
1380 # We can end here with per_package_search_dirs having no search dirs for any package
1381 # (single binary, where everything is installed into d/<pkg> is the most common case).
1382 #
1383 # This is not a problem in itself as the installation rules can still apply to the
1384 # source root and there should be no reason to install something from d/<pkg> into
1385 # d/<another-pkg>
1386 _add_build_install_dirs_to_per_package_search_dirs(
1387 build_system_install_dirs,
1388 per_package_search_dirs,
1389 _as_path,
1390 )
1391 else:
1392 # When we interact with `debhelper`, look for `debian/tmp-deb`, `debian/tmp-udeb`, and `debian/tmp`
1393 # as likely candidates
1394 for p in self.all_packages:
1395 if p not in per_package_search_dirs:
1396 per_package_search_dirs[p] = []
1397 search_path = dtmp_udeb_dir if p.is_udeb else dtmp_deb_dir
1398 existing = (
1399 {d.fs_path for d in per_package_search_dirs[p]}
1400 if search_path
1401 else frozenset()
1402 )
1403 if search_path and search_path.fs_path not in existing:
1404 _info(
1405 f"Implicit search path {search_path.fs_path} for {p.name}"
1406 )
1407 per_package_search_dirs[p].append(search_path)
1408 if dtmp_dir and dtmp_dir.fs_path not in existing:
1409 _info(f"Implicit search path {dtmp_dir.fs_path} for {p.name}")
1410 per_package_search_dirs[p].append(dtmp_dir)
1412 search_dirs = _determine_search_dir_order(
1413 per_package_search_dirs,
1414 into,
1415 source_root_dir,
1416 )
1417 check_for_uninstalled_dirs = tuple(
1418 s.search_dir
1419 for s in search_dirs
1420 if s.search_dir.fs_path != source_root_dir.fs_path
1421 )
1422 if enable_manifest_installation_feature:
1423 _present_installation_dirs(
1424 search_dirs, check_for_uninstalled_dirs, into
1425 )
1426 else:
1427 dtmp_dir = None
1428 dtmp_deb_dir = None
1429 dtmp_udeb_dir = None
1430 search_dirs = install_request_context.search_dirs
1431 into = frozenset(self._binary_packages.values())
1432 seen: set[BinaryPackage] = set()
1433 for search_dir in search_dirs:
1434 seen.update(search_dir.applies_to)
1436 missing = into - seen
1437 if missing: 1437 ↛ 1438line 1437 didn't jump to line 1438 because the condition on line 1437 was never true
1438 names = ", ".join(p.name for p in missing)
1439 raise ValueError(
1440 f"The following package(s) had no search dirs: {names}."
1441 " (Generally, the source root would be applicable to all packages)"
1442 )
1443 extra_names = seen - into
1444 if extra_names: 1444 ↛ 1445line 1444 didn't jump to line 1445 because the condition on line 1444 was never true
1445 names = ", ".join(p.name for p in extra_names)
1446 raise ValueError(
1447 f"The install_request_context referenced the following unknown package(s): {names}"
1448 )
1450 check_for_uninstalled_dirs = (
1451 install_request_context.check_for_uninstalled_dirs
1452 )
1454 install_rule_context = InstallRuleContext(search_dirs)
1455 check_dirs = [dtmp_dir, dtmp_deb_dir, dtmp_udeb_dir]
1457 if ( 1457 ↛ 1463line 1457 didn't jump to line 1463 because the condition on line 1457 was never true
1458 enable_manifest_installation_feature
1459 and self._install_rules is None
1460 # TODO: Should we also do this for full mode when build systems provided search dirs?
1461 and (tmpdirs := [d for d in check_dirs if d and os.path.isdir(d.fs_path)])
1462 ):
1463 first_tmp_dir = tmpdirs[0]
1464 msg = (
1465 "The build system appears to have provided the output of upstream build system's"
1466 f" install in {first_tmp_dir.fs_path}. However, these are no provisions for debputy to install"
1467 " any of that into any of the debian packages listed in debian/control."
1468 " To avoid accidentally creating empty packages, debputy will insist that you "
1469 " explicitly define an empty installation definition if you did not want to "
1470 " install any of those files even though they have been provided."
1471 ' Example: "installations: []"'
1472 )
1473 _error(msg)
1474 elif ( 1474 ↛ 1477line 1474 didn't jump to line 1477 because the condition on line 1474 was never true
1475 not enable_manifest_installation_feature and self._install_rules is not None
1476 ):
1477 _error(
1478 f"The `installations` feature cannot be used in {self.manifest_path} with this integration mode."
1479 f" Please remove or comment out the `installations` keyword."
1480 )
1482 for dctrl_bin in self.all_packages:
1483 package = dctrl_bin.name
1484 doc_main_package = self._detect_doc_main_package_for(dctrl_bin)
1486 install_rule_context[package] = BinaryPackageInstallRuleContext(
1487 dctrl_bin,
1488 InMemoryVirtualRootDir(),
1489 doc_main_package,
1490 )
1492 if enable_manifest_installation_feature: 1492 ↛ 1497line 1492 didn't jump to line 1497 because the condition on line 1492 was always true
1493 discard_rules = list(
1494 self.plugin_provided_feature_set.auto_discard_rules.values()
1495 )
1496 else:
1497 discard_rules = [
1498 self.plugin_provided_feature_set.auto_discard_rules["debian-dir"]
1499 ]
1500 path_matcher = SourcePathMatcher(discard_rules)
1502 source_condition_context = self._source_condition_context
1504 for dctrl_bin in self.active_packages:
1505 package = dctrl_bin.name
1506 if install_request_context: 1506 ↛ 1511line 1506 didn't jump to line 1511 because the condition on line 1506 was always true
1507 build_system_staging_dir = install_request_context.debian_pkg_dirs.get(
1508 package
1509 )
1510 else:
1511 build_system_staging_dir_fs_path = os.path.join("debian", package)
1512 if os.path.isdir(build_system_staging_dir_fs_path):
1513 build_system_staging_dir = OSFSROOverlay.create_root_dir(
1514 ".",
1515 build_system_staging_dir_fs_path,
1516 )
1517 else:
1518 build_system_staging_dir = None
1520 if build_system_staging_dir is not None:
1521 _install_everything_from_source_dir_if_present(
1522 dctrl_bin,
1523 self.substitution,
1524 path_matcher,
1525 install_rule_context,
1526 source_condition_context,
1527 build_system_staging_dir,
1528 )
1530 if self._install_rules:
1531 # FIXME: Check that every install rule remains used after transformations have run.
1532 # What we want to check is transformations do not exclude everything from an install
1533 # rule. The hard part here is that renaming (etc.) is fine, so we cannot 1:1 string
1534 # match.
1535 for install_rule in self._install_rules:
1536 install_rule.perform_install(
1537 path_matcher,
1538 install_rule_context,
1539 source_condition_context,
1540 )
1542 if enable_manifest_installation_feature: 1542 ↛ 1546line 1542 didn't jump to line 1546 because the condition on line 1542 was always true
1543 for search_dir in check_for_uninstalled_dirs:
1544 _detect_missing_installations(path_matcher, search_dir)
1546 for dctrl_bin in self.all_packages:
1547 package = dctrl_bin.name
1548 binary_install_rule_context = install_rule_context[package]
1549 build_system_pkg_staging_dir = os.path.join("debian", package)
1550 fs_root = binary_install_rule_context.fs_root
1552 context = self.package_transformations[package]
1553 if dctrl_bin.should_be_acted_on and enable_manifest_installation_feature:
1554 for special_install_rule in context.install_rules: 1554 ↛ 1555line 1554 didn't jump to line 1555 because the loop on line 1554 never started
1555 special_install_rule.perform_install(
1556 path_matcher,
1557 install_rule_context,
1558 source_condition_context,
1559 )
1561 if dctrl_bin.should_be_acted_on:
1562 self.apply_fs_transformations(package, fs_root)
1563 substvars_file = f"debian/{package}.substvars"
1564 substvars = FlushableSubstvars.load_from_path(
1565 substvars_file, missing_ok=True
1566 )
1567 # We do not want to touch the substvars file (non-clean rebuild contamination)
1568 substvars.substvars_path = None
1569 else:
1570 substvars = FlushableSubstvars()
1572 udeb_package = self._binary_packages.get(f"{package}-udeb")
1573 if udeb_package and not udeb_package.is_udeb: 1573 ↛ 1574line 1573 didn't jump to line 1574 because the condition on line 1573 was never true
1574 udeb_package = None
1576 package_metadata_context = PackageProcessingContextProvider(
1577 self,
1578 dctrl_bin,
1579 udeb_package,
1580 package_data_table,
1581 )
1583 ctrl_creator = BinaryCtrlAccessorProviderCreator(
1584 package_metadata_context,
1585 substvars,
1586 context.maintscript_snippets,
1587 context.substitution,
1588 )
1590 if not enable_manifest_installation_feature: 1590 ↛ 1591line 1590 didn't jump to line 1591 because the condition on line 1590 was never true
1591 assert_no_dbgsym_migration(dctrl_bin)
1592 dh_dbgsym_root_fs = dhe_dbgsym_root_dir(dctrl_bin)
1593 dh_dbgsym_root_path = OSFSROOverlay.create_root_dir(
1594 "",
1595 dh_dbgsym_root_fs,
1596 )
1597 dbgsym_root_fs = InMemoryVirtualRootDir()
1598 _install_everything_from_source_dir_if_present(
1599 dctrl_bin,
1600 self.substitution,
1601 path_matcher,
1602 install_rule_context,
1603 source_condition_context,
1604 dh_dbgsym_root_path,
1605 into_dir=dbgsym_root_fs,
1606 )
1607 dbgsym_build_ids = read_dbgsym_file(dctrl_bin)
1608 dbgsym_info = DbgsymInfo(
1609 dctrl_bin,
1610 dbgsym_root_fs,
1611 os.path.join(dh_dbgsym_root_fs, "DEBIAN"),
1612 dbgsym_build_ids,
1613 # TODO: Provide manifest feature to support this.
1614 False,
1615 )
1616 else:
1617 dbgsym_info = DbgsymInfo(
1618 dctrl_bin,
1619 InMemoryVirtualRootDir(),
1620 None,
1621 [],
1622 False,
1623 )
1625 package_data_dict[package] = BinaryPackageData(
1626 self.source_package,
1627 dctrl_bin,
1628 build_system_pkg_staging_dir,
1629 fs_root,
1630 substvars,
1631 package_metadata_context,
1632 ctrl_creator,
1633 dbgsym_info,
1634 )
1636 if enable_manifest_installation_feature: 1636 ↛ 1639line 1636 didn't jump to line 1639 because the condition on line 1636 was always true
1637 _list_automatic_discard_rules(path_matcher)
1639 return package_data_table
1641 def condition_context(
1642 self, binary_package: BinaryPackage | str | None
1643 ) -> ConditionContext:
1644 if binary_package is None: 1644 ↛ 1645line 1644 didn't jump to line 1645 because the condition on line 1644 was never true
1645 return self._source_condition_context
1646 if not isinstance(binary_package, str):
1647 binary_package = binary_package.name
1649 package_transformation = self.package_transformations[binary_package]
1650 return self._source_condition_context.replace(
1651 binary_package=package_transformation.binary_package,
1652 substitution=package_transformation.substitution,
1653 )
1655 def apply_fs_transformations(
1656 self,
1657 package: str,
1658 fs_root: InMemoryVirtualPathBase,
1659 ) -> None:
1660 if package in self._used_for: 1660 ↛ 1661line 1660 didn't jump to line 1661 because the condition on line 1660 was never true
1661 raise ValueError(
1662 f"data.tar contents for {package} has already been finalized!?"
1663 )
1664 if package not in self.package_transformations: 1664 ↛ 1665line 1664 didn't jump to line 1665 because the condition on line 1664 was never true
1665 raise ValueError(
1666 f'The package "{package}" was not relevant for the manifest!?'
1667 )
1668 package_transformation = self.package_transformations[package]
1669 condition_context = ConditionContext(
1670 binary_package=package_transformation.binary_package,
1671 substitution=package_transformation.substitution,
1672 deb_options_and_profiles=self._build_env,
1673 dpkg_architecture_variables=self._dpkg_architecture_variables,
1674 dpkg_arch_query_table=self.dpkg_arch_query_table,
1675 )
1676 norm_rules = list(
1677 builtin_mode_normalization_rules(
1678 self._dpkg_architecture_variables,
1679 package_transformation.binary_package,
1680 package_transformation.substitution,
1681 )
1682 )
1683 norm_mode_transformation_rule = ModeNormalizationTransformationRule(norm_rules)
1684 norm_mode_transformation_rule.transform_file_system(fs_root, condition_context)
1685 for transformation in package_transformation.transformations:
1686 transformation.run_transform_file_system(fs_root, condition_context)
1687 interpreter_normalization = NormalizeShebangLineTransformation()
1688 interpreter_normalization.transform_file_system(fs_root, condition_context)
1690 def finalize_data_tar_contents(
1691 self,
1692 package: str,
1693 fs_root: InMemoryVirtualPathBase,
1694 clamp_mtime_to: int,
1695 ) -> IntermediateManifest:
1696 if package in self._used_for: 1696 ↛ 1697line 1696 didn't jump to line 1697 because the condition on line 1696 was never true
1697 raise ValueError(
1698 f"data.tar contents for {package} has already been finalized!?"
1699 )
1700 if package not in self.package_transformations: 1700 ↛ 1701line 1700 didn't jump to line 1701 because the condition on line 1700 was never true
1701 raise ValueError(
1702 f'The package "{package}" was not relevant for the manifest!?'
1703 )
1704 self._used_for.add(package)
1706 # At this point, there so be no further mutations to the file system (because they will not
1707 # be present in the intermediate manifest)
1708 #
1709 # We use `setattr` because the official API says it is a read-only property, but we know
1710 # that is a InMemoryOverlayFSRootDirectory and it allows write.
1711 setattr(fs_root, "is_read_write", False)
1713 intermediate_manifest = list(
1714 _generate_intermediate_manifest(
1715 fs_root,
1716 clamp_mtime_to,
1717 )
1718 )
1719 return intermediate_manifest
1721 def apply_to_binary_staging_directory(
1722 self,
1723 package: str,
1724 fs_root: InMemoryVirtualPathBase,
1725 clamp_mtime_to: int,
1726 ) -> IntermediateManifest:
1727 self.apply_fs_transformations(package, fs_root)
1728 return self.finalize_data_tar_contents(package, fs_root, clamp_mtime_to)
1731@dataclasses.dataclass(slots=True)
1732class SearchDirOrderState:
1733 search_dir: VirtualPath
1734 applies_to: set[BinaryPackage] = dataclasses.field(default_factory=set)
1735 after: set[str] = dataclasses.field(default_factory=set)
1738def _present_installation_dirs(
1739 search_dirs: Sequence[SearchDir],
1740 checked_missing_dirs: Sequence[VirtualPath],
1741 all_pkgs: frozenset[BinaryPackage],
1742) -> None:
1743 _info("The following directories are considered search dirs (in order):")
1744 max_len = max((len(s.search_dir.fs_path) for s in search_dirs), default=1)
1745 for search_dir in search_dirs:
1746 applies_to = ""
1747 if search_dir.applies_to < all_pkgs:
1748 names = ", ".join(p.name for p in search_dir.applies_to)
1749 applies_to = f" [only applicable to: {names}]"
1750 remark = ""
1751 if not os.path.isdir(search_dir.search_dir.fs_path):
1752 remark = " (skipped; absent)"
1753 _info(f" * {search_dir.search_dir.fs_path:{max_len}}{applies_to}{remark}")
1755 if checked_missing_dirs:
1756 _info('The following directories are considered for "not-installed" paths;')
1757 for d in checked_missing_dirs:
1758 remark = ""
1759 if not os.path.isdir(d.fs_path):
1760 remark = " (skipped; absent)"
1761 _info(f" * {d.fs_path:{max_len}}{remark}")
1764def _determine_search_dir_order(
1765 requested: Mapping[BinaryPackage, list[VirtualPath]],
1766 all_pkgs: frozenset[BinaryPackage],
1767 source_root: VirtualPath,
1768) -> Sequence[SearchDir]:
1769 search_dir_table = {}
1770 assert requested.keys() <= all_pkgs
1771 for pkg in all_pkgs:
1772 paths = requested.get(pkg) or []
1773 previous_search_dir: SearchDirOrderState | None = None
1774 for path in paths:
1775 try:
1776 search_dir_state = search_dir_table[path.fs_path]
1777 except KeyError:
1778 search_dir_state = SearchDirOrderState(path)
1779 search_dir_table[path.fs_path] = search_dir_state
1780 search_dir_state.applies_to.add(pkg)
1781 if previous_search_dir is not None:
1782 search_dir_state.after.add(previous_search_dir.search_dir.fs_path)
1783 previous_search_dir = search_dir_state
1785 search_dirs_in_order = []
1786 released = set[str]()
1787 remaining = set()
1788 for search_dir_state in search_dir_table.values():
1789 if not (search_dir_state.after <= released):
1790 remaining.add(search_dir_state.search_dir.fs_path)
1791 continue
1792 search_dirs_in_order.append(search_dir_state)
1793 released.add(search_dir_state.search_dir.fs_path)
1795 while remaining:
1796 current_released = len(released)
1797 for fs_path in remaining:
1798 search_dir_state = search_dir_table[fs_path]
1799 if not search_dir_state.after.issubset(released):
1800 remaining.add(search_dir_state.search_dir.fs_path)
1801 continue
1802 search_dirs_in_order.append(search_dir_state)
1803 released.add(search_dir_state.search_dir.fs_path)
1805 if current_released == len(released):
1806 names = ", ".join(remaining)
1807 _error(
1808 f"There is a circular dependency (somewhere) between the search dirs: {names}."
1809 " Note that the search directories across all packages have to be ordered (and the"
1810 " source root should generally be last)"
1811 )
1812 remaining -= released
1814 search_dirs_in_order.append(
1815 SearchDirOrderState(
1816 source_root,
1817 set(all_pkgs),
1818 )
1819 )
1821 return tuple(
1822 # Avoid duplicating all_pkgs
1823 SearchDir(
1824 s.search_dir,
1825 frozenset(s.applies_to) if s.applies_to != all_pkgs else all_pkgs,
1826 )
1827 for s in search_dirs_in_order
1828 )