Coverage for src/debputy/plugin/api/impl_types.py: 78%
526 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-07 12:14 +0200
1import dataclasses
2import os.path
3import textwrap
4from typing import (
5 Optional,
6 Callable,
7 FrozenSet,
8 Dict,
9 List,
10 Tuple,
11 Generic,
12 TYPE_CHECKING,
13 TypeVar,
14 cast,
15 Any,
16 Sequence,
17 Union,
18 Type,
19 TypedDict,
20 Iterable,
21 Mapping,
22 NotRequired,
23 Literal,
24 Set,
25 Iterator,
26)
27from weakref import ref
29from debputy import DEBPUTY_DOC_ROOT_DIR
30from debputy.exceptions import (
31 DebputyFSIsROError,
32 PluginAPIViolationError,
33 PluginConflictError,
34 UnhandledOrUnexpectedErrorFromPluginError,
35)
36from debputy.filesystem_scan import as_path_def
37from debputy.installations import InstallRule
38from debputy.maintscript_snippet import DpkgMaintscriptHelperCommand
39from debputy.manifest_conditions import ManifestCondition
40from debputy.manifest_parser.base_types import DebputyParsedContent, TypeMapping
41from debputy.manifest_parser.exceptions import ManifestParseException
42from debputy.manifest_parser.util import AttributePath
43from debputy.packages import BinaryPackage
44from debputy.plugin.api import (
45 VirtualPath,
46 BinaryCtrlAccessor,
47 PackageProcessingContext,
48)
49from debputy.plugin.api.spec import (
50 DebputyPluginInitializer,
51 MetadataAutoDetector,
52 DpkgTriggerType,
53 ParserDocumentation,
54 PackageProcessor,
55 PathDef,
56 ParserAttributeDocumentation,
57 undocumented_attr,
58 documented_attr,
59 reference_documentation,
60 PackagerProvidedFileReferenceDocumentation,
61 TypeMappingDocumentation,
62)
63from debputy.substitution import VariableContext
64from debputy.transformation_rules import TransformationRule
65from debputy.util import _normalize_path, package_cross_check_precheck
67if TYPE_CHECKING:
68 from debputy.plugin.api.spec import (
69 ServiceDetector,
70 ServiceIntegrator,
71 PackageTypeSelector,
72 )
73 from debputy.manifest_parser.parser_data import ParserContextData
74 from debputy.highlevel_manifest import (
75 HighLevelManifest,
76 PackageTransformationDefinition,
77 BinaryPackageData,
78 )
81_PACKAGE_TYPE_DEB_ONLY = frozenset(["deb"])
82_ALL_PACKAGE_TYPES = frozenset(["deb", "udeb"])
85TD = TypeVar("TD", bound="Union[DebputyParsedContent, List[DebputyParsedContent]]")
86PF = TypeVar("PF")
87SF = TypeVar("SF")
88TP = TypeVar("TP")
89TTP = Type[TP]
91DIPKWHandler = Callable[[str, AttributePath, "ParserContextData"], TP]
92DIPHandler = Callable[[str, PF, AttributePath, "ParserContextData"], TP]
95def resolve_package_type_selectors(
96 package_type: "PackageTypeSelector",
97) -> FrozenSet[str]:
98 if package_type is _ALL_PACKAGE_TYPES or package_type is _PACKAGE_TYPE_DEB_ONLY:
99 return cast("FrozenSet[str]", package_type)
100 if isinstance(package_type, str):
101 return (
102 _PACKAGE_TYPE_DEB_ONLY
103 if package_type == "deb"
104 else frozenset([package_type])
105 )
106 else:
107 return frozenset(package_type)
110@dataclasses.dataclass(slots=True)
111class DebputyPluginMetadata:
112 plugin_name: str
113 api_compat_version: int
114 plugin_loader: Optional[Callable[[], Callable[["DebputyPluginInitializer"], None]]]
115 plugin_initializer: Optional[Callable[["DebputyPluginInitializer"], None]]
116 plugin_path: str
117 _is_initialized: bool = False
119 @property
120 def is_loaded(self) -> bool:
121 return self.plugin_initializer is not None
123 @property
124 def is_initialized(self) -> bool:
125 return self._is_initialized
127 def initialize_plugin(self, api: "DebputyPluginInitializer") -> None:
128 if self.is_initialized: 128 ↛ 129line 128 didn't jump to line 129, because the condition on line 128 was never true
129 raise RuntimeError("Cannot load plugins twice")
130 if not self.is_loaded:
131 self.load_plugin()
132 plugin_initializer = self.plugin_initializer
133 assert plugin_initializer is not None
134 plugin_initializer(api)
135 self._is_initialized = True
137 def load_plugin(self) -> None:
138 plugin_loader = self.plugin_loader
139 assert plugin_loader is not None
140 self.plugin_initializer = plugin_loader()
141 assert self.plugin_initializer is not None
144@dataclasses.dataclass(slots=True, frozen=True)
145class PluginProvidedParser(Generic[PF, TP]):
146 parser: "DeclarativeInputParser[PF]"
147 handler: Callable[[str, PF, "AttributePath", "ParserContextData"], TP]
148 plugin_metadata: DebputyPluginMetadata
150 def parse(
151 self,
152 name: str,
153 value: object,
154 attribute_path: "AttributePath",
155 *,
156 parser_context: "ParserContextData",
157 ) -> TP:
158 parsed_value = self.parser.parse_input(
159 value, attribute_path, parser_context=parser_context
160 )
161 return self.handler(name, parsed_value, attribute_path, parser_context)
164class PPFFormatParam(TypedDict):
165 priority: Optional[int]
166 name: str
167 owning_package: str
170@dataclasses.dataclass(slots=True, frozen=True)
171class PackagerProvidedFileClassSpec:
172 debputy_plugin_metadata: DebputyPluginMetadata
173 stem: str
174 installed_as_format: str
175 default_mode: int
176 default_priority: Optional[int]
177 allow_name_segment: bool
178 allow_architecture_segment: bool
179 post_formatting_rewrite: Optional[Callable[[str], str]]
180 packageless_is_fallback_for_all_packages: bool
181 reservation_only: bool
182 formatting_callback: Optional[Callable[[str, PPFFormatParam, VirtualPath], str]] = (
183 None
184 )
185 reference_documentation: Optional[PackagerProvidedFileReferenceDocumentation] = None
186 bug_950723: bool = False
188 @property
189 def supports_priority(self) -> bool:
190 return self.default_priority is not None
192 def compute_dest(
193 self,
194 assigned_name: str,
195 # Note this method is currently used 1:1 inside plugin tests.
196 *,
197 owning_package: Optional[str] = None,
198 assigned_priority: Optional[int] = None,
199 path: Optional[VirtualPath] = None,
200 ) -> Tuple[str, str]:
201 if assigned_priority is not None and not self.supports_priority: 201 ↛ 202line 201 didn't jump to line 202, because the condition on line 201 was never true
202 raise ValueError(
203 f"Cannot assign priority to packager provided files with stem"
204 f' "{self.stem}" (e.g., "debian/foo.{self.stem}"). They'
205 " do not use priority at all."
206 )
208 path_format = self.installed_as_format
209 if self.supports_priority and assigned_priority is None:
210 assigned_priority = self.default_priority
212 if owning_package is None:
213 owning_package = assigned_name
215 params: PPFFormatParam = {
216 "priority": assigned_priority,
217 "name": assigned_name,
218 "owning_package": owning_package,
219 }
221 if self.formatting_callback is not None:
222 if path is None: 222 ↛ 223line 222 didn't jump to line 223, because the condition on line 222 was never true
223 raise ValueError(
224 "The path parameter is required for PPFs with formatting_callback"
225 )
226 dest_path = self.formatting_callback(path_format, params, path)
227 else:
228 dest_path = path_format.format(**params)
230 dirname, basename = os.path.split(dest_path)
231 dirname = _normalize_path(dirname)
233 if self.post_formatting_rewrite:
234 basename = self.post_formatting_rewrite(basename)
235 return dirname, basename
238@dataclasses.dataclass(slots=True)
239class MetadataOrMaintscriptDetector:
240 plugin_metadata: DebputyPluginMetadata
241 detector_id: str
242 detector: MetadataAutoDetector
243 applies_to_package_types: FrozenSet[str]
244 enabled: bool = True
246 def applies_to(self, binary_package: BinaryPackage) -> bool:
247 return binary_package.package_type in self.applies_to_package_types
249 def run_detector(
250 self,
251 fs_root: "VirtualPath",
252 ctrl: "BinaryCtrlAccessor",
253 context: "PackageProcessingContext",
254 ) -> None:
255 try:
256 self.detector(fs_root, ctrl, context)
257 except DebputyFSIsROError as e: 257 ↛ 266line 257 didn't jump to line 266
258 nv = self.plugin_metadata.plugin_name
259 raise PluginAPIViolationError(
260 f'The plugin {nv} violated the API contract for "metadata detectors"'
261 " by attempting to mutate the provided file system in its metadata detector"
262 f" with id {self.detector_id}. File system mutation is *not* supported at"
263 " this stage (file system layout is committed and the attempted changes"
264 " would be lost)."
265 ) from e
266 except (ChildProcessError, RuntimeError, AttributeError) as e:
267 nv = f"{self.plugin_metadata.plugin_name}"
268 raise UnhandledOrUnexpectedErrorFromPluginError(
269 f"The plugin {nv} threw an unhandled or unexpected exception from its metadata"
270 f" detector with id {self.detector_id}."
271 ) from e
274class DeclarativeInputParser(Generic[TD]):
275 @property
276 def inline_reference_documentation(self) -> Optional[ParserDocumentation]:
277 return None
279 @property
280 def reference_documentation_url(self) -> Optional[str]:
281 doc = self.inline_reference_documentation
282 return doc.documentation_reference_url if doc is not None else None
284 def parse_input(
285 self,
286 value: object,
287 path: "AttributePath",
288 *,
289 parser_context: Optional["ParserContextData"] = None,
290 ) -> TD:
291 raise NotImplementedError
294class DelegatingDeclarativeInputParser(DeclarativeInputParser[TD]):
295 __slots__ = ("delegate", "_reference_documentation")
297 def __init__(
298 self,
299 delegate: DeclarativeInputParser[TD],
300 *,
301 inline_reference_documentation: Optional[ParserDocumentation] = None,
302 ) -> None:
303 self.delegate = delegate
304 self._reference_documentation = inline_reference_documentation
306 @property
307 def inline_reference_documentation(self) -> Optional[ParserDocumentation]:
308 doc = self._reference_documentation
309 if doc is None:
310 return self.delegate.inline_reference_documentation
311 return doc
314class ListWrappedDeclarativeInputParser(DelegatingDeclarativeInputParser[TD]):
315 __slots__ = ()
317 def _doc_url_error_suffix(self, *, see_url_version: bool = False) -> str:
318 doc_url = self.reference_documentation_url
319 if doc_url is not None: 319 ↛ 323line 319 didn't jump to line 323, because the condition on line 319 was never false
320 if see_url_version: 320 ↛ 322line 320 didn't jump to line 322, because the condition on line 320 was never false
321 return f" Please see {doc_url} for the documentation."
322 return f" (Documentation: {doc_url})"
323 return ""
325 def parse_input(
326 self,
327 value: object,
328 path: "AttributePath",
329 *,
330 parser_context: Optional["ParserContextData"] = None,
331 ) -> TD:
332 if not isinstance(value, list):
333 doc_ref = self._doc_url_error_suffix(see_url_version=True)
334 raise ManifestParseException(
335 f"The attribute {path.path} must be a list.{doc_ref}"
336 )
337 result = []
338 delegate = self.delegate
339 for idx, element in enumerate(value):
340 element_path = path[idx]
341 result.append(
342 delegate.parse_input(
343 element,
344 element_path,
345 parser_context=parser_context,
346 )
347 )
348 return result
351class DispatchingParserBase(Generic[TP]):
352 def __init__(self, manifest_attribute_path_template: str) -> None:
353 self.manifest_attribute_path_template = manifest_attribute_path_template
354 self._parsers: Dict[str, PluginProvidedParser[Any, TP]] = {}
356 def is_known_keyword(self, keyword: str) -> bool:
357 return keyword in self._parsers
359 def registered_keywords(self) -> Iterable[str]:
360 yield from self._parsers
362 def parser_for(self, keyword: str) -> PluginProvidedParser[Any, TP]:
363 return self._parsers[keyword]
365 def register_keyword(
366 self,
367 keyword: Union[str, Sequence[str]],
368 handler: DIPKWHandler,
369 plugin_metadata: DebputyPluginMetadata,
370 *,
371 inline_reference_documentation: Optional[ParserDocumentation] = None,
372 ) -> None:
373 reference_documentation_url = None
374 if inline_reference_documentation: 374 ↛ 386line 374 didn't jump to line 386, because the condition on line 374 was never false
375 if inline_reference_documentation.attribute_doc: 375 ↛ 376line 375 didn't jump to line 376, because the condition on line 375 was never true
376 raise ValueError(
377 "Cannot provide per-attribute documentation for a value-less keyword!"
378 )
379 if inline_reference_documentation.alt_parser_description: 379 ↛ 380line 379 didn't jump to line 380, because the condition on line 379 was never true
380 raise ValueError(
381 "Cannot provide non-mapping-format documentation for a value-less keyword!"
382 )
383 reference_documentation_url = (
384 inline_reference_documentation.documentation_reference_url
385 )
386 parser = DeclarativeValuelessKeywordInputParser(
387 inline_reference_documentation,
388 documentation_reference=reference_documentation_url,
389 )
391 def _combined_handler(
392 name: str,
393 _ignored: Any,
394 attr_path: AttributePath,
395 context: "ParserContextData",
396 ) -> TP:
397 return handler(name, attr_path, context)
399 p = PluginProvidedParser(
400 parser,
401 _combined_handler,
402 plugin_metadata,
403 )
405 self._add_parser(keyword, p)
407 def register_parser(
408 self,
409 keyword: Union[str, List[str]],
410 parser: "DeclarativeInputParser[PF]",
411 handler: Callable[[str, PF, "AttributePath", "ParserContextData"], TP],
412 plugin_metadata: DebputyPluginMetadata,
413 ) -> None:
414 p = PluginProvidedParser(
415 parser,
416 handler,
417 plugin_metadata,
418 )
419 self._add_parser(keyword, p)
421 def _add_parser(
422 self,
423 keyword: Union[str, List[str]],
424 ppp: "PluginProvidedParser[PF, TP]",
425 ) -> None:
426 ks = [keyword] if isinstance(keyword, str) else keyword
427 for k in ks:
428 existing_parser = self._parsers.get(k)
429 if existing_parser is not None: 429 ↛ 430line 429 didn't jump to line 430
430 message = (
431 f'The rule name "{k}" is already taken by the plugin'
432 f" {existing_parser.plugin_metadata.plugin_name}. This conflict was triggered"
433 f" when plugin {ppp.plugin_metadata.plugin_name} attempted to register its parser."
434 )
435 raise PluginConflictError(
436 message,
437 existing_parser.plugin_metadata,
438 ppp.plugin_metadata,
439 )
440 self._new_parser(k, ppp)
442 def _new_parser(self, keyword: str, ppp: "PluginProvidedParser[PF, TP]") -> None:
443 self._parsers[keyword] = ppp
445 def parse_input(
446 self,
447 orig_value: object,
448 attribute_path: "AttributePath",
449 *,
450 parser_context: "ParserContextData",
451 ) -> TP:
452 raise NotImplementedError
455class DispatchingObjectParser(
456 DispatchingParserBase[Mapping[str, Any]],
457 DeclarativeInputParser[Mapping[str, Any]],
458):
459 def __init__(
460 self,
461 manifest_attribute_path_template: str,
462 *,
463 parser_documentation: Optional[ParserDocumentation] = None,
464 ) -> None:
465 super().__init__(manifest_attribute_path_template)
466 self._attribute_documentation: List[ParserAttributeDocumentation] = []
467 if parser_documentation is None: 467 ↛ 468line 467 didn't jump to line 468, because the condition on line 467 was never true
468 parser_documentation = reference_documentation()
469 self._parser_documentation = parser_documentation
471 @property
472 def reference_documentation_url(self) -> Optional[str]:
473 return self._parser_documentation.documentation_reference_url
475 @property
476 def inline_reference_documentation(self) -> Optional[ParserDocumentation]:
477 ref_doc = self._parser_documentation
478 return reference_documentation(
479 title=ref_doc.title,
480 description=ref_doc.description,
481 attributes=self._attribute_documentation,
482 reference_documentation_url=self.reference_documentation_url,
483 )
485 def _new_parser(self, keyword: str, ppp: "PluginProvidedParser[PF, TP]") -> None:
486 super()._new_parser(keyword, ppp)
487 doc = ppp.parser.inline_reference_documentation
488 if doc is None or doc.description is None:
489 self._attribute_documentation.append(undocumented_attr(keyword))
490 else:
491 self._attribute_documentation.append(
492 documented_attr(keyword, doc.description)
493 )
495 def register_child_parser(
496 self,
497 keyword: str,
498 parser: "DispatchingObjectParser",
499 plugin_metadata: DebputyPluginMetadata,
500 *,
501 on_end_parse_step: Optional[
502 Callable[
503 [str, Optional[Mapping[str, Any]], AttributePath, "ParserContextData"],
504 None,
505 ]
506 ] = None,
507 nested_in_package_context: bool = False,
508 ) -> None:
509 def _handler(
510 name: str,
511 value: Mapping[str, Any],
512 path: AttributePath,
513 parser_context: "ParserContextData",
514 ) -> Mapping[str, Any]:
515 on_end_parse_step(name, value, path, parser_context)
516 return value
518 if nested_in_package_context:
519 parser = InPackageContextParser(
520 keyword,
521 parser,
522 )
524 p = PluginProvidedParser(
525 parser,
526 _handler,
527 plugin_metadata,
528 )
529 self._add_parser(keyword, p)
531 def parse_input(
532 self,
533 orig_value: object,
534 attribute_path: "AttributePath",
535 *,
536 parser_context: "ParserContextData",
537 ) -> TP:
538 doc_ref = ""
539 if self.reference_documentation_url is not None: 539 ↛ 543line 539 didn't jump to line 543, because the condition on line 539 was never false
540 doc_ref = (
541 f" Please see {self.reference_documentation_url} for the documentation."
542 )
543 if not isinstance(orig_value, dict):
544 raise ManifestParseException(
545 f"The attribute {attribute_path.path} must be a non-empty mapping.{doc_ref}"
546 )
547 if not orig_value: 547 ↛ 548line 547 didn't jump to line 548, because the condition on line 547 was never true
548 raise ManifestParseException(
549 f"The attribute {attribute_path.path} must be a non-empty mapping.{doc_ref}"
550 )
551 result = {}
552 unknown_keys = orig_value.keys() - self._parsers.keys()
553 if unknown_keys: 553 ↛ 554line 553 didn't jump to line 554, because the condition on line 553 was never true
554 first_key = next(iter(unknown_keys))
555 remaining_valid_attributes = self._parsers.keys() - orig_value.keys()
556 if not remaining_valid_attributes:
557 raise ManifestParseException(
558 f'The attribute "{first_key}" is not applicable at {attribute_path.path} (with the'
559 f" current set of plugins).{doc_ref}"
560 )
561 remaining_valid_attribute_names = ", ".join(remaining_valid_attributes)
562 raise ManifestParseException(
563 f'The attribute "{first_key}" is not applicable at {attribute_path.path}(with the current set'
564 " of plugins). Possible attributes available (and not already used) are:"
565 f" {remaining_valid_attribute_names}.{doc_ref}"
566 )
567 # Parse order is important for the root level (currently we use rule registration order)
568 for key, provided_parser in self._parsers.items():
569 value = orig_value.get(key)
570 if value is None:
571 if isinstance(provided_parser.parser, DispatchingObjectParser):
572 provided_parser.handler(
573 key, {}, attribute_path[key], parser_context
574 )
575 continue
576 value_path = attribute_path[key]
577 if provided_parser is None: 577 ↛ 578line 577 didn't jump to line 578, because the condition on line 577 was never true
578 valid_keys = ", ".join(sorted(self._parsers.keys()))
579 raise ManifestParseException(
580 f'Unknown or unsupported option "{key}" at {value_path.path}.'
581 " Valid options at this location are:"
582 f" {valid_keys}\n{doc_ref}"
583 )
584 parsed_value = provided_parser.parse(
585 key, value, value_path, parser_context=parser_context
586 )
587 result[key] = parsed_value
588 return result
591@dataclasses.dataclass(slots=True, frozen=True)
592class PackageContextData(Generic[TP]):
593 resolved_package_name: str
594 value: TP
597class InPackageContextParser(
598 DelegatingDeclarativeInputParser[Mapping[str, PackageContextData[TP]]]
599):
600 def __init__(
601 self,
602 manifest_attribute_path_template: str,
603 delegate: DeclarativeInputParser[TP],
604 *,
605 parser_documentation: Optional[ParserDocumentation] = None,
606 ) -> None:
607 self.manifest_attribute_path_template = manifest_attribute_path_template
608 self._attribute_documentation: List[ParserAttributeDocumentation] = []
609 super().__init__(delegate, inline_reference_documentation=parser_documentation)
611 def parse_input(
612 self,
613 orig_value: object,
614 attribute_path: "AttributePath",
615 *,
616 parser_context: Optional["ParserContextData"] = None,
617 ) -> TP:
618 assert parser_context is not None
619 doc_ref = ""
620 if self.reference_documentation_url is not None: 620 ↛ 624line 620 didn't jump to line 624, because the condition on line 620 was never false
621 doc_ref = (
622 f" Please see {self.reference_documentation_url} for the documentation."
623 )
624 if not isinstance(orig_value, dict) or not orig_value: 624 ↛ 625line 624 didn't jump to line 625, because the condition on line 624 was never true
625 raise ManifestParseException(
626 f"The attribute {attribute_path.path} must be a non-empty mapping.{doc_ref}"
627 )
628 delegate = self.delegate
629 result = {}
630 for package_name_raw, value in orig_value.items():
632 definition_source = attribute_path[package_name_raw]
633 package_name = package_name_raw
634 if "{{" in package_name:
635 package_name = parser_context.substitution.substitute(
636 package_name_raw,
637 definition_source.path,
638 )
639 package_state: PackageTransformationDefinition
640 with parser_context.binary_package_context(package_name) as package_state:
641 if package_state.is_auto_generated_package: 641 ↛ 643line 641 didn't jump to line 643, because the condition on line 641 was never true
642 # Maybe lift (part) of this restriction.
643 raise ManifestParseException(
644 f'Cannot define rules for package "{package_name}" (at {definition_source.path}). It is an'
645 " auto-generated package."
646 )
647 parsed_value = delegate.parse_input(
648 value, definition_source, parser_context=parser_context
649 )
650 result[package_name_raw] = PackageContextData(
651 package_name, parsed_value
652 )
653 return result
656class DispatchingTableParser(
657 DispatchingParserBase[TP],
658 DeclarativeInputParser[TP],
659):
660 def __init__(self, base_type: TTP, manifest_attribute_path_template: str) -> None:
661 super().__init__(manifest_attribute_path_template)
662 self.base_type = base_type
664 def parse_input(
665 self,
666 orig_value: object,
667 attribute_path: "AttributePath",
668 *,
669 parser_context: "ParserContextData",
670 ) -> TP:
671 if isinstance(orig_value, str): 671 ↛ 672line 671 didn't jump to line 672, because the condition on line 671 was never true
672 key = orig_value
673 value = None
674 value_path = attribute_path
675 elif isinstance(orig_value, dict): 675 ↛ 686line 675 didn't jump to line 686, because the condition on line 675 was never false
676 if len(orig_value) != 1: 676 ↛ 677line 676 didn't jump to line 677, because the condition on line 676 was never true
677 valid_keys = ", ".join(sorted(self._parsers.keys()))
678 raise ManifestParseException(
679 f'The mapping "{attribute_path.path}" had two keys, but it should only have one top level key.'
680 " Maybe you are missing a list marker behind the second key or some indentation. The"
681 f" possible keys are: {valid_keys}"
682 )
683 key, value = next(iter(orig_value.items()))
684 value_path = attribute_path[key]
685 else:
686 raise ManifestParseException(
687 f"The attribute {attribute_path.path} must be a string or a mapping."
688 )
689 provided_parser = self._parsers.get(key)
690 if provided_parser is None: 690 ↛ 691line 690 didn't jump to line 691, because the condition on line 690 was never true
691 valid_keys = ", ".join(sorted(self._parsers.keys()))
692 raise ManifestParseException(
693 f'Unknown or unsupported action "{key}" at {value_path.path}.'
694 " Valid actions at this location are:"
695 f" {valid_keys}"
696 )
697 return provided_parser.parse(
698 key, value, value_path, parser_context=parser_context
699 )
702@dataclasses.dataclass(slots=True)
703class DeclarativeValuelessKeywordInputParser(DeclarativeInputParser[None]):
704 inline_reference_documentation: Optional[ParserDocumentation] = None
705 documentation_reference: Optional[str] = None
707 def parse_input(
708 self,
709 value: object,
710 path: "AttributePath",
711 *,
712 parser_context: Optional["ParserContextData"] = None,
713 ) -> TD:
714 if value is None:
715 return cast("TD", value)
716 if self.documentation_reference is not None:
717 doc_ref = f" (Documentation: {self.documentation_reference})"
718 else:
719 doc_ref = ""
720 raise ManifestParseException(
721 f"Expected attribute {path.path} to be a string.{doc_ref}"
722 )
725SUPPORTED_DISPATCHABLE_TABLE_PARSERS = {
726 InstallRule: "installations",
727 TransformationRule: "packages.{{PACKAGE}}.transformations",
728 DpkgMaintscriptHelperCommand: "packages.{{PACKAGE}}.conffile-management",
729 ManifestCondition: "*.when",
730}
732OPARSER_MANIFEST_ROOT = "<ROOT>"
733OPARSER_PACKAGES_ROOT = "packages"
734OPARSER_PACKAGES = "packages.{{PACKAGE}}"
735OPARSER_MANIFEST_DEFINITIONS = "definitions"
737SUPPORTED_DISPATCHABLE_OBJECT_PARSERS = {
738 OPARSER_MANIFEST_ROOT: reference_documentation(
739 reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md",
740 ),
741 OPARSER_MANIFEST_DEFINITIONS: reference_documentation(
742 title="Packager provided definitions",
743 description="Reusable packager provided definitions such as manifest variables.",
744 reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md#packager-provided-definitions",
745 ),
746 OPARSER_PACKAGES: reference_documentation(
747 title="Binary package rules",
748 description=textwrap.dedent(
749 """\
750 Inside the manifest, the `packages` mapping can be used to define requests for the binary packages
751 you want `debputy` to produce. Each key inside `packages` must be the name of a binary package
752 defined in `debian/control`. The value is a dictionary defining which features that `debputy`
753 should apply to that binary package. An example could be:
755 packages:
756 foo:
757 transformations:
758 - create-symlink:
759 path: usr/share/foo/my-first-symlink
760 target: /usr/share/bar/symlink-target
761 - create-symlink:
762 path: usr/lib/{{DEB_HOST_MULTIARCH}}/my-second-symlink
763 target: /usr/lib/{{DEB_HOST_MULTIARCH}}/baz/symlink-target
764 bar:
765 transformations:
766 - create-directories:
767 - some/empty/directory.d
768 - another/empty/integration-point.d
769 - create-directories:
770 path: a/third-empty/directory.d
771 owner: www-data
772 group: www-data
774 In this case, `debputy` will create some symlinks inside the `foo` package and some directories for
775 the `bar` package. The following subsections define the keys you can use under each binary package.
776 """
777 ),
778 reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md#binary-package-rules",
779 ),
780}
783@dataclasses.dataclass(slots=True)
784class PluginProvidedManifestVariable:
785 plugin_metadata: DebputyPluginMetadata
786 variable_name: str
787 variable_value: Optional[Union[str, Callable[[VariableContext], str]]]
788 is_context_specific_variable: bool
789 variable_reference_documentation: Optional[str] = None
790 is_documentation_placeholder: bool = False
791 is_for_special_case: bool = False
793 @property
794 def is_internal(self) -> bool:
795 return self.variable_name.startswith("_") or ":_" in self.variable_name
797 @property
798 def is_token(self) -> bool:
799 return self.variable_name.startswith("token:")
801 def resolve(self, variable_context: VariableContext) -> str:
802 value_resolver = self.variable_value
803 if isinstance(value_resolver, str):
804 res = value_resolver
805 else:
806 res = value_resolver(variable_context)
807 return res
810@dataclasses.dataclass(slots=True, frozen=True)
811class AutomaticDiscardRuleExample:
812 content: Sequence[Tuple[PathDef, bool]]
813 description: Optional[str] = None
816def automatic_discard_rule_example(
817 *content: Union[str, PathDef, Tuple[Union[str, PathDef], bool]],
818 example_description: Optional[str] = None,
819) -> AutomaticDiscardRuleExample:
820 """Provide an example for an automatic discard rule
822 The return value of this method should be passed to the `examples` parameter of
823 `automatic_discard_rule` method - either directly for a single example or as a
824 part of a sequence of examples.
826 >>> # Possible example for an exclude rule for ".la" files
827 >>> # Example shows two files; The ".la" file that will be removed and another file that
828 >>> # will be kept.
829 >>> automatic_discard_rule_example( # doctest: +ELLIPSIS
830 ... "usr/lib/libfoo.la",
831 ... ("usr/lib/libfoo.so.1.0.0", False),
832 ... )
833 AutomaticDiscardRuleExample(...)
835 Keep in mind that you have to explicitly include directories that are relevant for the test
836 if you want them shown. Also, if a directory is excluded, all path beneath it will be
837 automatically excluded in the example as well. Your example data must account for that.
839 >>> # Possible example for python cache file discard rule
840 >>> # In this example, we explicitly list the __pycache__ directory itself because we
841 >>> # want it shown in the output (otherwise, we could have omitted it)
842 >>> automatic_discard_rule_example( # doctest: +ELLIPSIS
843 ... (".../foo.py", False),
844 ... ".../__pycache__/",
845 ... ".../__pycache__/...",
846 ... ".../foo.pyc",
847 ... ".../foo.pyo",
848 ... )
849 AutomaticDiscardRuleExample(...)
851 Note: Even if `__pycache__` had been implicit, the result would have been the same. However,
852 the rendered example would not have shown the directory on its own. The use of `...` as
853 path names is useful for denoting "anywhere" or "anything". Though, there is nothing "magic"
854 about this name - it happens to be allowed as a path name (unlike `.` or `..`).
856 These examples can be seen via `debputy plugin show automatic-discard-rules <name-here>`.
858 :param content: The content of the example. Each element can be either a path definition or
859 a tuple of a path definition followed by a verdict (boolean). Each provided path definition
860 describes the paths to be presented in the example. Implicit paths such as parent
861 directories will be created but not shown in the example. Therefore, if a directory is
862 relevant to the example, be sure to explicitly list it.
864 The verdict associated with a path determines whether the path should be discarded (when
865 True) or kept (when False). When a path is not explicitly associated with a verdict, the
866 verdict is assumed to be discarded (True).
867 :param example_description: An optional description displayed together with the example.
868 :return: An opaque data structure containing the example.
869 """
870 example = []
871 for d in content:
872 if not isinstance(d, tuple):
873 pd = d
874 verdict = True
875 else:
876 pd, verdict = d
878 path_def = as_path_def(pd)
879 example.append((path_def, verdict))
881 if not example: 881 ↛ 882line 881 didn't jump to line 882, because the condition on line 881 was never true
882 raise ValueError("At least one path must be given for an example")
884 return AutomaticDiscardRuleExample(
885 tuple(example),
886 description=example_description,
887 )
890@dataclasses.dataclass(slots=True, frozen=True)
891class PluginProvidedPackageProcessor:
892 processor_id: str
893 applies_to_package_types: FrozenSet[str]
894 package_processor: PackageProcessor
895 dependencies: FrozenSet[Tuple[str, str]]
896 plugin_metadata: DebputyPluginMetadata
898 def applies_to(self, binary_package: BinaryPackage) -> bool:
899 return binary_package.package_type in self.applies_to_package_types
901 @property
902 def dependency_id(self) -> Tuple[str, str]:
903 return self.plugin_metadata.plugin_name, self.processor_id
905 def run_package_processor(
906 self,
907 fs_root: "VirtualPath",
908 unused: None,
909 context: "PackageProcessingContext",
910 ) -> None:
911 self.package_processor(fs_root, unused, context)
914@dataclasses.dataclass(slots=True, frozen=True)
915class PluginProvidedDiscardRule:
916 name: str
917 plugin_metadata: DebputyPluginMetadata
918 discard_check: Callable[[VirtualPath], bool]
919 reference_documentation: Optional[str]
920 examples: Sequence[AutomaticDiscardRuleExample] = tuple()
922 def should_discard(self, path: VirtualPath) -> bool:
923 return self.discard_check(path)
926@dataclasses.dataclass(slots=True, frozen=True)
927class ServiceManagerDetails:
928 service_manager: str
929 service_detector: "ServiceDetector"
930 service_integrator: "ServiceIntegrator"
931 plugin_metadata: DebputyPluginMetadata
934ReferenceValue = TypedDict(
935 "ReferenceValue",
936 {
937 "description": str,
938 },
939)
942def _reference_data_value(
943 *,
944 description: str,
945) -> ReferenceValue:
946 return {
947 "description": description,
948 }
951KnownPackagingFileCategories = Literal[
952 "generated",
953 "generic-template",
954 "ppf-file",
955 "ppf-control-file",
956 "maint-config",
957 "pkg-metadata",
958 "pkg-helper-config",
959 "testing",
960 "lint-config",
961]
962KNOWN_PACKAGING_FILE_CATEGORY_DESCRIPTIONS: Mapping[
963 KnownPackagingFileCategories, ReferenceValue
964] = {
965 "generated": _reference_data_value(
966 description="The file is (likely) generated from another file"
967 ),
968 "generic-template": _reference_data_value(
969 description="The file is (likely) a generic template that generates a known packaging file. While the"
970 " file is annotated as if it was the target file, the file might uses a custom template"
971 " language inside it."
972 ),
973 "ppf-file": _reference_data_value(
974 description="Packager provided file to be installed on the file system - usually as-is."
975 " When `install-pattern` or `install-path` are provided, this is where the file is installed."
976 ),
977 "ppf-control-file": _reference_data_value(
978 description="Packager provided file that becomes a control file - possible after processing. "
979 " If `install-pattern` or `install-path` are provided, they denote where the is placed"
980 " (generally, this will be of the form `DEBIAN/<name>`)"
981 ),
982 "maint-config": _reference_data_value(
983 description="Maintenance configuration for a specific tool that the maintainer uses (tool / style preferences)"
984 ),
985 "pkg-metadata": _reference_data_value(
986 description="The file is related to standard package metadata (usually documented in Debian Policy)"
987 ),
988 "pkg-helper-config": _reference_data_value(
989 description="The file is packaging helper configuration or instruction file"
990 ),
991 "testing": _reference_data_value(
992 description="The file is related to automated testing (autopkgtests, salsa/gitlab CI)."
993 ),
994 "lint-config": _reference_data_value(
995 description="The file is related to a linter (such as overrides for false-positives or style preferences)"
996 ),
997}
999KnownPackagingConfigFeature = Literal[
1000 "dh-filearray",
1001 "dh-filedoublearray",
1002 "dh-hash-subst",
1003 "dh-dollar-subst",
1004 "dh-glob",
1005 "dh-partial-glob",
1006 "dh-late-glob",
1007 "dh-glob-after-execute",
1008 "dh-executable-config",
1009 "dh-custom-format",
1010 "dh-file-list",
1011 "dh-install-list",
1012 "dh-install-list-dest-dir-like-dh_install",
1013 "dh-install-list-fixed-dest-dir",
1014 "dh-fixed-dest-dir",
1015 "dh-exec-rename",
1016 "dh-docs-only",
1017]
1019KNOWN_PACKAGING_FILE_CONFIG_FEATURE_DESCRIPTION: Mapping[
1020 KnownPackagingConfigFeature, ReferenceValue
1021] = {
1022 "dh-filearray": _reference_data_value(
1023 description="The file will be read as a list of space/newline separated tokens",
1024 ),
1025 "dh-filedoublearray": _reference_data_value(
1026 description="Each line in the file will be read as a list of space-separated tokens",
1027 ),
1028 "dh-hash-subst": _reference_data_value(
1029 description="Supports debhelper #PACKAGE# style substitutions (udebs often excluded)",
1030 ),
1031 "dh-dollar-subst": _reference_data_value(
1032 description="Supports debhelper ${PACKAGE} style substitutions (usually requires compat 13+)",
1033 ),
1034 "dh-glob": _reference_data_value(
1035 description="Supports standard debhelper globing",
1036 ),
1037 "dh-partial-glob": _reference_data_value(
1038 description="Supports standard debhelper globing but only to a subset of the values (implies dh-late-glob)",
1039 ),
1040 "dh-late-glob": _reference_data_value(
1041 description="Globbing is done separately instead of using the built-in function",
1042 ),
1043 "dh-glob-after-execute": _reference_data_value(
1044 description="When the dh config file is executable, the generated output will be subject to globbing",
1045 ),
1046 "dh-executable-config": _reference_data_value(
1047 description="If marked executable, debhelper will execute the file and read its output",
1048 ),
1049 "dh-custom-format": _reference_data_value(
1050 description="The dh tool will or may have a custom parser for this file",
1051 ),
1052 "dh-file-list": _reference_data_value(
1053 description="The dh file contains a list of paths to be processed",
1054 ),
1055 "dh-install-list": _reference_data_value(
1056 description="The dh file contains a list of paths/globs to be installed but the tool specific knowledge"
1057 " required to understand the file cannot be conveyed via this interface.",
1058 ),
1059 "dh-install-list-dest-dir-like-dh_install": _reference_data_value(
1060 description="The dh file is processed similar to dh_install (notably dest-dir handling derived"
1061 " from the path or the last token on the line)",
1062 ),
1063 "dh-install-list-fixed-dest-dir": _reference_data_value(
1064 description="The dh file is an install list and the dest-dir is always the same for all patterns"
1065 " (when `install-pattern` or `install-path` are provided, they identify the directory - not the file location)",
1066 ),
1067 "dh-exec-rename": _reference_data_value(
1068 description="When `dh-exec` is the interpreter of this dh config file, its renaming (=>) feature can be"
1069 " requested/used",
1070 ),
1071 "dh-docs-only": _reference_data_value(
1072 description="The dh config file is used for documentation only. Implicit <!nodocs> Build-Profiles support",
1073 ),
1074}
1076CONFIG_FEATURE_ALIASES: Dict[
1077 KnownPackagingConfigFeature, List[Tuple[KnownPackagingConfigFeature, int]]
1078] = {
1079 "dh-filearray": [
1080 ("dh-filearray", 0),
1081 ("dh-executable-config", 9),
1082 ("dh-dollar-subst", 13),
1083 ],
1084 "dh-filedoublearray": [
1085 ("dh-filedoublearray", 0),
1086 ("dh-executable-config", 9),
1087 ("dh-dollar-subst", 13),
1088 ],
1089}
1092def _implies(
1093 features: List[KnownPackagingConfigFeature],
1094 seen: Set[KnownPackagingConfigFeature],
1095 implying: Sequence[KnownPackagingConfigFeature],
1096 implied: KnownPackagingConfigFeature,
1097) -> None:
1098 if implied in seen:
1099 return
1100 if all(f in seen for f in implying):
1101 seen.add(implied)
1102 features.append(implied)
1105def expand_known_packaging_config_features(
1106 compat_level: int,
1107 features: List[KnownPackagingConfigFeature],
1108) -> List[KnownPackagingConfigFeature]:
1109 final_features: List[KnownPackagingConfigFeature] = []
1110 seen = set()
1111 for feature in features:
1112 expanded = CONFIG_FEATURE_ALIASES.get(feature)
1113 if not expanded:
1114 expanded = [(feature, 0)]
1115 for v, c in expanded:
1116 if compat_level < c or v in seen:
1117 continue
1118 seen.add(v)
1119 final_features.append(v)
1120 if "dh-glob" in seen and "dh-late-glob" in seen:
1121 final_features.remove("dh-glob")
1123 _implies(final_features, seen, ["dh-partial-glob"], "dh-late-glob")
1124 _implies(
1125 final_features,
1126 seen,
1127 ["dh-late-glob", "dh-executable-config"],
1128 "dh-glob-after-execute",
1129 )
1130 return sorted(final_features)
1133class InstallPatternDHCompatRule(DebputyParsedContent):
1134 install_pattern: NotRequired[str]
1135 add_config_features: NotRequired[List[KnownPackagingConfigFeature]]
1136 starting_with_compat_level: NotRequired[int]
1139class KnownPackagingFileInfo(DebputyParsedContent):
1140 # Exposed directly in the JSON plugin parsing; be careful with changes
1141 path: NotRequired[str]
1142 pkgfile: NotRequired[str]
1143 detection_method: NotRequired[Literal["path", "dh.pkgfile"]]
1144 file_categories: NotRequired[List[KnownPackagingFileCategories]]
1145 documentation_uris: NotRequired[List[str]]
1146 debputy_cmd_templates: NotRequired[List[List[str]]]
1147 debhelper_commands: NotRequired[List[str]]
1148 config_features: NotRequired[List[KnownPackagingConfigFeature]]
1149 install_pattern: NotRequired[str]
1150 dh_compat_rules: NotRequired[List[InstallPatternDHCompatRule]]
1151 default_priority: NotRequired[int]
1152 post_formatting_rewrite: NotRequired[Literal["period-to-underscore"]]
1153 packageless_is_fallback_for_all_packages: NotRequired[bool]
1156@dataclasses.dataclass(slots=True)
1157class PluginProvidedKnownPackagingFile:
1158 info: KnownPackagingFileInfo
1159 detection_method: Literal["path", "dh.pkgfile"]
1160 detection_value: str
1161 plugin_metadata: DebputyPluginMetadata
1164@dataclasses.dataclass(slots=True, frozen=True)
1165class PluginProvidedTypeMapping:
1166 mapped_type: TypeMapping[Any, Any]
1167 reference_documentation: Optional[TypeMappingDocumentation]
1168 plugin_metadata: DebputyPluginMetadata
1171class PackageDataTable:
1172 def __init__(self, package_data_table: Mapping[str, "BinaryPackageData"]) -> None:
1173 self._package_data_table = package_data_table
1174 # This is enabled for metadata-detectors. But it is deliberate not enabled for package processors,
1175 # because it is not clear how it should interact with dependencies. For metadata-detectors, things
1176 # read-only and there are no dependencies, so we cannot "get them wrong".
1177 self.enable_cross_package_checks = False
1179 def __iter__(self) -> Iterator["BinaryPackageData"]:
1180 return iter(self._package_data_table.values())
1182 def __getitem__(self, item: str) -> "BinaryPackageData":
1183 return self._package_data_table[item]
1185 def __contains__(self, item: str) -> bool:
1186 return item in self._package_data_table
1189class PackageProcessingContextProvider(PackageProcessingContext):
1190 __slots__ = (
1191 "_manifest",
1192 "_binary_package",
1193 "_related_udeb_package",
1194 "_package_data_table",
1195 "_cross_check_cache",
1196 )
1198 def __init__(
1199 self,
1200 manifest: "HighLevelManifest",
1201 binary_package: BinaryPackage,
1202 related_udeb_package: Optional[BinaryPackage],
1203 package_data_table: PackageDataTable,
1204 ) -> None:
1205 self._manifest = manifest
1206 self._binary_package = binary_package
1207 self._related_udeb_package = related_udeb_package
1208 self._package_data_table = ref(package_data_table)
1209 self._cross_check_cache: Optional[
1210 Sequence[Tuple[BinaryPackage, "VirtualPath"]]
1211 ] = None
1213 def _package_state_for(
1214 self,
1215 package: BinaryPackage,
1216 ) -> "PackageTransformationDefinition":
1217 return self._manifest.package_state_for(package.name)
1219 def _package_version_for(
1220 self,
1221 package: BinaryPackage,
1222 ) -> str:
1223 package_state = self._package_state_for(package)
1224 version = package_state.binary_version
1225 if version is not None:
1226 return version
1227 return self._manifest.source_version(
1228 include_binnmu_version=not package.is_arch_all
1229 )
1231 @property
1232 def binary_package(self) -> BinaryPackage:
1233 return self._binary_package
1235 @property
1236 def related_udeb_package(self) -> Optional[BinaryPackage]:
1237 return self._related_udeb_package
1239 @property
1240 def binary_package_version(self) -> str:
1241 return self._package_version_for(self._binary_package)
1243 @property
1244 def related_udeb_package_version(self) -> Optional[str]:
1245 udeb = self._related_udeb_package
1246 if udeb is None:
1247 return None
1248 return self._package_version_for(udeb)
1250 def accessible_package_roots(self) -> Iterable[Tuple[BinaryPackage, "VirtualPath"]]:
1251 package_table = self._package_data_table()
1252 if package_table is None:
1253 raise ReferenceError(
1254 "Internal error: package_table was garbage collected too early"
1255 )
1256 if not package_table.enable_cross_package_checks:
1257 raise PluginAPIViolationError(
1258 "Cross package content checks are not available at this time."
1259 )
1260 cache = self._cross_check_cache
1261 if cache is None:
1262 matches = []
1263 pkg = self.binary_package
1264 for pkg_data in package_table:
1265 if pkg_data.binary_package.name == pkg.name:
1266 continue
1267 res = package_cross_check_precheck(pkg, pkg_data.binary_package)
1268 if not res[0]:
1269 continue
1270 matches.append((pkg_data.binary_package, pkg_data.fs_root))
1271 cache = tuple(matches) if matches else tuple()
1272 self._cross_check_cache = cache
1273 return cache
1276@dataclasses.dataclass(slots=True, frozen=True)
1277class PluginProvidedTrigger:
1278 dpkg_trigger_type: DpkgTriggerType
1279 dpkg_trigger_target: str
1280 provider: DebputyPluginMetadata
1281 provider_source_id: str
1283 def serialized_format(self) -> str:
1284 return f"{self.dpkg_trigger_type} {self.dpkg_trigger_target}"