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