Coverage for src/debputy/plugin/api/impl.py: 60%
900 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-19 09:13 +0000
1import contextlib
2import dataclasses
3import functools
4import importlib
5import importlib.resources
6import importlib.util
7import inspect
8import itertools
9import json
10import os
11import re
12import subprocess
13import sys
14from abc import ABC
15from collections.abc import Callable, Iterable, Sequence, Iterator, Mapping, Container
16from importlib.resources.abc import Traversable
17from io import IOBase
18from json import JSONDecodeError
19from pathlib import Path
20from types import NoneType
21from typing import (
22 IO,
23 AbstractSet,
24 cast,
25 Any,
26 Literal,
27 TYPE_CHECKING,
28 is_typeddict,
29 AnyStr,
30 overload,
31)
33import debputy
34from debputy.exceptions import (
35 DebputySubstitutionError,
36 PluginConflictError,
37 PluginMetadataError,
38 PluginBaseError,
39 PluginInitializationError,
40 PluginAPIViolationError,
41 PluginNotFoundError,
42 PluginIncorrectRegistrationError,
43)
44from debputy.maintscript_snippet import (
45 DPKG_DEB_CONTROL_SCRIPTS,
46 MaintscriptSnippetContainer,
47 UnboundMaintscriptSnippet,
48 SnippetResolver,
49 SnippetAnchor,
50 PackageMaintscriptSnippetContainer,
51)
52from debputy.manifest_parser.exceptions import ManifestParseException
53from debputy.manifest_parser.parser_data import ParserContextData
54from debputy.manifest_parser.tagging_types import TypeMapping
55from debputy.manifest_parser.util import AttributePath
56from debputy.plugin.api.doc_parsing import (
57 DEBPUTY_DOC_REFERENCE_DATA_PARSER,
58 parser_type_name,
59 DebputyParsedDoc,
60)
61from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
62from debputy.plugin.api.impl_types import (
63 DebputyPluginMetadata,
64 PackagerProvidedFileClassSpec,
65 MetadataOrMaintscriptDetector,
66 PluginProvidedTrigger,
67 TTP,
68 DIPHandler,
69 PF,
70 SF,
71 DIPKWHandler,
72 PluginProvidedManifestVariable,
73 PluginProvidedPackageProcessor,
74 PluginProvidedDiscardRule,
75 AutomaticDiscardRuleExample,
76 PPFFormatParam,
77 ServiceManagerDetails,
78 KnownPackagingFileInfo,
79 PluginProvidedKnownPackagingFile,
80 DHCompatibilityBasedRule,
81 PluginProvidedTypeMapping,
82 PluginProvidedBuildSystemAutoDetection,
83 BSR,
84 TP,
85)
86from debputy.plugin.api.plugin_parser import (
87 PLUGIN_METADATA_PARSER,
88 PluginJsonMetadata,
89 PLUGIN_PPF_PARSER,
90 PackagerProvidedFileJsonDescription,
91 PLUGIN_MANIFEST_VARS_PARSER,
92 PLUGIN_KNOWN_PACKAGING_FILES_PARSER,
93)
94from debputy.plugin.api.spec import (
95 MaintscriptAccessor,
96 Maintscript,
97 DpkgTriggerType,
98 BinaryCtrlAccessor,
99 PackageProcessingContext,
100 MetadataAutoDetector,
101 PluginInitializationEntryPoint,
102 DebputyPluginInitializer,
103 FlushableSubstvars,
104 ParserDocumentation,
105 PackageProcessor,
106 VirtualPath,
107 ServiceIntegrator,
108 ServiceDetector,
109 ServiceRegistry,
110 ServiceDefinition,
111 DSD,
112 ServiceUpgradeRule,
113 PackagerProvidedFileReferenceDocumentation,
114 packager_provided_file_reference_documentation,
115 TypeMappingDocumentation,
116 DebputyIntegrationMode,
117 _DEBPUTY_DISPATCH_METADATA_ATTR_NAME,
118 BuildSystemManifestRuleMetadata,
119 INTEGRATION_MODE_FULL,
120 only_integrations,
121 DebputyPluginDefinition,
122)
123from debputy.plugin.api.std_docs import _STD_ATTR_DOCS
124from debputy.plugin.plugin_state import (
125 run_in_context_of_plugin,
126 run_in_context_of_plugin_wrap_errors,
127 wrap_plugin_code,
128 register_manifest_type_value_in_context,
129)
130from debputy.plugins.debputy.to_be_api_types import (
131 BuildRuleParsedFormat,
132 BSPF,
133 debputy_build_system,
134)
135from debputy.substitution import (
136 Substitution,
137 VariableNameState,
138 SUBST_VAR_RE,
139 VariableContext,
140)
141from debputy.util import (
142 _normalize_path,
143 POSTINST_DEFAULT_CONDITION,
144 _error,
145 print_command,
146 _warn,
147 _debug_log,
148 PackageTypeSelector,
149)
150from debputy.version import debputy_doc_root_dir
151from debputy.yaml import MANIFEST_YAML
153if TYPE_CHECKING:
154 from debputy.highlevel_manifest import HighLevelManifest
156PLUGIN_TEST_SUFFIX = re.compile(r"_(?:t|test|check)(?:_([a-z0-9_]+))?[.]py$")
157PLUGIN_PYTHON_RES_PATH = importlib.resources.files(debputy.plugins.__name__)
160def _validate_known_packaging_file_dh_compat_rules(
161 dh_compat_rules: list[DHCompatibilityBasedRule] | None,
162) -> None:
163 max_compat = None
164 if not dh_compat_rules: 164 ↛ 167line 164 didn't jump to line 167 because the condition on line 164 was always true
165 return
166 dh_compat_rule: DHCompatibilityBasedRule
167 for idx, dh_compat_rule in enumerate(dh_compat_rules):
168 dh_version = dh_compat_rule.get("starting_with_debhelper_version")
169 compat = dh_compat_rule.get("starting_with_compat_level")
171 remaining = dh_compat_rule.keys() - {
172 "after_debhelper_version",
173 "starting_with_compat_level",
174 }
175 if not remaining:
176 raise ValueError(
177 f"The dh compat-rule at index {idx} does not affect anything / not have any rules!? So why have it?"
178 )
179 if dh_version is None and compat is None and idx < len(dh_compat_rules) - 1:
180 raise ValueError(
181 f"The dh compat-rule at index {idx} is not the last and is missing either"
182 " before-debhelper-version or before-compat-level"
183 )
184 if compat is not None and compat < 0:
185 raise ValueError(
186 f"There is no compat below 1 but dh compat-rule at {idx} wants to declare some rule"
187 f" for something that appeared when migrating from {compat} to {compat + 1}."
188 )
190 if max_compat is None:
191 max_compat = compat
192 elif compat is not None:
193 if compat >= max_compat:
194 raise ValueError(
195 f"The dh compat-rule at {idx} should be moved earlier than the entry for compat {max_compat}."
196 )
197 max_compat = compat
199 install_pattern = dh_compat_rule.get("install_pattern")
200 if (
201 install_pattern is not None
202 and _normalize_path(install_pattern, with_prefix=False) != install_pattern
203 ):
204 raise ValueError(
205 f"The install-pattern in dh compat-rule at {idx} must be normalized as"
206 f' "{_normalize_path(install_pattern, with_prefix=False)}".'
207 )
210class DebputyPluginInitializerProvider(DebputyPluginInitializer):
211 __slots__ = (
212 "_plugin_metadata",
213 "_feature_set",
214 "_plugin_detector_ids",
215 "_substitution",
216 "_unloaders",
217 "_is_doc_cache_resolved",
218 "_doc_cache",
219 "_registered_manifest_types",
220 "_load_started",
221 )
223 def __init__(
224 self,
225 plugin_metadata: DebputyPluginMetadata,
226 feature_set: PluginProvidedFeatureSet,
227 substitution: Substitution,
228 ) -> None:
229 self._plugin_metadata: DebputyPluginMetadata = plugin_metadata
230 self._feature_set = feature_set
231 self._plugin_detector_ids: set[str] = set()
232 self._substitution = substitution
233 self._unloaders: list[Callable[[], None]] = []
234 self._is_doc_cache_resolved: bool = False
235 self._doc_cache: DebputyParsedDoc | None = None
236 self._registered_manifest_types: dict[type[Any], DebputyPluginMetadata] = {}
237 self._load_started = False
239 @property
240 def plugin_metadata(self) -> DebputyPluginMetadata:
241 return self._plugin_metadata
243 def unload_plugin(self) -> None:
244 if self._load_started:
245 for unloader in self._unloaders:
246 unloader()
247 del self._feature_set.plugin_data[self._plugin_name]
249 def load_plugin(self) -> None:
250 metadata = self._plugin_metadata
251 if metadata.plugin_name in self._feature_set.plugin_data: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 raise PluginConflictError(
253 f'The plugin "{metadata.plugin_name}" has already been loaded!?',
254 metadata,
255 metadata,
256 )
257 assert (
258 metadata.api_compat_version == 1
259 ), f"Unsupported plugin API compat version {metadata.api_compat_version}"
260 self._feature_set.plugin_data[metadata.plugin_name] = metadata
261 self._load_started = True
262 assert not metadata.is_initialized
263 try:
264 metadata.initialize_plugin(self)
265 except Exception as e:
266 initializer = metadata.plugin_initializer
267 if ( 267 ↛ 272line 267 didn't jump to line 272 because the condition on line 267 was never true
268 isinstance(e, TypeError)
269 and initializer is not None
270 and not callable(initializer)
271 ):
272 raise PluginMetadataError(
273 f"The specified entry point for plugin {metadata.plugin_name} does not appear to be a"
274 f" callable (callable returns False). The specified entry point identifies"
275 f' itself as "{initializer.__qualname__}".'
276 ) from e
277 if isinstance(e, PluginBaseError): 277 ↛ 279line 277 didn't jump to line 279 because the condition on line 277 was always true
278 raise
279 raise PluginInitializationError(
280 f"Exception while attempting to load plugin {metadata.plugin_name}"
281 ) from e
283 def _resolve_docs(self) -> DebputyParsedDoc | None:
284 doc_cache = self._doc_cache
285 if doc_cache is not None:
286 return doc_cache
288 plugin_doc_path = self._plugin_metadata.plugin_doc_path
289 if plugin_doc_path is None or self._is_doc_cache_resolved:
290 self._is_doc_cache_resolved = True
291 return None
292 try:
293 with plugin_doc_path.open("r", encoding="utf-8") as fd:
294 raw = MANIFEST_YAML.load(fd)
295 except FileNotFoundError:
296 _debug_log(
297 f"No documentation file found for {self._plugin_name}. Expected it at {plugin_doc_path}"
298 )
299 self._is_doc_cache_resolved = True
300 return None
301 attr_path = AttributePath.root_path(plugin_doc_path)
302 try:
303 ref = DEBPUTY_DOC_REFERENCE_DATA_PARSER.parse_input(raw, attr_path)
304 except ManifestParseException as e:
305 raise ValueError(
306 f"Could not parse documentation in {plugin_doc_path}: {e.message}"
307 ) from e
308 try:
309 res = DebputyParsedDoc.from_ref_data(ref)
310 except ValueError as e:
311 raise ValueError(
312 f"Could not parse documentation in {plugin_doc_path}: {e.args[0]}"
313 ) from e
315 self._doc_cache = res
316 self._is_doc_cache_resolved = True
317 return res
319 def _pluggable_manifest_docs_for(
320 self,
321 rule_type: TTP | str,
322 rule_name: str | list[str],
323 *,
324 inline_reference_documentation: ParserDocumentation | None = None,
325 ) -> ParserDocumentation | None:
326 ref_data = self._resolve_docs()
327 if ref_data is not None:
328 primary_rule_name = (
329 rule_name if isinstance(rule_name, str) else rule_name[0]
330 )
331 rule_ref = f"{parser_type_name(rule_type)}::{primary_rule_name}"
332 resolved_docs = ref_data.pluggable_manifest_rules.get(rule_ref)
333 if resolved_docs is not None:
334 if inline_reference_documentation is not None: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 raise ValueError(
336 f"Conflicting docs for {rule_ref}: Was provided one in the API call and one via"
337 f" {self._plugin_metadata.plugin_doc_path}. Please remove one of the two, so"
338 f" there is only one doc reference"
339 )
340 return resolved_docs
341 return inline_reference_documentation
343 def packager_provided_file(
344 self,
345 stem: str,
346 installed_path: str,
347 *,
348 default_mode: int = 0o0644,
349 default_priority: int | None = None,
350 allow_name_segment: bool = True,
351 allow_architecture_segment: bool = False,
352 post_formatting_rewrite: Callable[[str], str] | None = None,
353 packageless_is_fallback_for_all_packages: bool = False,
354 package_types: PackageTypeSelector = PackageTypeSelector.ALL,
355 reservation_only: bool = False,
356 format_callback: None | (
357 Callable[[str, PPFFormatParam, VirtualPath], str]
358 ) = None,
359 reference_documentation: None | (
360 PackagerProvidedFileReferenceDocumentation
361 ) = None,
362 ) -> None:
363 packager_provided_files = self._feature_set.packager_provided_files
364 existing = packager_provided_files.get(stem)
366 if format_callback is not None and self._plugin_name != "debputy": 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 raise ValueError(
368 "Sorry; Using format_callback is a debputy-internal"
369 f" API. Triggered by plugin {self._plugin_name}"
370 )
372 if installed_path.endswith("/"): 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 raise ValueError(
374 f'The installed_path ends with "/" indicating it is a directory, but it must be a file.'
375 f" Triggered by plugin {self._plugin_name}."
376 )
378 installed_path = _normalize_path(installed_path)
380 has_name_var = "{name}" in installed_path
382 if installed_path.startswith("./DEBIAN") or reservation_only:
383 # Special-case, used for control files.
384 if self._plugin_name != "debputy": 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true
385 raise ValueError(
386 "Sorry; Using DEBIAN as install path or/and reservation_only is a debputy-internal"
387 f" API. Triggered by plugin {self._plugin_name}"
388 )
389 elif not has_name_var and "{owning_package}" not in installed_path: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 raise ValueError(
391 'The installed_path must contain a "{name}" (preferred) or a "{owning_package}"'
392 " substitution (or have installed_path end with a slash). Otherwise, the installed"
393 f" path would caused file-conflicts. Triggered by plugin {self._plugin_name}"
394 )
396 if allow_name_segment and not has_name_var: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 raise ValueError(
398 'When allow_name_segment is True, the installed_path must have a "{name}" substitution'
399 " variable. Otherwise, the name segment will not work properly. Triggered by"
400 f" plugin {self._plugin_name}"
401 )
403 if ( 403 ↛ 408line 403 didn't jump to line 408 because the condition on line 403 was never true
404 default_priority is not None
405 and "{priority}" not in installed_path
406 and "{priority:02}" not in installed_path
407 ):
408 raise ValueError(
409 'When default_priority is not None, the installed_path should have a "{priority}"'
410 ' or a "{priority:02}" substitution variable. Otherwise, the priority would be lost.'
411 f" Triggered by plugin {self._plugin_name}"
412 )
414 if existing is not None:
415 if existing.debputy_plugin_metadata.plugin_name != self._plugin_name: 415 ↛ 422line 415 didn't jump to line 422 because the condition on line 415 was always true
416 message = (
417 f'The stem "{stem}" is registered twice for packager provided files.'
418 f" Once by {existing.debputy_plugin_metadata.plugin_name} and once"
419 f" by {self._plugin_name}"
420 )
421 else:
422 message = (
423 f"Bug in the plugin {self._plugin_name}: It tried to register the"
424 f' stem "{stem}" twice for packager provided files.'
425 )
426 raise PluginConflictError(
427 message, existing.debputy_plugin_metadata, self._plugin_metadata
428 )
429 packager_provided_files[stem] = PackagerProvidedFileClassSpec(
430 self._plugin_metadata,
431 stem,
432 installed_path,
433 default_mode=default_mode,
434 default_priority=default_priority,
435 allow_name_segment=allow_name_segment,
436 allow_architecture_segment=allow_architecture_segment,
437 post_formatting_rewrite=post_formatting_rewrite,
438 packageless_is_fallback_for_all_packages=packageless_is_fallback_for_all_packages,
439 package_types=package_types,
440 reservation_only=reservation_only,
441 formatting_callback=format_callback,
442 reference_documentation=reference_documentation,
443 )
445 def _unload() -> None:
446 del packager_provided_files[stem]
448 self._unloaders.append(_unload)
450 def metadata_or_maintscript_detector(
451 self,
452 auto_detector_id: str,
453 auto_detector: MetadataAutoDetector,
454 *,
455 package_types: PackageTypeSelector = PackageTypeSelector.DEB,
456 ) -> None:
457 if auto_detector_id in self._plugin_detector_ids: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true
458 raise ValueError(
459 f"The plugin {self._plugin_name} tried to register"
460 f' "{auto_detector_id}" twice'
461 )
462 self._plugin_detector_ids.add(auto_detector_id)
463 all_detectors = self._feature_set.metadata_maintscript_detectors
464 if self._plugin_name not in all_detectors:
465 all_detectors[self._plugin_name] = []
466 all_detectors[self._plugin_name].append(
467 MetadataOrMaintscriptDetector(
468 detector_id=auto_detector_id,
469 detector=wrap_plugin_code(self._plugin_name, auto_detector),
470 plugin_metadata=self._plugin_metadata,
471 applies_to_package_types=package_types,
472 enabled=True,
473 )
474 )
476 def _unload() -> None:
477 if self._plugin_name in all_detectors:
478 del all_detectors[self._plugin_name]
480 self._unloaders.append(_unload)
482 def document_builtin_variable(
483 self,
484 variable_name: str,
485 variable_reference_documentation: str,
486 *,
487 is_context_specific: bool = False,
488 is_for_special_case: bool = False,
489 ) -> None:
490 manifest_variables = self._feature_set.manifest_variables
491 self._restricted_api()
492 state = self._substitution.variable_state(variable_name)
493 if state == VariableNameState.UNDEFINED: 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true
494 raise ValueError(
495 f"The plugin {self._plugin_name} attempted to document built-in {variable_name},"
496 f" but it is not known to be a variable"
497 )
499 assert variable_name not in manifest_variables
501 manifest_variables[variable_name] = PluginProvidedManifestVariable(
502 self._plugin_metadata,
503 variable_name,
504 None,
505 is_context_specific_variable=is_context_specific,
506 variable_reference_documentation=variable_reference_documentation,
507 is_documentation_placeholder=True,
508 is_for_special_case=is_for_special_case,
509 )
511 def _unload() -> None:
512 del manifest_variables[variable_name]
514 self._unloaders.append(_unload)
516 def manifest_variable_provider(
517 self,
518 provider: Callable[[VariableContext], Mapping[str, str]],
519 variables: Sequence[str] | Mapping[str, str | None],
520 ) -> None:
521 self._restricted_api()
522 cached_provider = functools.lru_cache(None)(provider)
523 permitted_variables = frozenset(variables)
524 variables_iter: Iterable[tuple[str, str | None]]
525 if not isinstance(variables, Mapping): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true
526 variables_iter = zip(variables, itertools.repeat(None))
527 else:
528 variables_iter = variables.items()
530 checked_vars = False
531 manifest_variables = self._feature_set.manifest_variables
532 plugin_name = self._plugin_name
534 def _value_resolver_generator(
535 variable_name: str,
536 ) -> Callable[[VariableContext], str]:
537 def _value_resolver(variable_context: VariableContext) -> str:
538 res = cached_provider(variable_context)
539 nonlocal checked_vars
540 if not checked_vars: 540 ↛ 551line 540 didn't jump to line 551 because the condition on line 540 was always true
541 if permitted_variables != res.keys(): 541 ↛ 542line 541 didn't jump to line 542 because the condition on line 541 was never true
542 expected = ", ".join(sorted(permitted_variables))
543 actual = ", ".join(sorted(res))
544 raise PluginAPIViolationError(
545 f"The plugin {plugin_name} claimed to provide"
546 f" the following variables {expected},"
547 f" but when resolving the variables, the plugin provided"
548 f" {actual}. These two lists should have been the same."
549 )
550 checked_vars = False
551 return res[variable_name]
553 return _value_resolver
555 for varname, vardoc in variables_iter:
556 self._check_variable_name(varname)
557 manifest_variables[varname] = PluginProvidedManifestVariable(
558 self._plugin_metadata,
559 varname,
560 _value_resolver_generator(varname),
561 is_context_specific_variable=False,
562 variable_reference_documentation=vardoc,
563 )
565 def _unload() -> None:
566 raise PluginInitializationError(
567 "Cannot unload manifest_variable_provider (not implemented)"
568 )
570 self._unloaders.append(_unload)
572 def _check_variable_name(self, variable_name: str) -> None:
573 manifest_variables = self._feature_set.manifest_variables
574 existing = manifest_variables.get(variable_name)
576 if existing is not None:
577 if existing.plugin_metadata.plugin_name == self._plugin_name: 577 ↛ 583line 577 didn't jump to line 583 because the condition on line 577 was always true
578 message = (
579 f"Bug in the plugin {self._plugin_name}: It tried to register the"
580 f' manifest variable "{variable_name}" twice.'
581 )
582 else:
583 message = (
584 f"The plugins {existing.plugin_metadata.plugin_name} and {self._plugin_name}"
585 f" both tried to provide the manifest variable {variable_name}"
586 )
587 raise PluginConflictError(
588 message, existing.plugin_metadata, self._plugin_metadata
589 )
590 if not SUBST_VAR_RE.match("{{" + variable_name + "}}"):
591 raise ValueError(
592 f"The plugin {self._plugin_name} attempted to declare {variable_name},"
593 f" which is not a valid variable name"
594 )
596 namespace = ""
597 variable_basename = variable_name
598 if ":" in variable_name:
599 namespace, variable_basename = variable_name.rsplit(":", 1)
600 assert namespace != ""
601 assert variable_name != ""
603 if namespace != "" and namespace not in ("token", "path"):
604 raise ValueError(
605 f"The plugin {self._plugin_name} attempted to declare {variable_name},"
606 f" which is in the reserved namespace {namespace}"
607 )
609 variable_name_upper = variable_name.upper()
610 if (
611 variable_name_upper.startswith(("DEB_", "DPKG_", "DEBPUTY"))
612 or variable_basename.startswith("_")
613 or variable_basename.upper().startswith("DEBPUTY")
614 ) and self._plugin_name != "debputy":
615 raise ValueError(
616 f"The plugin {self._plugin_name} attempted to declare {variable_name},"
617 f" which is a variable name reserved by debputy"
618 )
620 state = self._substitution.variable_state(variable_name)
621 if state != VariableNameState.UNDEFINED and self._plugin_name != "debputy":
622 raise ValueError(
623 f"The plugin {self._plugin_name} attempted to declare {variable_name},"
624 f" which would shadow a built-in variable"
625 )
627 def package_processor(
628 self,
629 processor_id: str,
630 processor: PackageProcessor,
631 *,
632 depends_on_processor: Iterable[str] = (),
633 package_types: PackageTypeSelector = PackageTypeSelector.DEB,
634 ) -> None:
635 self._restricted_api(allowed_plugins={"lua", "debputy-self-hosting"})
636 package_processors = self._feature_set.all_package_processors
637 dependencies = set()
638 processor_key = (self._plugin_name, processor_id)
640 if processor_key in package_processors: 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true
641 raise PluginConflictError(
642 f"The plugin {self._plugin_name} already registered a processor with id {processor_id}",
643 self._plugin_metadata,
644 self._plugin_metadata,
645 )
647 for depends_ref in depends_on_processor:
648 if isinstance(depends_ref, str): 648 ↛ 662line 648 didn't jump to line 662 because the condition on line 648 was always true
649 if (self._plugin_name, depends_ref) in package_processors: 649 ↛ 651line 649 didn't jump to line 651 because the condition on line 649 was always true
650 depends_key = (self._plugin_name, depends_ref)
651 elif ("debputy", depends_ref) in package_processors:
652 depends_key = ("debputy", depends_ref)
653 else:
654 raise ValueError(
655 f'Could not resolve dependency "{depends_ref}" for'
656 f' "{processor_id}". It was not provided by the plugin itself'
657 f" ({self._plugin_name}) nor debputy."
658 )
659 else:
660 # TODO: Add proper dependencies first, at which point we should probably resolve "name"
661 # via the direct dependencies.
662 assert False
664 existing_processor = package_processors.get(depends_key)
665 if existing_processor is None: 665 ↛ 668line 665 didn't jump to line 668 because the condition on line 665 was never true
666 # We currently require the processor to be declared already. If this ever changes,
667 # PluginProvidedFeatureSet.package_processors_in_order will need an update
668 dplugin_name, dprocessor_name = depends_key
669 available_processors = ", ".join(
670 n for p, n in package_processors.keys() if p == dplugin_name
671 )
672 raise ValueError(
673 f"The plugin {dplugin_name} does not provide a processor called"
674 f" {dprocessor_name}. Available processors for that plugin are:"
675 f" {available_processors}"
676 )
677 dependencies.add(depends_key)
679 package_processors[processor_key] = PluginProvidedPackageProcessor(
680 processor_id,
681 package_types,
682 wrap_plugin_code(self._plugin_name, processor),
683 frozenset(dependencies),
684 self._plugin_metadata,
685 )
687 def _unload() -> None:
688 del package_processors[processor_key]
690 self._unloaders.append(_unload)
692 def automatic_discard_rule(
693 self,
694 name: str,
695 should_discard: Callable[[VirtualPath], bool],
696 *,
697 rule_reference_documentation: str | None = None,
698 examples: (
699 AutomaticDiscardRuleExample | Sequence[AutomaticDiscardRuleExample]
700 ) = (),
701 ) -> None:
702 """Register an automatic discard rule
704 An automatic discard rule is basically applied to *every* path about to be installed in to any package.
705 If any discard rule concludes that a path should not be installed, then the path is not installed.
706 In the case where the discard path is a:
708 * directory: Then the entire directory is excluded along with anything beneath it.
709 * symlink: Then the symlink itself (but not its target) is excluded.
710 * hardlink: Then the current hardlink will not be installed, but other instances of it will be.
712 Note: Discarded files are *never* deleted by `debputy`. They just make `debputy` skip the file.
714 Automatic discard rules should be written with the assumption that directories will be tested
715 before their content *when it is relevant* for the discard rule to examine whether the directory
716 can be excluded.
718 The packager can via the manifest overrule automatic discard rules by explicitly listing the path
719 without any globs. As example:
721 installations:
722 - install:
723 sources:
724 - usr/lib/libfoo.la # <-- This path is always installed
725 # (Discard rules are never asked in this case)
726 #
727 - usr/lib/*.so* # <-- Discard rules applies to any path beneath usr/lib and can exclude matches
728 # Though, they will not examine `libfoo.la` as it has already been installed
729 #
730 # Note: usr/lib itself is never tested in this case (it is assumed to be
731 # explicitly requested). But any subdir of usr/lib will be examined.
733 When an automatic discard rule is evaluated, it can see the source path currently being considered
734 for installation. While it can look at "surrounding" context (like parent directory), it will not
735 know whether those paths are to be installed or will be installed.
737 :param name: A user visible name discard rule. It can be used on the command line, so avoid shell
738 metacharacters and spaces.
739 :param should_discard: A callable that is the implementation of the automatic discard rule. It will receive
740 a VirtualPath representing the *source* path about to be installed. If callable returns `True`, then the
741 path is discarded. If it returns `False`, the path is not discarded (by this rule at least).
742 A source path will either be from the root of the source tree or the root of a search directory such as
743 `debian/tmp`. Where the path will be installed is not available at the time the discard rule is
744 evaluated.
745 :param rule_reference_documentation: Optionally, the reference documentation to be shown when a user
746 looks up this automatic discard rule.
747 :param examples: Provide examples for the rule. Use the automatic_discard_rule_example function to
748 generate the examples.
750 """
751 self._restricted_api()
752 auto_discard_rules = self._feature_set.auto_discard_rules
753 existing = auto_discard_rules.get(name)
754 if existing is not None: 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 if existing.plugin_metadata.plugin_name == self._plugin_name:
756 message = (
757 f"Bug in the plugin {self._plugin_name}: It tried to register the"
758 f' automatic discard rule "{name}" twice.'
759 )
760 else:
761 message = (
762 f"The plugins {existing.plugin_metadata.plugin_name} and {self._plugin_name}"
763 f" both tried to provide the automatic discard rule {name}"
764 )
765 raise PluginConflictError(
766 message, existing.plugin_metadata, self._plugin_metadata
767 )
768 examples = (
769 (examples,)
770 if isinstance(examples, AutomaticDiscardRuleExample)
771 else tuple(examples)
772 )
773 auto_discard_rules[name] = PluginProvidedDiscardRule(
774 name,
775 self._plugin_metadata,
776 should_discard,
777 rule_reference_documentation,
778 examples,
779 )
781 def _unload() -> None:
782 del auto_discard_rules[name]
784 self._unloaders.append(_unload)
786 def service_provider(
787 self,
788 service_manager: str,
789 detector: ServiceDetector,
790 integrator: ServiceIntegrator,
791 ) -> None:
792 self._restricted_api()
793 service_managers = self._feature_set.service_managers
794 existing = service_managers.get(service_manager)
795 if existing is not None: 795 ↛ 796line 795 didn't jump to line 796 because the condition on line 795 was never true
796 if existing.plugin_metadata.plugin_name == self._plugin_name:
797 message = (
798 f"Bug in the plugin {self._plugin_name}: It tried to register the"
799 f' service manager "{service_manager}" twice.'
800 )
801 else:
802 message = (
803 f"The plugins {existing.plugin_metadata.plugin_name} and {self._plugin_name}"
804 f' both tried to provide the service manager "{service_manager}"'
805 )
806 raise PluginConflictError(
807 message, existing.plugin_metadata, self._plugin_metadata
808 )
809 service_managers[service_manager] = ServiceManagerDetails(
810 service_manager,
811 wrap_plugin_code(self._plugin_name, detector),
812 wrap_plugin_code(self._plugin_name, integrator),
813 self._plugin_metadata,
814 )
816 def _unload() -> None:
817 del service_managers[service_manager]
819 self._unloaders.append(_unload)
821 def manifest_variable(
822 self,
823 variable_name: str,
824 value: str,
825 *,
826 variable_reference_documentation: str | None = None,
827 ) -> None:
828 self._check_variable_name(variable_name)
829 manifest_variables = self._feature_set.manifest_variables
830 try:
831 resolved_value = self._substitution.substitute(
832 value, "Plugin initialization"
833 )
834 depends_on_variable = resolved_value != value
835 except DebputySubstitutionError:
836 depends_on_variable = True
837 if depends_on_variable:
838 raise ValueError(
839 f"The plugin {self._plugin_name} attempted to declare {variable_name} with value {value!r}."
840 f" This value depends on another variable, which is not supported. This restriction may be"
841 f" lifted in the future."
842 )
844 manifest_variables[variable_name] = PluginProvidedManifestVariable(
845 self._plugin_metadata,
846 variable_name,
847 value,
848 is_context_specific_variable=False,
849 variable_reference_documentation=variable_reference_documentation,
850 )
852 def _unload() -> None:
853 # We need to check it was never resolved
854 raise PluginInitializationError(
855 "Cannot unload manifest_variable (not implemented)"
856 )
858 self._unloaders.append(_unload)
860 @property
861 def _plugin_name(self) -> str:
862 return self._plugin_metadata.plugin_name
864 def provide_manifest_keyword(
865 self,
866 rule_type: TTP,
867 rule_name: str | list[str],
868 handler: DIPKWHandler,
869 *,
870 inline_reference_documentation: ParserDocumentation | None = None,
871 ) -> None:
872 self._restricted_api()
873 parser_generator = self._feature_set.manifest_parser_generator
874 if rule_type not in parser_generator.dispatchable_table_parsers: 874 ↛ 875line 874 didn't jump to line 875 because the condition on line 874 was never true
875 types = ", ".join(
876 sorted(x.__name__ for x in parser_generator.dispatchable_table_parsers)
877 )
878 raise ValueError(
879 f"The rule_type was not a supported type. It must be one of {types}"
880 )
882 inline_reference_documentation = self._pluggable_manifest_docs_for(
883 rule_type,
884 rule_name,
885 inline_reference_documentation=inline_reference_documentation,
886 )
888 dispatching_parser = parser_generator.dispatchable_table_parsers[rule_type]
889 dispatching_parser.register_keyword(
890 rule_name,
891 wrap_plugin_code(self._plugin_name, handler),
892 self._plugin_metadata,
893 inline_reference_documentation=inline_reference_documentation,
894 )
896 def _unload() -> None:
897 raise PluginInitializationError(
898 "Cannot unload provide_manifest_keyword (not implemented)"
899 )
901 self._unloaders.append(_unload)
903 def pluggable_object_parser(
904 self,
905 rule_type: str,
906 rule_name: str,
907 *,
908 object_parser_key: str | None = None,
909 on_end_parse_step: None | (
910 Callable[
911 [str, Mapping[str, Any] | None, AttributePath, ParserContextData],
912 None,
913 ]
914 ) = None,
915 nested_in_package_context: bool = False,
916 ) -> None:
917 self._restricted_api()
918 if object_parser_key is None: 918 ↛ 919line 918 didn't jump to line 919 because the condition on line 918 was never true
919 object_parser_key = rule_name
921 parser_generator = self._feature_set.manifest_parser_generator
922 dispatchable_object_parsers = parser_generator.dispatchable_object_parsers
923 if rule_type not in dispatchable_object_parsers: 923 ↛ 924line 923 didn't jump to line 924 because the condition on line 923 was never true
924 types = ", ".join(sorted(dispatchable_object_parsers))
925 raise ValueError(
926 f"The rule_type was not a supported type. It must be one of {types}"
927 )
928 if object_parser_key not in dispatchable_object_parsers: 928 ↛ 929line 928 didn't jump to line 929 because the condition on line 928 was never true
929 types = ", ".join(sorted(dispatchable_object_parsers))
930 raise ValueError(
931 f"The object_parser_key was not a supported type. It must be one of {types}"
932 )
933 parent_dispatcher = dispatchable_object_parsers[rule_type]
934 child_dispatcher = dispatchable_object_parsers[object_parser_key]
936 if on_end_parse_step is not None: 936 ↛ 939line 936 didn't jump to line 939 because the condition on line 936 was always true
937 on_end_parse_step = wrap_plugin_code(self._plugin_name, on_end_parse_step)
939 parent_dispatcher.register_child_parser(
940 rule_name,
941 child_dispatcher,
942 self._plugin_metadata,
943 on_end_parse_step=on_end_parse_step,
944 nested_in_package_context=nested_in_package_context,
945 )
947 def _unload() -> None:
948 raise PluginInitializationError(
949 "Cannot unload pluggable_object_parser (not implemented)"
950 )
952 self._unloaders.append(_unload)
954 def pluggable_manifest_rule(
955 self,
956 rule_type: TTP | str,
957 rule_name: str | Sequence[str],
958 parsed_format: type[PF],
959 handler: DIPHandler,
960 *,
961 as_keyword_handler: DIPKWHandler | None = None,
962 source_format: SF | None = None,
963 inline_reference_documentation: ParserDocumentation | None = None,
964 expected_debputy_integration_mode: None | (
965 Container[DebputyIntegrationMode]
966 ) = None,
967 apply_standard_attribute_documentation: bool = False,
968 register_value: bool = True,
969 ) -> None:
970 # When changing this, consider which types will be unrestricted
971 self._restricted_api()
972 if apply_standard_attribute_documentation and sys.version_info < (3, 12): 972 ↛ 973line 972 didn't jump to line 973 because the condition on line 972 was never true
973 _error(
974 f"The plugin {self._plugin_metadata.plugin_name} requires python 3.12 due to"
975 f" its use of apply_standard_attribute_documentation"
976 )
977 feature_set = self._feature_set
978 parser_generator = feature_set.manifest_parser_generator
979 if isinstance(rule_type, str):
980 if rule_type not in parser_generator.dispatchable_object_parsers: 980 ↛ 981line 980 didn't jump to line 981 because the condition on line 980 was never true
981 types = ", ".join(sorted(parser_generator.dispatchable_object_parsers))
982 raise ValueError(
983 f"The rule_type was not a supported type. It must be one of {types}"
984 )
985 dispatching_parser = parser_generator.dispatchable_object_parsers[rule_type]
986 signature = inspect.signature(handler)
987 if ( 987 ↛ 991line 987 didn't jump to line 991 because the condition on line 987 was never true
988 signature.return_annotation is signature.empty
989 or signature.return_annotation == NoneType
990 ):
991 raise ValueError(
992 "The handler must have a return type (that is not None)"
993 )
994 register_as_type = signature.return_annotation
995 else:
996 # Dispatchable types cannot be resolved
997 register_as_type = None
998 if rule_type not in parser_generator.dispatchable_table_parsers: 998 ↛ 999line 998 didn't jump to line 999 because the condition on line 998 was never true
999 types = ", ".join(
1000 sorted(
1001 x.__name__ for x in parser_generator.dispatchable_table_parsers
1002 )
1003 )
1004 raise ValueError(
1005 f"The rule_type was not a supported type. It must be one of {types}"
1006 )
1007 dispatching_parser = parser_generator.dispatchable_table_parsers[rule_type]
1009 if register_as_type is not None and not register_value:
1010 register_as_type = None
1012 if register_as_type is not None:
1013 existing_registration = self._registered_manifest_types.get(
1014 register_as_type
1015 )
1016 if existing_registration is not None: 1016 ↛ 1017line 1016 didn't jump to line 1017 because the condition on line 1016 was never true
1017 raise ValueError(
1018 f"Cannot register rule {rule_name!r} for plugin {self._plugin_name}. The plugin {existing_registration.plugin_name} already registered a manifest rule with type {register_as_type!r}"
1019 )
1020 self._registered_manifest_types[register_as_type] = self._plugin_metadata
1022 inline_reference_documentation = self._pluggable_manifest_docs_for(
1023 rule_type,
1024 rule_name,
1025 inline_reference_documentation=inline_reference_documentation,
1026 )
1028 if apply_standard_attribute_documentation: 1028 ↛ 1029line 1028 didn't jump to line 1029 because the condition on line 1028 was never true
1029 docs = _STD_ATTR_DOCS
1030 else:
1031 docs = None
1033 parser = feature_set.manifest_parser_generator.generate_parser(
1034 parsed_format,
1035 source_content=source_format,
1036 inline_reference_documentation=inline_reference_documentation,
1037 expected_debputy_integration_mode=expected_debputy_integration_mode,
1038 automatic_docs=docs,
1039 allow_none_value=as_keyword_handler is not None,
1040 )
1042 def _registering_handler(
1043 name: str,
1044 parsed_data: PF | None,
1045 attribute_path: AttributePath,
1046 parser_context: ParserContextData,
1047 ) -> TP:
1048 if parsed_data is not None: 1048 ↛ 1051line 1048 didn't jump to line 1051 because the condition on line 1048 was always true
1049 value = handler(name, parsed_data, attribute_path, parser_context)
1050 else:
1051 if as_keyword_handler is None:
1052 raise AssertionError(
1053 f"{name} was used and allowed as a keyword-only, but there was no handler for it"
1054 )
1055 value = as_keyword_handler(name, attribute_path, parser_context)
1056 if register_as_type is not None:
1057 register_manifest_type_value_in_context(register_as_type, value)
1058 return value
1060 dispatching_parser.register_parser(
1061 rule_name,
1062 parser,
1063 wrap_plugin_code(self._plugin_name, _registering_handler),
1064 self._plugin_metadata,
1065 )
1067 def _unload() -> None:
1068 raise PluginInitializationError(
1069 "Cannot unload pluggable_manifest_rule (not implemented)"
1070 )
1072 self._unloaders.append(_unload)
1074 def register_build_system(
1075 self,
1076 build_system_definition: type[BSPF],
1077 ) -> None:
1078 self._restricted_api()
1079 if not is_typeddict(build_system_definition): 1079 ↛ 1080line 1079 didn't jump to line 1080 because the condition on line 1079 was never true
1080 raise PluginInitializationError(
1081 f"Expected build_system_definition to be a subclass of {BuildRuleParsedFormat.__name__},"
1082 f" but got {build_system_definition.__name__} instead"
1083 )
1084 metadata = getattr(
1085 build_system_definition,
1086 _DEBPUTY_DISPATCH_METADATA_ATTR_NAME,
1087 None,
1088 )
1089 if not isinstance(metadata, BuildSystemManifestRuleMetadata): 1089 ↛ 1090line 1089 didn't jump to line 1090 because the condition on line 1089 was never true
1090 raise PluginIncorrectRegistrationError(
1091 f"The {build_system_definition.__qualname__} type should have been annotated with"
1092 f" @{debputy_build_system.__name__}."
1093 )
1094 assert len(metadata.manifest_keywords) == 1
1095 build_system_impl = metadata.build_system_impl
1096 assert build_system_impl is not None
1097 manifest_keyword = next(iter(metadata.manifest_keywords))
1098 self.pluggable_manifest_rule(
1099 metadata.dispatched_type,
1100 metadata.manifest_keywords,
1101 build_system_definition,
1102 # pluggable_manifest_rule does the wrapping
1103 metadata.unwrapped_constructor,
1104 source_format=metadata.source_format,
1105 inline_reference_documentation=metadata.online_reference_documentation,
1106 expected_debputy_integration_mode=only_integrations(INTEGRATION_MODE_FULL),
1107 )
1108 self._auto_detectable_build_system(
1109 manifest_keyword,
1110 build_system_impl,
1111 constructor=wrap_plugin_code(
1112 self._plugin_name,
1113 build_system_impl,
1114 ),
1115 shadowing_build_systems_when_active=metadata.auto_detection_shadow_build_systems,
1116 )
1118 def _auto_detectable_build_system(
1119 self,
1120 manifest_keyword: str,
1121 rule_type: type[BSR],
1122 *,
1123 shadowing_build_systems_when_active: frozenset[str] = frozenset(),
1124 constructor: None | (
1125 Callable[[BuildRuleParsedFormat, AttributePath, "HighLevelManifest"], BSR]
1126 ) = None,
1127 ) -> None:
1128 self._restricted_api()
1129 feature_set = self._feature_set
1130 existing = feature_set.auto_detectable_build_systems.get(rule_type)
1131 if existing is not None: 1131 ↛ 1132line 1131 didn't jump to line 1132 because the condition on line 1131 was never true
1132 bs_name = rule_type.__class__.__name__
1133 if existing.plugin_metadata.plugin_name == self._plugin_name:
1134 message = (
1135 f"Bug in the plugin {self._plugin_name}: It tried to register the"
1136 f' auto-detection of the build system "{bs_name}" twice.'
1137 )
1138 else:
1139 message = (
1140 f"The plugins {existing.plugin_metadata.plugin_name} and {self._plugin_name}"
1141 f' both tried to provide auto-detection of the build system "{bs_name}"'
1142 )
1143 raise PluginConflictError(
1144 message, existing.plugin_metadata, self._plugin_metadata
1145 )
1147 if constructor is None: 1147 ↛ 1149line 1147 didn't jump to line 1149 because the condition on line 1147 was never true
1149 def impl(
1150 attributes: BuildRuleParsedFormat,
1151 attribute_path: AttributePath,
1152 manifest: "HighLevelManifest",
1153 ) -> BSR:
1154 return rule_type(attributes, attribute_path, manifest)
1156 else:
1157 impl = constructor
1159 feature_set.auto_detectable_build_systems[rule_type] = (
1160 PluginProvidedBuildSystemAutoDetection(
1161 manifest_keyword,
1162 rule_type,
1163 wrap_plugin_code(self._plugin_name, rule_type.auto_detect_build_system),
1164 impl,
1165 shadowing_build_systems_when_active,
1166 self._plugin_metadata,
1167 )
1168 )
1170 def _unload() -> None:
1171 try:
1172 del feature_set.auto_detectable_build_systems[rule_type]
1173 except KeyError:
1174 pass
1176 self._unloaders.append(_unload)
1178 def known_packaging_files(
1179 self,
1180 packaging_file_details: KnownPackagingFileInfo,
1181 ) -> None:
1182 known_packaging_files = self._feature_set.known_packaging_files
1183 detection_method = packaging_file_details.get(
1184 "detection_method", cast("Literal['path']", "path")
1185 )
1186 path = packaging_file_details.get("path")
1187 dhpkgfile = packaging_file_details.get("pkgfile")
1189 packaging_file_details = packaging_file_details.copy()
1191 if detection_method == "path": 1191 ↛ 1207line 1191 didn't jump to line 1207 because the condition on line 1191 was always true
1192 if dhpkgfile is not None: 1192 ↛ 1193line 1192 didn't jump to line 1193 because the condition on line 1192 was never true
1193 raise ValueError(
1194 'The "pkgfile" attribute cannot be used when detection-method is "path" (or omitted)'
1195 )
1196 if path is None: 1196 ↛ 1197line 1196 didn't jump to line 1197 because the condition on line 1196 was never true
1197 raise ValueError(
1198 'The "path" attribute must be present when detection-method is "path" (or omitted)'
1199 )
1200 if path != _normalize_path(path, with_prefix=False): 1200 ↛ 1201line 1200 didn't jump to line 1201 because the condition on line 1200 was never true
1201 raise ValueError(
1202 f"The path for known packaging files must be normalized. Please replace"
1203 f' "{path}" with "{_normalize_path(path, with_prefix=False)}"'
1204 )
1205 detection_value = path
1206 else:
1207 assert detection_method == "dh.pkgfile"
1208 if path is not None:
1209 raise ValueError(
1210 'The "path" attribute cannot be used when detection-method is "dh.pkgfile"'
1211 )
1212 if dhpkgfile is None:
1213 raise ValueError(
1214 'The "pkgfile" attribute must be present when detection-method is "dh.pkgfile"'
1215 )
1216 if "/" in dhpkgfile:
1217 raise ValueError(
1218 'The "pkgfile" attribute ḿust be a name stem such as "install" (no "/" are allowed)'
1219 )
1220 detection_value = dhpkgfile
1221 key = f"{detection_method}::{detection_value}"
1222 existing = known_packaging_files.get(key)
1223 if existing is not None: 1223 ↛ 1224line 1223 didn't jump to line 1224 because the condition on line 1223 was never true
1224 if existing.plugin_metadata.plugin_name != self._plugin_name:
1225 message = (
1226 f'The key "{key}" is registered twice for known packaging files.'
1227 f" Once by {existing.plugin_metadata.plugin_name} and once by {self._plugin_name}"
1228 )
1229 else:
1230 message = (
1231 f"Bug in the plugin {self._plugin_name}: It tried to register the"
1232 f' key "{key}" twice for known packaging files.'
1233 )
1234 raise PluginConflictError(
1235 message, existing.plugin_metadata, self._plugin_metadata
1236 )
1237 _validate_known_packaging_file_dh_compat_rules(
1238 packaging_file_details.get("dh_compat_rules")
1239 )
1240 known_packaging_files[key] = PluginProvidedKnownPackagingFile(
1241 packaging_file_details,
1242 detection_method,
1243 detection_value,
1244 self._plugin_metadata,
1245 )
1247 def _unload() -> None:
1248 del known_packaging_files[key]
1250 self._unloaders.append(_unload)
1252 def register_mapped_type(
1253 self,
1254 type_mapping: TypeMapping,
1255 *,
1256 reference_documentation: TypeMappingDocumentation | None = None,
1257 ) -> None:
1258 self._restricted_api()
1259 target_type = type_mapping.target_type
1260 mapped_types = self._feature_set.mapped_types
1261 existing = mapped_types.get(target_type)
1262 if existing is not None: 1262 ↛ 1263line 1262 didn't jump to line 1263 because the condition on line 1262 was never true
1263 if existing.plugin_metadata.plugin_name != self._plugin_name:
1264 message = (
1265 f'The key "{target_type.__name__}" is registered twice for known packaging files.'
1266 f" Once by {existing.plugin_metadata.plugin_name} and once by {self._plugin_name}"
1267 )
1268 else:
1269 message = (
1270 f"Bug in the plugin {self._plugin_name}: It tried to register the"
1271 f' key "{target_type.__name__}" twice for known packaging files.'
1272 )
1273 raise PluginConflictError(
1274 message, existing.plugin_metadata, self._plugin_metadata
1275 )
1276 parser_generator = self._feature_set.manifest_parser_generator
1277 # TODO: Wrap the mapper in the plugin context
1278 mapped_types[target_type] = PluginProvidedTypeMapping(
1279 type_mapping, reference_documentation, self._plugin_metadata
1280 )
1281 parser_generator.register_mapped_type(type_mapping)
1283 def _restricted_api(
1284 self,
1285 *,
1286 allowed_plugins: set[str] | frozenset[str] = frozenset(),
1287 ) -> None:
1288 if self._plugin_name != "debputy" and self._plugin_name not in allowed_plugins: 1288 ↛ 1289line 1288 didn't jump to line 1289 because the condition on line 1288 was never true
1289 raise PluginAPIViolationError(
1290 f"Plugin {self._plugin_name} attempted to access a debputy-only API."
1291 " If you are the maintainer of this plugin and want access to this"
1292 " API, please file a feature request to make this public."
1293 " (The API is currently private as it is unstable.)"
1294 )
1297class MaintscriptAccessorProviderBase(MaintscriptAccessor, ABC):
1298 __slots__ = ()
1300 def _append_script(
1301 self,
1302 caller_name: str,
1303 maintscript: Maintscript,
1304 full_script: str,
1305 /,
1306 perform_substitution: bool = True,
1307 ) -> None:
1308 raise NotImplementedError
1310 @classmethod
1311 def _apply_condition_to_script(
1312 cls,
1313 condition: str,
1314 run_snippet: str,
1315 /,
1316 indent: bool | None = None,
1317 ) -> str:
1318 if indent is None:
1319 # We auto-determine this based on heredocs currently
1320 indent = "<<" not in run_snippet
1322 if indent:
1323 run_snippet = "".join(" " + x for x in run_snippet.splitlines(True))
1324 if not run_snippet.endswith("\n"):
1325 run_snippet += "\n"
1326 condition_line = f"if {condition}; then\n"
1327 end_line = "fi\n"
1328 return "".join((condition_line, run_snippet, end_line))
1330 def on_configure(
1331 self,
1332 run_snippet: str,
1333 /,
1334 indent: bool | None = None,
1335 perform_substitution: bool = True,
1336 skip_on_rollback: bool = False,
1337 ) -> None:
1338 condition = POSTINST_DEFAULT_CONDITION
1339 if skip_on_rollback: 1339 ↛ 1340line 1339 didn't jump to line 1340 because the condition on line 1339 was never true
1340 condition = '[ "$1" = "configure" ]'
1341 return self._append_script(
1342 "on_configure",
1343 "postinst",
1344 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1345 perform_substitution=perform_substitution,
1346 )
1348 def on_initial_install(
1349 self,
1350 run_snippet: str,
1351 /,
1352 indent: bool | None = None,
1353 perform_substitution: bool = True,
1354 ) -> None:
1355 condition = '[ "$1" = "configure" -a -z "$2" ]'
1356 return self._append_script(
1357 "on_initial_install",
1358 "postinst",
1359 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1360 perform_substitution=perform_substitution,
1361 )
1363 def on_upgrade(
1364 self,
1365 run_snippet: str,
1366 /,
1367 indent: bool | None = None,
1368 perform_substitution: bool = True,
1369 ) -> None:
1370 condition = '[ "$1" = "configure" -a -n "$2" ]'
1371 return self._append_script(
1372 "on_upgrade",
1373 "postinst",
1374 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1375 perform_substitution=perform_substitution,
1376 )
1378 def on_upgrade_from(
1379 self,
1380 version: str,
1381 run_snippet: str,
1382 /,
1383 indent: bool | None = None,
1384 perform_substitution: bool = True,
1385 ) -> None:
1386 condition = '[ "$1" = "configure" ] && dpkg --compare-versions le-nl "$2"'
1387 return self._append_script(
1388 "on_upgrade_from",
1389 "postinst",
1390 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1391 perform_substitution=perform_substitution,
1392 )
1394 def on_before_removal(
1395 self,
1396 run_snippet: str,
1397 /,
1398 indent: bool | None = None,
1399 perform_substitution: bool = True,
1400 ) -> None:
1401 condition = '[ "$1" = "remove" ]'
1402 return self._append_script(
1403 "on_before_removal",
1404 "prerm",
1405 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1406 perform_substitution=perform_substitution,
1407 )
1409 def on_removed(
1410 self,
1411 run_snippet: str,
1412 /,
1413 indent: bool | None = None,
1414 perform_substitution: bool = True,
1415 ) -> None:
1416 condition = '[ "$1" = "remove" ]'
1417 return self._append_script(
1418 "on_removed",
1419 "postrm",
1420 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1421 perform_substitution=perform_substitution,
1422 )
1424 def on_purge(
1425 self,
1426 run_snippet: str,
1427 /,
1428 indent: bool | None = None,
1429 perform_substitution: bool = True,
1430 ) -> None:
1431 condition = '[ "$1" = "purge" ]'
1432 return self._append_script(
1433 "on_purge",
1434 "postrm",
1435 self._apply_condition_to_script(condition, run_snippet, indent=indent),
1436 perform_substitution=perform_substitution,
1437 )
1439 def unconditionally_in_script(
1440 self,
1441 maintscript: Maintscript,
1442 run_snippet: str,
1443 /,
1444 perform_substitution: bool = True,
1445 ) -> None:
1446 if maintscript not in DPKG_DEB_CONTROL_SCRIPTS: 1446 ↛ 1447line 1446 didn't jump to line 1447 because the condition on line 1446 was never true
1447 raise ValueError(
1448 f'Unknown script "{maintscript}". Should have been one of:'
1449 f' {", ".join(sorted(DPKG_DEB_CONTROL_SCRIPTS))}'
1450 )
1451 return self._append_script(
1452 "unconditionally_in_script",
1453 maintscript,
1454 run_snippet,
1455 perform_substitution=perform_substitution,
1456 )
1459class MaintscriptAccessorProvider(MaintscriptAccessorProviderBase):
1460 __slots__ = (
1461 "_plugin_metadata",
1462 "_maintscript_snippets",
1463 "_plugin_source_id",
1464 "_package_substitution",
1465 "_default_snippet_anchor",
1466 )
1468 def __init__(
1469 self,
1470 plugin_metadata: DebputyPluginMetadata,
1471 plugin_source_id: str,
1472 maintscript_snippets: PackageMaintscriptSnippetContainer,
1473 package_substitution: Substitution,
1474 *,
1475 default_snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE,
1476 ):
1477 self._plugin_metadata = plugin_metadata
1478 self._plugin_source_id = plugin_source_id
1479 self._maintscript_snippets = maintscript_snippets
1480 self._package_substitution = package_substitution
1481 self._default_snippet_anchor = default_snippet_anchor
1483 def _append_script(
1484 self,
1485 caller_name: str,
1486 maintscript: Maintscript,
1487 full_script: str,
1488 /,
1489 perform_substitution: bool = True,
1490 ) -> None:
1491 def_source = f"{self._plugin_metadata.plugin_name} ({self._plugin_source_id})"
1492 if perform_substitution:
1493 full_script = self._package_substitution.substitute(full_script, def_source)
1495 snippet = UnboundMaintscriptSnippet(
1496 snippet=SnippetResolver.snippet(full_script),
1497 definition_source=def_source,
1498 snippet_anchor=self._default_snippet_anchor,
1499 )
1500 self._maintscript_snippets[maintscript].append(snippet)
1503class BinaryCtrlAccessorProviderBase(BinaryCtrlAccessor):
1504 __slots__ = (
1505 "_plugin_metadata",
1506 "_plugin_source_id",
1507 "_package_metadata_context",
1508 "_triggers",
1509 "_substvars",
1510 "_maintscript",
1511 "_shlibs_details",
1512 )
1514 def __init__(
1515 self,
1516 plugin_metadata: DebputyPluginMetadata,
1517 plugin_source_id: str,
1518 package_metadata_context: PackageProcessingContext,
1519 triggers: dict[tuple[DpkgTriggerType, str], PluginProvidedTrigger],
1520 substvars: FlushableSubstvars,
1521 shlibs_details: tuple[str | None, list[str] | None],
1522 ) -> None:
1523 self._plugin_metadata = plugin_metadata
1524 self._plugin_source_id = plugin_source_id
1525 self._package_metadata_context = package_metadata_context
1526 self._triggers = triggers
1527 self._substvars = substvars
1528 self._maintscript: MaintscriptAccessor | None = None
1529 self._shlibs_details = shlibs_details
1531 def _create_maintscript_accessor(self) -> MaintscriptAccessor:
1532 raise NotImplementedError
1534 def dpkg_trigger(self, trigger_type: DpkgTriggerType, trigger_target: str) -> None:
1535 """Register a declarative dpkg level trigger
1537 The provided trigger will be added to the package's metadata (the triggers file of the control.tar).
1539 If the trigger has already been added previously, a second call with the same trigger data will be ignored.
1540 """
1541 key = (trigger_type, trigger_target)
1542 if key in self._triggers: 1542 ↛ 1543line 1542 didn't jump to line 1543 because the condition on line 1542 was never true
1543 return
1544 self._triggers[key] = PluginProvidedTrigger(
1545 dpkg_trigger_type=trigger_type,
1546 dpkg_trigger_target=trigger_target,
1547 provider=self._plugin_metadata,
1548 provider_source_id=self._plugin_source_id,
1549 )
1551 @property
1552 def maintscript(self) -> MaintscriptAccessor:
1553 maintscript = self._maintscript
1554 if maintscript is None:
1555 maintscript = self._create_maintscript_accessor()
1556 self._maintscript = maintscript
1557 return maintscript
1559 @property
1560 def substvars(self) -> FlushableSubstvars:
1561 return self._substvars
1563 def dpkg_shlibdeps(self, paths: Sequence[VirtualPath]) -> None:
1564 binary_package = self._package_metadata_context.binary_package
1565 with self.substvars.flush() as substvars_file:
1566 dpkg_cmd = ["dpkg-shlibdeps", f"-T{substvars_file}"]
1567 if binary_package.is_udeb:
1568 dpkg_cmd.append("-tudeb")
1569 if binary_package.is_essential: 1569 ↛ 1570line 1569 didn't jump to line 1570 because the condition on line 1569 was never true
1570 dpkg_cmd.append("-dPre-Depends")
1571 shlibs_local, shlib_dirs = self._shlibs_details
1572 if shlibs_local is not None: 1572 ↛ 1573line 1572 didn't jump to line 1573 because the condition on line 1572 was never true
1573 dpkg_cmd.append(f"-L{shlibs_local}")
1574 if shlib_dirs: 1574 ↛ 1575line 1574 didn't jump to line 1575 because the condition on line 1574 was never true
1575 dpkg_cmd.extend(f"-l{sd}" for sd in shlib_dirs)
1576 dpkg_cmd.extend(p.fs_path for p in paths)
1577 print_command(*dpkg_cmd)
1578 try:
1579 subprocess.check_call(dpkg_cmd)
1580 except subprocess.CalledProcessError:
1581 _error(
1582 f"Attempting to auto-detect dependencies via dpkg-shlibdeps for {binary_package.name} failed. Please"
1583 " review the output from dpkg-shlibdeps above to understand what went wrong."
1584 )
1587class BinaryCtrlAccessorProvider(BinaryCtrlAccessorProviderBase):
1588 __slots__ = (
1589 "_maintscript",
1590 "_maintscript_snippets",
1591 "_package_substitution",
1592 )
1594 def __init__(
1595 self,
1596 plugin_metadata: DebputyPluginMetadata,
1597 plugin_source_id: str,
1598 package_metadata_context: PackageProcessingContext,
1599 triggers: dict[tuple[DpkgTriggerType, str], PluginProvidedTrigger],
1600 substvars: FlushableSubstvars,
1601 maintscript_snippets: PackageMaintscriptSnippetContainer,
1602 package_substitution: Substitution,
1603 shlibs_details: tuple[str | None, list[str] | None],
1604 *,
1605 default_snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE,
1606 ) -> None:
1607 super().__init__(
1608 plugin_metadata,
1609 plugin_source_id,
1610 package_metadata_context,
1611 triggers,
1612 substvars,
1613 shlibs_details,
1614 )
1615 self._maintscript_snippets = maintscript_snippets
1616 self._package_substitution = package_substitution
1617 self._maintscript = MaintscriptAccessorProvider(
1618 plugin_metadata,
1619 plugin_source_id,
1620 maintscript_snippets,
1621 package_substitution,
1622 default_snippet_anchor=default_snippet_anchor,
1623 )
1625 def _create_maintscript_accessor(self) -> MaintscriptAccessor:
1626 return MaintscriptAccessorProvider(
1627 self._plugin_metadata,
1628 self._plugin_source_id,
1629 self._maintscript_snippets,
1630 self._package_substitution,
1631 )
1634class BinaryCtrlAccessorProviderCreator:
1635 def __init__(
1636 self,
1637 package_metadata_context: PackageProcessingContext,
1638 substvars: FlushableSubstvars,
1639 maintscript_snippets: PackageMaintscriptSnippetContainer,
1640 substitution: Substitution,
1641 ) -> None:
1642 self._package_metadata_context = package_metadata_context
1643 self._substvars = substvars
1644 self._maintscript_snippets = maintscript_snippets
1645 self._substitution = substitution
1646 self._triggers: dict[tuple[DpkgTriggerType, str], PluginProvidedTrigger] = {}
1647 self.shlibs_details: tuple[str | None, list[str] | None] = None, None
1649 def for_plugin(
1650 self,
1651 plugin_metadata: DebputyPluginMetadata,
1652 plugin_source_id: str,
1653 *,
1654 default_snippet_anchor: SnippetAnchor | None = None,
1655 ) -> BinaryCtrlAccessor:
1656 return BinaryCtrlAccessorProvider(
1657 plugin_metadata,
1658 plugin_source_id,
1659 self._package_metadata_context,
1660 self._triggers,
1661 self._substvars,
1662 self._maintscript_snippets,
1663 self._substitution,
1664 self.shlibs_details,
1665 default_snippet_anchor=SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE,
1666 )
1668 def generated_triggers(self) -> Iterable[PluginProvidedTrigger]:
1669 return self._triggers.values()
1672def _resolve_bundled_plugin_docs_path(
1673 plugin_name: str,
1674 loader: PluginInitializationEntryPoint | None,
1675) -> Traversable | Path | None:
1676 plugin_module = getattr(loader, "__module__")
1677 assert plugin_module is not None
1678 plugin_package_name = sys.modules[plugin_module].__package__
1679 return importlib.resources.files(plugin_package_name).joinpath(
1680 f"{plugin_name}_docs.yaml"
1681 )
1684def plugin_metadata_for_debputys_own_plugin(
1685 loader: PluginInitializationEntryPoint | None = None,
1686) -> DebputyPluginMetadata:
1687 if loader is None:
1688 from debputy.plugins.debputy.debputy_plugin import (
1689 initialize_debputy_features,
1690 )
1692 loader = initialize_debputy_features
1693 plugin_name = "debputy"
1694 return DebputyPluginMetadata(
1695 plugin_name="debputy",
1696 api_compat_version=1,
1697 plugin_initializer=loader,
1698 plugin_loader=None,
1699 plugin_doc_path_resolver=lambda: _resolve_bundled_plugin_docs_path(
1700 plugin_name,
1701 loader,
1702 ),
1703 plugin_path="<bundled>",
1704 )
1707def load_plugin_features(
1708 plugin_search_dirs: Sequence[str],
1709 substitution: Substitution,
1710 requested_plugins_only: Sequence[str] | None = None,
1711 required_plugins: set[str] | None = None,
1712 plugin_feature_set: PluginProvidedFeatureSet | None = None,
1713 debug_mode: bool = False,
1714) -> PluginProvidedFeatureSet:
1715 if plugin_feature_set is None:
1716 plugin_feature_set = PluginProvidedFeatureSet()
1717 plugins = [plugin_metadata_for_debputys_own_plugin()]
1718 unloadable_plugins = set()
1719 if required_plugins:
1720 plugins.extend(
1721 find_json_plugins(
1722 plugin_search_dirs,
1723 required_plugins,
1724 )
1725 )
1726 if requested_plugins_only is not None:
1727 plugins.extend(
1728 find_json_plugins(
1729 plugin_search_dirs,
1730 requested_plugins_only,
1731 )
1732 )
1733 else:
1734 auto_loaded = _find_all_json_plugins(
1735 plugin_search_dirs,
1736 required_plugins if required_plugins is not None else frozenset(),
1737 debug_mode=debug_mode,
1738 )
1739 for plugin_metadata in auto_loaded:
1740 plugins.append(plugin_metadata)
1741 unloadable_plugins.add(plugin_metadata.plugin_name)
1743 for plugin_metadata in plugins:
1744 api = DebputyPluginInitializerProvider(
1745 plugin_metadata, plugin_feature_set, substitution
1746 )
1747 try:
1748 api.load_plugin()
1749 except PluginBaseError as e:
1750 if plugin_metadata.plugin_name not in unloadable_plugins:
1751 raise
1752 if debug_mode:
1753 _warn(
1754 f"The optional plugin {plugin_metadata.plugin_name} failed during load. Re-raising due"
1755 f" to --debug/-d or DEBPUTY_DEBUG=1"
1756 )
1757 raise
1758 try:
1759 api.unload_plugin()
1760 except Exception:
1761 _warn(
1762 f"Failed to load optional {plugin_metadata.plugin_name} and an error was raised when trying to"
1763 " clean up after the half-initialized plugin. Re-raising load error as the partially loaded"
1764 " module might have tainted the feature set."
1765 )
1766 raise e from None
1767 _warn(
1768 f"The optional plugin {plugin_metadata.plugin_name} failed during load. The plugin was"
1769 f" deactivated. Use debug mode (--debug/DEBPUTY_DEBUG=1) to show the stacktrace"
1770 f" (the warning will become an error)"
1771 )
1773 return plugin_feature_set
1776def find_json_plugin(
1777 search_dirs: Sequence[str],
1778 requested_plugin: str,
1779) -> DebputyPluginMetadata:
1780 r = list(find_json_plugins(search_dirs, [requested_plugin]))
1781 assert len(r) == 1
1782 return r[0]
1785def find_related_implementation_files_for_plugin(
1786 plugin_metadata: DebputyPluginMetadata,
1787) -> list[str]:
1788 if plugin_metadata.is_bundled:
1789 plugin_name = plugin_metadata.plugin_name
1790 _error(
1791 f"Cannot run find related files for {plugin_name}: The plugin seems to be bundled"
1792 " or loaded via a mechanism that does not support detecting its tests."
1793 )
1795 if plugin_metadata.is_from_python_path:
1796 plugin_name = plugin_metadata.plugin_name
1797 # Maybe they could be, but that is for another day.
1798 _error(
1799 f"Cannot run find related files for {plugin_name}: The plugin is installed into python path"
1800 " and these are not supported."
1801 )
1802 files = []
1803 module_name, module_file = _find_plugin_implementation_file(
1804 plugin_metadata.plugin_name,
1805 plugin_metadata.plugin_path,
1806 )
1807 if os.path.isfile(module_file):
1808 files.append(module_file)
1809 else:
1810 if not plugin_metadata.is_loaded:
1811 plugin_metadata.load_plugin()
1812 if module_name in sys.modules:
1813 _error(
1814 f'The plugin {plugin_metadata.plugin_name} uses the "module"" key in its'
1815 f" JSON metadata file ({plugin_metadata.plugin_path}) and cannot be "
1816 f" installed via this method. The related Python would not be installed"
1817 f" (which would result in a plugin that would fail to load)"
1818 )
1820 return files
1823def find_tests_for_plugin(
1824 plugin_metadata: DebputyPluginMetadata,
1825) -> list[str]:
1826 plugin_name = plugin_metadata.plugin_name
1827 plugin_path = plugin_metadata.plugin_path
1829 if plugin_metadata.is_bundled:
1830 _error(
1831 f"Cannot run tests for {plugin_name}: The plugin seems to be bundled or loaded via a"
1832 " mechanism that does not support detecting its tests."
1833 )
1835 if plugin_metadata.is_from_python_path:
1836 plugin_name = plugin_metadata.plugin_name
1837 # Maybe they could be, but that is for another day.
1838 _error(
1839 f"Cannot run find related files for {plugin_name}: The plugin is installed into python path"
1840 " and these are not supported."
1841 )
1843 plugin_dir = os.path.dirname(plugin_path)
1844 test_basename_prefix = plugin_metadata.plugin_name.replace("-", "_")
1845 tests = []
1846 with os.scandir(plugin_dir) as dir_iter:
1847 for p in dir_iter:
1848 if (
1849 p.is_file()
1850 and p.name.startswith(test_basename_prefix)
1851 and PLUGIN_TEST_SUFFIX.search(p.name)
1852 ):
1853 tests.append(p.path)
1854 return tests
1857def find_json_plugins(
1858 search_dirs: Sequence[str],
1859 requested_plugins: Iterable[str],
1860) -> Iterable[DebputyPluginMetadata]:
1861 for plugin_name_or_path in requested_plugins: 1861 ↛ exitline 1861 didn't return from function 'find_json_plugins' because the loop on line 1861 didn't complete
1862 if "/" in plugin_name_or_path: 1862 ↛ 1863line 1862 didn't jump to line 1863 because the condition on line 1862 was never true
1863 if not os.path.isfile(plugin_name_or_path):
1864 raise PluginNotFoundError(
1865 f"Unable to load the plugin {plugin_name_or_path}: The path is not a file."
1866 ' (Because the plugin name contains "/", it is assumed to be a path and search path'
1867 " is not used."
1868 )
1869 yield parse_json_plugin_desc(plugin_name_or_path)
1870 return
1871 for search_dir in search_dirs: 1871 ↛ 1880line 1871 didn't jump to line 1880 because the loop on line 1871 didn't complete
1872 path = os.path.join(
1873 search_dir, "debputy", "plugins", f"{plugin_name_or_path}.json"
1874 )
1875 if not os.path.isfile(path): 1875 ↛ 1876line 1875 didn't jump to line 1876 because the condition on line 1875 was never true
1876 continue
1877 yield parse_json_plugin_desc(path)
1878 return
1880 path_root = PLUGIN_PYTHON_RES_PATH
1881 pp_path = path_root.joinpath(f"{plugin_name_or_path}.json")
1882 if pp_path or pp_path.is_file():
1883 with pp_path.open() as fd:
1884 yield parse_json_plugin_desc(
1885 f"PYTHONPATH:debputy/plugins/{pp_path.name}",
1886 fd=fd,
1887 is_from_python_path=True,
1888 )
1889 return
1891 search_dir_str = ":".join(search_dirs)
1892 raise PluginNotFoundError(
1893 f"Unable to load the plugin {plugin_name_or_path}: Could not find {plugin_name_or_path}.json in the"
1894 f" debputy/plugins subdir of any of the search dirs ({search_dir_str})"
1895 )
1898def _find_all_json_plugins(
1899 search_dirs: Sequence[str],
1900 required_plugins: AbstractSet[str],
1901 debug_mode: bool = False,
1902) -> Iterable[DebputyPluginMetadata]:
1903 seen = set(required_plugins)
1904 error_seen = False
1905 for search_dir in search_dirs:
1906 try:
1907 dir_fd = os.scandir(os.path.join(search_dir, "debputy", "plugins"))
1908 except FileNotFoundError:
1909 continue
1910 with dir_fd:
1911 for entry in dir_fd:
1912 if (
1913 not entry.is_file(follow_symlinks=True)
1914 or not entry.name.endswith(".json")
1915 or entry.name in seen
1916 ):
1917 continue
1918 seen.add(entry.name)
1919 try:
1920 plugin_metadata = parse_json_plugin_desc(entry.path)
1921 except PluginBaseError as e:
1922 if debug_mode:
1923 raise
1924 if not error_seen:
1925 error_seen = True
1926 _warn(
1927 f"Failed to load the plugin in {entry.path} due to the following error: {e.message}"
1928 )
1929 else:
1930 _warn(
1931 f"Failed to load plugin in {entry.path} due to errors (not shown)."
1932 )
1933 else:
1934 yield plugin_metadata
1936 for pp_entry in PLUGIN_PYTHON_RES_PATH.iterdir():
1937 if (
1938 not pp_entry.name.endswith(".json")
1939 or not pp_entry.is_file()
1940 or pp_entry.name in seen
1941 ):
1942 continue
1943 seen.add(pp_entry.name)
1944 with pp_entry.open() as fd:
1945 yield parse_json_plugin_desc(
1946 f"PYTHONPATH:debputy/plugins/{pp_entry.name}",
1947 fd=fd,
1948 is_from_python_path=True,
1949 )
1952def _find_plugin_implementation_file(
1953 plugin_name: str,
1954 json_file_path: str,
1955) -> tuple[str, str]:
1956 guessed_module_basename = plugin_name.replace("-", "_")
1957 module_name = f"debputy.plugins.{guessed_module_basename}"
1958 module_fs_path = os.path.join(
1959 os.path.dirname(json_file_path), f"{guessed_module_basename}.py"
1960 )
1961 return module_name, module_fs_path
1964def _resolve_module_initializer(
1965 plugin_name: str,
1966 plugin_initializer_name: str,
1967 module_name: str | None,
1968 json_file_path: str,
1969) -> PluginInitializationEntryPoint:
1970 module = None
1971 module_fs_path = None
1972 if module_name is None: 1972 ↛ 2000line 1972 didn't jump to line 2000 because the condition on line 1972 was always true
1973 module_name, module_fs_path = _find_plugin_implementation_file(
1974 plugin_name, json_file_path
1975 )
1976 if os.path.isfile(module_fs_path): 1976 ↛ 2000line 1976 didn't jump to line 2000 because the condition on line 1976 was always true
1977 spec = importlib.util.spec_from_file_location(module_name, module_fs_path)
1978 if spec is None: 1978 ↛ 1979line 1978 didn't jump to line 1979 because the condition on line 1978 was never true
1979 raise PluginInitializationError(
1980 f"Failed to load {plugin_name} (path: {module_fs_path})."
1981 " The spec_from_file_location function returned None."
1982 )
1983 mod = importlib.util.module_from_spec(spec)
1984 loader = spec.loader
1985 if loader is None: 1985 ↛ 1986line 1985 didn't jump to line 1986 because the condition on line 1985 was never true
1986 raise PluginInitializationError(
1987 f"Failed to load {plugin_name} (path: {module_fs_path})."
1988 " Python could not find a suitable loader (spec.loader was None)"
1989 )
1990 sys.modules[module_name] = mod
1991 try:
1992 run_in_context_of_plugin(plugin_name, loader.exec_module, mod)
1993 except (Exception, GeneratorExit) as e:
1994 raise PluginInitializationError(
1995 f"Failed to load {plugin_name} (path: {module_fs_path})."
1996 " The module threw an exception while being loaded."
1997 ) from e
1998 module = mod
2000 if module is None: 2000 ↛ 2001line 2000 didn't jump to line 2001 because the condition on line 2000 was never true
2001 try:
2002 module = run_in_context_of_plugin(
2003 plugin_name, importlib.import_module, module_name
2004 )
2005 except ModuleNotFoundError as e:
2006 if module_fs_path is None:
2007 raise PluginMetadataError(
2008 f'The plugin defined in "{json_file_path}" wanted to load the module "{module_name}", but'
2009 " this module is not available in the python search path"
2010 ) from e
2011 raise PluginInitializationError(
2012 f"Failed to load {plugin_name}. Tried loading it from"
2013 f' "{module_fs_path}" (which did not exist) and PYTHONPATH as'
2014 f" {module_name} (where it was not found either). Please ensure"
2015 " the module code is installed in the correct spot or provide an"
2016 f' explicit "module" definition in {json_file_path}.'
2017 ) from e
2019 plugin_initializer = run_in_context_of_plugin_wrap_errors(
2020 plugin_name,
2021 getattr,
2022 module,
2023 plugin_initializer_name,
2024 None,
2025 )
2027 if plugin_initializer is None: 2027 ↛ 2028line 2027 didn't jump to line 2028 because the condition on line 2027 was never true
2028 raise PluginMetadataError(
2029 f'The plugin defined in {json_file_path} claimed that module "{module_name}" would have an'
2030 f' attribute called "{plugin_initializer_name}" to initialize the plugin. However, that attribute'
2031 " does not exist or cannot be resolved. Please correct the plugin metadata or initializer name"
2032 " in the Python module."
2033 )
2034 if isinstance(plugin_initializer, DebputyPluginDefinition):
2035 return plugin_initializer.initialize
2036 if not callable(plugin_initializer): 2036 ↛ 2037line 2036 didn't jump to line 2037 because the condition on line 2036 was never true
2037 raise PluginMetadataError(
2038 f'The plugin defined in {json_file_path} claimed that module "{module_name}" would have an'
2039 f' attribute called "{plugin_initializer_name}" for initializing the plugin. While that'
2040 " attribute exists, it is neither a `DebputyPluginDefinition`"
2041 " (`plugin_definition = define_debputy_plugin()`) nor is it `callable`"
2042 " (`def initialize(api: DebputyPluginInitializer) -> None:`)."
2043 )
2044 return cast("PluginInitializationEntryPoint", plugin_initializer)
2047def _json_plugin_loader(
2048 plugin_name: str,
2049 plugin_json_metadata: PluginJsonMetadata,
2050 json_file_path: str,
2051 attribute_path: AttributePath,
2052) -> Callable[["DebputyPluginInitializer"], None]:
2053 api_compat = plugin_json_metadata["api_compat_version"]
2054 module_name = plugin_json_metadata.get("module")
2055 plugin_initializer_name = plugin_json_metadata.get("plugin_initializer")
2056 packager_provided_files_raw = plugin_json_metadata.get(
2057 "packager_provided_files", []
2058 )
2059 manifest_variables_raw = plugin_json_metadata.get("manifest_variables")
2060 known_packaging_files_raw = plugin_json_metadata.get("known_packaging_files")
2061 if api_compat != 1: 2061 ↛ 2062line 2061 didn't jump to line 2062 because the condition on line 2061 was never true
2062 raise PluginMetadataError(
2063 f'The plugin defined in "{json_file_path}" requires API compat level {api_compat}, but this'
2064 f" version of debputy only supports API compat version of 1"
2065 )
2066 if plugin_initializer_name is not None and "." in plugin_initializer_name: 2066 ↛ 2067line 2066 didn't jump to line 2067 because the condition on line 2066 was never true
2067 p = attribute_path["plugin_initializer"]
2068 raise PluginMetadataError(
2069 f'The "{p}" attribute must not contain ".". Problematic file is "{json_file_path}".'
2070 )
2072 plugin_initializers = []
2074 if plugin_initializer_name is not None:
2075 plugin_initializer = _resolve_module_initializer(
2076 plugin_name,
2077 plugin_initializer_name,
2078 module_name,
2079 json_file_path,
2080 )
2081 plugin_initializers.append(plugin_initializer)
2083 if known_packaging_files_raw:
2084 kpf_root_path = attribute_path["known_packaging_files"]
2085 known_packaging_files = []
2086 for k, v in enumerate(known_packaging_files_raw):
2087 kpf_path = kpf_root_path[k]
2088 p = v.get("path")
2089 if isinstance(p, str): 2089 ↛ 2091line 2089 didn't jump to line 2091 because the condition on line 2089 was always true
2090 kpf_path.path_hint = p
2091 if plugin_name.startswith("debputy-") and isinstance(v, dict): 2091 ↛ 2103line 2091 didn't jump to line 2103 because the condition on line 2091 was always true
2092 docs = v.get("documentation-uris")
2093 if docs is not None and isinstance(docs, list):
2094 docs = [
2095 (
2096 d.replace("@DEBPUTY_DOC_ROOT_DIR@", debputy_doc_root_dir())
2097 if isinstance(d, str)
2098 else d
2099 )
2100 for d in docs
2101 ]
2102 v["documentation-uris"] = docs
2103 known_packaging_file: KnownPackagingFileInfo = (
2104 PLUGIN_KNOWN_PACKAGING_FILES_PARSER.parse_input(
2105 v,
2106 kpf_path,
2107 )
2108 )
2109 known_packaging_files.append((kpf_path, known_packaging_file))
2111 def _initialize_json_provided_known_packaging_files(
2112 api: DebputyPluginInitializerProvider,
2113 ) -> None:
2114 for p, details in known_packaging_files:
2115 try:
2116 api.known_packaging_files(details)
2117 except ValueError as ex:
2118 raise PluginMetadataError(
2119 f"Error while processing {p.path} defined in {json_file_path}: {ex.args[0]}"
2120 )
2122 plugin_initializers.append(_initialize_json_provided_known_packaging_files)
2124 if manifest_variables_raw:
2125 manifest_var_path = attribute_path["manifest_variables"]
2126 manifest_variables = [
2127 PLUGIN_MANIFEST_VARS_PARSER.parse_input(p, manifest_var_path[i])
2128 for i, p in enumerate(manifest_variables_raw)
2129 ]
2131 def _initialize_json_provided_manifest_vars(
2132 api: DebputyPluginInitializer,
2133 ) -> None:
2134 for idx, manifest_variable in enumerate(manifest_variables):
2135 name = manifest_variable["name"]
2136 value = manifest_variable["value"]
2137 doc = manifest_variable.get("reference_documentation")
2138 try:
2139 api.manifest_variable(
2140 name, value, variable_reference_documentation=doc
2141 )
2142 except ValueError as ex:
2143 var_path = manifest_var_path[idx]
2144 raise PluginMetadataError(
2145 f"Error while processing {var_path.path} defined in {json_file_path}: {ex.args[0]}"
2146 )
2148 plugin_initializers.append(_initialize_json_provided_manifest_vars)
2150 if packager_provided_files_raw:
2151 ppf_path = attribute_path["packager_provided_files"]
2152 ppfs = [
2153 PLUGIN_PPF_PARSER.parse_input(p, ppf_path[i])
2154 for i, p in enumerate(packager_provided_files_raw)
2155 ]
2157 def _initialize_json_provided_ppfs(api: DebputyPluginInitializer) -> None:
2158 ppf: PackagerProvidedFileJsonDescription
2159 for idx, ppf in enumerate(ppfs):
2160 c = dict(ppf)
2161 stem = ppf["stem"]
2162 installed_path = ppf["installed_path"]
2163 default_mode = ppf.get("default_mode")
2164 ref_doc_dict = ppf.get("reference_documentation")
2165 if default_mode is not None: 2165 ↛ 2168line 2165 didn't jump to line 2168 because the condition on line 2165 was always true
2166 c["default_mode"] = default_mode.octal_mode
2168 if ref_doc_dict is not None: 2168 ↛ 2173line 2168 didn't jump to line 2173 because the condition on line 2168 was always true
2169 ref_doc = packager_provided_file_reference_documentation(
2170 **ref_doc_dict
2171 )
2172 else:
2173 ref_doc = None
2175 for k in [
2176 "stem",
2177 "installed_path",
2178 "reference_documentation",
2179 ]:
2180 try:
2181 del c[k]
2182 except KeyError:
2183 pass
2185 try:
2186 api.packager_provided_file(stem, installed_path, reference_documentation=ref_doc, **c) # type: ignore
2187 except ValueError as ex:
2188 p_path = ppf_path[idx]
2189 raise PluginMetadataError(
2190 f"Error while processing {p_path.path} defined in {json_file_path}: {ex.args[0]}"
2191 )
2193 plugin_initializers.append(_initialize_json_provided_ppfs)
2195 if not plugin_initializers: 2195 ↛ 2196line 2195 didn't jump to line 2196 because the condition on line 2195 was never true
2196 raise PluginMetadataError(
2197 f"The plugin defined in {json_file_path} does not seem to provide features"
2198 f" known by this version of `debputy`, such as module + plugin-initializer"
2199 f" or packager-provided-files. The plugin might be missing content, or"
2200 f" this version of `debputy` might not have the feature set to support it."
2201 )
2203 if len(plugin_initializers) == 1:
2204 return plugin_initializers[0]
2206 def _chain_loader(api: DebputyPluginInitializer) -> None:
2207 for initializer in plugin_initializers:
2208 initializer(api)
2210 return _chain_loader
2213@overload
2214@contextlib.contextmanager
2215def _open( 2215 ↛ exitline 2215 didn't return from function '_open' because
2216 path: str,
2217 fd: IO[AnyStr] | IOBase = ...,
2218) -> Iterator[IO[AnyStr] | IOBase]: ...
2221@overload
2222@contextlib.contextmanager
2223def _open(path: str, fd: None = None) -> Iterator[IO[bytes]]: ... 2223 ↛ exitline 2223 didn't return from function '_open' because
2226@contextlib.contextmanager
2227def _open(
2228 path: str, fd: IO[AnyStr] | IOBase | None = None
2229) -> Iterator[IO[AnyStr] | IOBase]:
2230 if fd is not None:
2231 yield fd
2232 else:
2233 with open(path, "rb") as fd:
2234 yield fd
2237def _resolve_json_plugin_docs_path(
2238 plugin_name: str,
2239 plugin_path: str,
2240) -> Traversable | Path | None:
2241 plugin_dir = os.path.dirname(plugin_path)
2242 return Path(os.path.join(plugin_dir, plugin_name + "_docs.yaml"))
2245def parse_json_plugin_desc(
2246 path: str,
2247 *,
2248 fd: IO[AnyStr] | IOBase | None = None,
2249 is_from_python_path: bool = False,
2250) -> DebputyPluginMetadata:
2251 with _open(path, fd=fd) as rfd:
2252 try:
2253 raw = json.load(rfd)
2254 except JSONDecodeError as e:
2255 raise PluginMetadataError(
2256 f'The plugin defined in "{path}" could not be parsed as valid JSON: {e.args[0]}'
2257 ) from e
2258 plugin_name = os.path.basename(path)
2259 if plugin_name.endswith(".json"):
2260 plugin_name = plugin_name[:-5]
2261 elif plugin_name.endswith(".json.in"):
2262 plugin_name = plugin_name[:-8]
2264 if plugin_name == "debputy": 2264 ↛ 2266line 2264 didn't jump to line 2266 because the condition on line 2264 was never true
2265 # Provide a better error message than "The plugin has already loaded!?"
2266 raise PluginMetadataError(
2267 f'The plugin named {plugin_name} must be bundled with `debputy`. Please rename "{path}" so it does not'
2268 f" clash with the bundled plugin of same name."
2269 )
2271 attribute_path = AttributePath.root_path(raw)
2273 try:
2274 plugin_json_metadata = PLUGIN_METADATA_PARSER.parse_input(
2275 raw,
2276 attribute_path,
2277 )
2278 except ManifestParseException as e:
2279 raise PluginMetadataError(
2280 f'The plugin defined in "{path}" was valid JSON but could not be parsed: {e.message}'
2281 ) from e
2282 api_compat = plugin_json_metadata["api_compat_version"]
2284 return DebputyPluginMetadata(
2285 plugin_name=plugin_name,
2286 plugin_loader=lambda: _json_plugin_loader(
2287 plugin_name,
2288 plugin_json_metadata,
2289 path,
2290 attribute_path,
2291 ),
2292 api_compat_version=api_compat,
2293 plugin_doc_path_resolver=lambda: _resolve_json_plugin_docs_path(
2294 plugin_name, path
2295 ),
2296 plugin_initializer=None,
2297 plugin_path=path,
2298 is_from_python_path=is_from_python_path,
2299 )
2302@dataclasses.dataclass(slots=True, frozen=True)
2303class ServiceDefinitionImpl(ServiceDefinition[DSD]):
2304 name: str
2305 names: Sequence[str]
2306 path: VirtualPath
2307 type_of_service: str
2308 service_scope: str
2309 auto_enable_on_install: bool
2310 auto_start_on_install: bool
2311 on_upgrade: ServiceUpgradeRule
2312 definition_source: str
2313 is_plugin_provided_definition: bool
2314 service_context: DSD | None
2316 def replace(self, **changes: Any) -> "ServiceDefinitionImpl[DSD]":
2317 return dataclasses.replace(self, **changes)
2320class ServiceRegistryImpl(ServiceRegistry[DSD]):
2321 __slots__ = ("_service_manager_details", "_service_definitions", "_seen_services")
2323 def __init__(self, service_manager_details: ServiceManagerDetails) -> None:
2324 self._service_manager_details = service_manager_details
2325 self._service_definitions: list[ServiceDefinition[DSD]] = []
2326 self._seen_services: set[tuple[str, str, str]] = set()
2328 @property
2329 def detected_services(self) -> Sequence[ServiceDefinition[DSD]]:
2330 return self._service_definitions
2332 def register_service(
2333 self,
2334 path: VirtualPath,
2335 name: str | list[str],
2336 *,
2337 type_of_service: str = "service", # "timer", etc.
2338 service_scope: str = "system",
2339 enable_by_default: bool = True,
2340 start_by_default: bool = True,
2341 default_upgrade_rule: ServiceUpgradeRule = "restart",
2342 service_context: DSD | None = None,
2343 ) -> None:
2344 names = name if isinstance(name, list) else [name]
2345 if len(names) < 1:
2346 raise ValueError(
2347 f"The service must have at least one name - {path.absolute} did not have any"
2348 )
2349 for n in names:
2350 key = (n, type_of_service, service_scope)
2351 if key in self._seen_services:
2352 raise PluginAPIViolationError(
2353 f"The service manager (from {self._service_manager_details.plugin_metadata.plugin_name}) used"
2354 f" the service name {n} (type: {type_of_service}, scope: {service_scope}) twice. This is not"
2355 " allowed by the debputy plugin API."
2356 )
2357 # TODO: We cannot create a service definition immediate once the manifest is involved
2358 self._service_definitions.append(
2359 ServiceDefinitionImpl(
2360 names[0],
2361 names,
2362 path,
2363 type_of_service,
2364 service_scope,
2365 enable_by_default,
2366 start_by_default,
2367 default_upgrade_rule,
2368 f"Auto-detected by plugin {self._service_manager_details.plugin_metadata.plugin_name}",
2369 True,
2370 service_context,
2371 )
2372 )