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