Coverage for src/debputy/plugin/api/spec.py: 88%
322 statements
« prev ^ index » next coverage.py v7.6.0, created at 2025-01-27 13:59 +0000
1import contextlib
2import dataclasses
3import os
4import tempfile
5import textwrap
6from typing import (
7 Iterable,
8 Optional,
9 Callable,
10 Literal,
11 Union,
12 Iterator,
13 overload,
14 FrozenSet,
15 Sequence,
16 TypeVar,
17 Any,
18 TYPE_CHECKING,
19 TextIO,
20 BinaryIO,
21 Generic,
22 ContextManager,
23 List,
24 Type,
25 Tuple,
26 get_args,
27 Container,
28 final,
29)
31from debian.substvars import Substvars
33from debputy import util
34from debputy.exceptions import TestPathWithNonExistentFSPathError, PureVirtualPathError
35from debputy.interpreter import Interpreter, extract_shebang_interpreter_from_file
36from debputy.manifest_parser.tagging_types import DebputyDispatchableType
37from debputy.manifest_parser.util import parse_symbolic_mode
38from debputy.packages import BinaryPackage
39from debputy.types import S
41if TYPE_CHECKING:
42 from debputy.manifest_parser.base_types import (
43 StaticFileSystemOwner,
44 StaticFileSystemGroup,
45 )
46 from debputy.plugin.api.doc_parsing import ParserRefDocumentation
47 from debputy.plugin.api.impl_types import DIPHandler
48 from debputy.plugin.debputy.to_be_api_types import BuildSystemRule
51DP = TypeVar("DP", bound=DebputyDispatchableType)
54PluginInitializationEntryPoint = Callable[["DebputyPluginInitializer"], None]
55MetadataAutoDetector = Callable[
56 ["VirtualPath", "BinaryCtrlAccessor", "PackageProcessingContext"], None
57]
58PackageProcessor = Callable[["VirtualPath", None, "PackageProcessingContext"], None]
59DpkgTriggerType = Literal[
60 "activate",
61 "activate-await",
62 "activate-noawait",
63 "interest",
64 "interest-await",
65 "interest-noawait",
66]
67Maintscript = Literal["postinst", "preinst", "prerm", "postrm"]
68PackageTypeSelector = Union[Literal["deb", "udeb"], Iterable[Literal["deb", "udeb"]]]
69ServiceUpgradeRule = Literal[
70 "do-nothing",
71 "reload",
72 "restart",
73 "stop-then-start",
74]
76DSD = TypeVar("DSD")
77ServiceDetector = Callable[
78 ["VirtualPath", "ServiceRegistry[DSD]", "PackageProcessingContext"],
79 None,
80]
81ServiceIntegrator = Callable[
82 [
83 Sequence["ServiceDefinition[DSD]"],
84 "BinaryCtrlAccessor",
85 "PackageProcessingContext",
86 ],
87 None,
88]
90PMT = TypeVar("PMT")
91DebputyIntegrationMode = Literal[
92 "full",
93 "dh-sequence-zz-debputy",
94 "dh-sequence-zz-debputy-rrr",
95]
97INTEGRATION_MODE_FULL: DebputyIntegrationMode = "full"
98INTEGRATION_MODE_DH_DEBPUTY_RRR: DebputyIntegrationMode = "dh-sequence-zz-debputy-rrr"
99INTEGRATION_MODE_DH_DEBPUTY: DebputyIntegrationMode = "dh-sequence-zz-debputy"
100ALL_DEBPUTY_INTEGRATION_MODES: FrozenSet[DebputyIntegrationMode] = frozenset(
101 get_args(DebputyIntegrationMode)
102)
104_DEBPUTY_DISPATCH_METADATA_ATTR_NAME = "_debputy_dispatch_metadata"
107def only_integrations(
108 *integrations: DebputyIntegrationMode,
109) -> Container[DebputyIntegrationMode]:
110 return frozenset(integrations)
113def not_integrations(
114 *integrations: DebputyIntegrationMode,
115) -> Container[DebputyIntegrationMode]:
116 return ALL_DEBPUTY_INTEGRATION_MODES - frozenset(integrations)
119@dataclasses.dataclass(slots=True, frozen=True)
120class PackagerProvidedFileReferenceDocumentation:
121 description: Optional[str] = None
122 format_documentation_uris: Sequence[str] = tuple()
124 def replace(self, **changes: Any) -> "PackagerProvidedFileReferenceDocumentation":
125 return dataclasses.replace(self, **changes)
128def packager_provided_file_reference_documentation(
129 *,
130 description: Optional[str] = None,
131 format_documentation_uris: Optional[Sequence[str]] = tuple(),
132) -> PackagerProvidedFileReferenceDocumentation:
133 """Provide documentation for a given packager provided file.
135 :param description: Textual description presented to the user.
136 :param format_documentation_uris: A sequence of URIs to documentation that describes
137 the format of the file. Most relevant first.
138 :return:
139 """
140 uris = tuple(format_documentation_uris) if format_documentation_uris else tuple()
141 return PackagerProvidedFileReferenceDocumentation(
142 description=description,
143 format_documentation_uris=uris,
144 )
147class PathMetadataReference(Generic[PMT]):
148 """An accessor to plugin provided metadata
150 This is a *short-lived* reference to a piece of metadata. It should *not* be stored beyond
151 the boundaries of the current plugin execution context as it can be become invalid (as an
152 example, if the path associated with this path is removed, then this reference become invalid)
153 """
155 @property
156 def is_present(self) -> bool:
157 """Determine whether the value has been set
159 If the current plugin cannot access the value, then this method unconditionally returns
160 `False` regardless of whether the value is there.
162 :return: `True` if the value has been set to a not None value (and not been deleted).
163 Otherwise, this property is `False`.
164 """
165 raise NotImplementedError
167 @property
168 def can_read(self) -> bool:
169 """Test whether it is possible to read the metadata
171 Note: That the metadata being readable does *not* imply that the metadata is present.
173 :return: True if it is possible to read the metadata. This is always True for the
174 owning plugin.
175 """
176 raise NotImplementedError
178 @property
179 def can_write(self) -> bool:
180 """Test whether it is possible to update the metadata
182 :return: True if it is possible to update the metadata.
183 """
184 raise NotImplementedError
186 @property
187 def value(self) -> Optional[PMT]:
188 """Fetch the currently stored value if present.
190 :return: The value previously stored if any. Returns `None` if the value was never
191 stored, explicitly set to `None` or was deleted.
192 """
193 raise NotImplementedError
195 @value.setter
196 def value(self, value: Optional[PMT]) -> None:
197 """Replace any current value with the provided value
199 This operation is only possible if the path is writable *and* the caller is from
200 the owning plugin OR the owning plugin made the reference read-write.
201 """
202 raise NotImplementedError
204 @value.deleter
205 def value(self) -> None:
206 """Delete any current value.
208 This has the same effect as setting the value to `None`. It has the same restrictions
209 as the value setter.
210 """
211 self.value = None
214@dataclasses.dataclass(slots=True)
215class PathDef:
216 path_name: str
217 mode: Optional[int] = None
218 mtime: Optional[int] = None
219 has_fs_path: Optional[bool] = None
220 fs_path: Optional[str] = None
221 link_target: Optional[str] = None
222 content: Optional[str] = None
223 materialized_content: Optional[str] = None
226@dataclasses.dataclass(slots=True, frozen=True)
227class DispatchablePluggableManifestRuleMetadata(Generic[DP]):
228 """NOT PUBLIC API (used internally by part of the public API)"""
230 manifest_keywords: Sequence[str]
231 dispatched_type: Type[DP]
232 unwrapped_constructor: "DIPHandler"
233 expected_debputy_integration_mode: Optional[Container[DebputyIntegrationMode]] = (
234 None
235 )
236 online_reference_documentation: Optional["ParserDocumentation"] = None
237 apply_standard_attribute_documentation: bool = False
238 source_format: Optional[Any] = None
241@dataclasses.dataclass(slots=True, frozen=True)
242class BuildSystemManifestRuleMetadata(DispatchablePluggableManifestRuleMetadata):
243 build_system_impl: Optional[Type["BuildSystemRule"]] = None
244 auto_detection_shadow_build_systems: FrozenSet[str] = frozenset()
247def virtual_path_def(
248 path_name: str,
249 /,
250 mode: Optional[int] = None,
251 mtime: Optional[int] = None,
252 fs_path: Optional[str] = None,
253 link_target: Optional[str] = None,
254 content: Optional[str] = None,
255 materialized_content: Optional[str] = None,
256) -> PathDef:
257 """Define a virtual path for use with examples or, in tests, `build_virtual_file_system`
259 :param path_name: The full path. Must start with "./". If it ends with "/", the path will be interpreted
260 as a directory (the `is_dir` attribute will be True). Otherwise, it will be a symlink or file depending
261 on whether a `link_target` is provided.
262 :param mode: The mode to use for this path. Defaults to 0644 for files and 0755 for directories. The mode
263 should be None for symlinks.
264 :param mtime: Define the last modified time for this path. If not provided, debputy will provide a default
265 if the mtime attribute is accessed.
266 :param fs_path: Define a file system path for this path. This causes `has_fs_path` to return True and the
267 `fs_path` attribute will return this value. The test is required to make this path available to the extent
268 required. Note that the virtual file system will *not* examine the provided path in any way nor attempt
269 to resolve defaults from the path.
270 :param link_target: A target for the symlink. Providing a not None value for this parameter will make the
271 path a symlink.
272 :param content: The content of the path (if opened). The path must be a file.
273 :param materialized_content: Same as `content` except `debputy` will put the contents into a physical file
274 as needed. Cannot be used with `content` or `fs_path`.
275 :return: An *opaque* object to be passed to `build_virtual_file_system`. While the exact type is provided
276 to aid with typing, the type name and its behavior is not part of the API.
277 """
279 is_dir = path_name.endswith("/")
280 is_symlink = link_target is not None
282 if is_symlink:
283 if mode is not None:
284 raise ValueError(
285 f'Please do not provide mode for symlinks. Triggered by "{path_name}"'
286 )
287 if is_dir:
288 raise ValueError(
289 "Path name looks like a directory, but a symlink target was also provided."
290 f' Please remove the trailing slash OR the symlink_target. Triggered by "{path_name}"'
291 )
293 if content and (is_dir or is_symlink): 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 raise ValueError(
295 "Content was defined however, the path appears to be a directory a or a symlink"
296 f' Please remove the content, the trailing slash OR the symlink_target. Triggered by "{path_name}"'
297 )
299 if materialized_content is not None:
300 if content is not None: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 raise ValueError(
302 "The materialized_content keyword is mutually exclusive with the content keyword."
303 f' Triggered by "{path_name}"'
304 )
305 if fs_path is not None: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true
306 raise ValueError(
307 "The materialized_content keyword is mutually exclusive with the fs_path keyword."
308 f' Triggered by "{path_name}"'
309 )
310 return PathDef(
311 path_name,
312 mode=mode,
313 mtime=mtime,
314 has_fs_path=bool(fs_path) or materialized_content is not None,
315 fs_path=fs_path,
316 link_target=link_target,
317 content=content,
318 materialized_content=materialized_content,
319 )
322class PackageProcessingContext:
323 """Context for auto-detectors of metadata and package processors (no instantiation)
325 This object holds some context related data for the metadata detector or/and package
326 processors. It may receive new attributes in the future.
327 """
329 __slots__ = ()
331 @property
332 def binary_package(self) -> BinaryPackage:
333 """The binary package stanza from `debian/control`"""
334 raise NotImplementedError
336 @property
337 def binary_package_version(self) -> str:
338 """The version of the binary package
340 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.
341 """
342 raise NotImplementedError
344 @property
345 def related_udeb_package(self) -> Optional[BinaryPackage]:
346 """An udeb related to this binary package (if any)"""
347 raise NotImplementedError
349 @property
350 def related_udeb_package_version(self) -> Optional[str]:
351 """The version of the related udeb package (if present)
353 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.
354 """
355 raise NotImplementedError
357 def accessible_package_roots(self) -> Iterable[Tuple[BinaryPackage, "VirtualPath"]]:
358 raise NotImplementedError
360 # """The source package stanza from `debian/control`"""
361 # source_package: SourcePackage
364class DebputyPluginInitializer:
365 __slots__ = ()
367 def packager_provided_file(
368 self,
369 stem: str,
370 installed_path: str,
371 *,
372 default_mode: int = 0o0644,
373 default_priority: Optional[int] = None,
374 allow_name_segment: bool = True,
375 allow_architecture_segment: bool = False,
376 post_formatting_rewrite: Optional[Callable[[str], str]] = None,
377 packageless_is_fallback_for_all_packages: bool = False,
378 reservation_only: bool = False,
379 reference_documentation: Optional[
380 PackagerProvidedFileReferenceDocumentation
381 ] = None,
382 ) -> None:
383 """Register a packager provided file (debian/<pkg>.foo)
385 Register a packager provided file that debputy should automatically detect and install for the
386 packager (example `debian/foo.tmpfiles` -> `debian/foo/usr/lib/tmpfiles.d/foo.conf`). A packager
387 provided file typically identified by a package prefix and a "stem" and by convention placed
388 in the `debian/` directory.
390 Like debhelper, debputy also supports the `foo.bar.tmpfiles` variant where the file is to be
391 installed into the `foo` package but be named after the `bar` segment rather than the package name.
392 This feature can be controlled via the `allow_name_segment` parameter.
394 :param stem: The "stem" of the file. This would be the `tmpfiles` part of `debian/foo.tmpfiles`.
395 Note that this value must be unique across all registered packager provided files.
396 :param installed_path: A format string describing where the file should be installed. Would be
397 `/usr/lib/tmpfiles.d/{name}.conf` from the example above.
399 The caller should provide a string with one or more of the placeholders listed below (usually `{name}`
400 should be one of them). The format affect the entire path.
402 The following placeholders are supported:
403 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)
404 * `{priority}` / `{priority:02}` - The priority of the file. Only provided priorities are used (that
405 is, default_priority is not None). The latter variant ensuring that the priority takes at least
406 two characters and the `0` character is left-padded for priorities that takes less than two
407 characters.
408 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.
409 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.
411 The path is always interpreted as relative to the binary package root.
413 :param default_mode: The mode the installed file should have by default. Common options are 0o0644 (the default)
414 or 0o0755 (for files that must be executable).
415 :param allow_architecture_segment: If True, the file may have an optional "architecture" segment at the end
416 (`foo.tmpfiles.amd64`), which marks it architecture specific. When False, debputy will detect the
417 "architecture" segment and report the use as an error. Note the architecture segment is only allowed for
418 arch:any packages. If a file targeting an arch:all package uses an architecture specific file it will
419 always result in an error.
420 :param allow_name_segment: If True, the file may have an optional "name" segment after the package name prefix.
421 (`foo.<name-here>.tmpfiles`). When False, debputy will detect the "name" segment and report the use as an
422 error.
423 :param default_priority: Special-case option for packager files that are installed into directories that have
424 "parse ordering" or "priority". These files will generally be installed as something like `20-foo.conf`
425 where the `20-` denotes their "priority". If the plugin is registering such a file type, then it should
426 provide a default priority.
428 The following placeholders are supported:
429 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)
430 * `{priority}` - The priority of the file. Only provided priorities are used (that is, default_priority
431 is not None)
432 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.
433 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.
434 :param post_formatting_rewrite: An optional "name correcting" callback. It receives the formatted name and can
435 do any transformation required. The primary use-case for this is to replace "forbidden" characters. The most
436 common case for debputy itself is to replace "." with "_" for tools that refuse to work with files containing
437 "." (`lambda x: x.replace(".", "_")`). The callback operates on basename of formatted version of the
438 `installed_path` and the callback should return the basename.
439 :param packageless_is_fallback_for_all_packages: If True, the packageless variant (such as, `debian/changelog`)
440 is a fallback for every package.
441 :param reference_documentation: Reference documentation for the packager provided file. Use the
442 packager_provided_file_reference_documentation function to provide the value for this parameter.
443 :param reservation_only: When True, tell debputy that the plugin reserves this packager provided file, but that
444 debputy should not actually install it automatically. This is useful in the cases, where the plugin
445 needs to process the file before installing it. The file will be marked as provided by this plugin. This
446 enables introspection and detects conflicts if other plugins attempts to claim the file.
447 """
448 raise NotImplementedError
450 def metadata_or_maintscript_detector(
451 self,
452 auto_detector_id: str,
453 auto_detector: MetadataAutoDetector,
454 *,
455 package_type: PackageTypeSelector = "deb",
456 ) -> None:
457 """Provide a pre-assembly hook that can affect the metadata/maintscript of binary ("deb") packages
459 The provided hook will be run once per binary package to be assembled, and it can see all the content
460 ("data.tar") planned to be included in the deb. The hook may do any *read-only* analysis of this content
461 and provide metadata, alter substvars or inject maintscript snippets. However, the hook must *not*
462 change the content ("data.tar") part of the deb.
464 The hook will be run unconditionally for all binary packages built. When the hook does not apply to all
465 packages, it must provide its own (internal) logic for detecting whether it is relevant and reduced itself
466 to a no-op if it should not apply to the current package.
468 Hooks are run in "some implementation defined order" and should not rely on being run before or after
469 any other hook.
471 The hooks are only applied to packages defined in `debian/control`. Notably, the metadata detector will
472 not apply to auto-generated `-dbgsym` packages (as those are not listed explicitly in `debian/control`).
474 :param auto_detector_id: A plugin-wide unique ID for this detector. Packagers may use this ID for disabling
475 the detector and accordingly the ID is part of the plugin's API toward the packager.
476 :param auto_detector: The code to be called that will be run at the metadata generation state (once for each
477 binary package).
478 :param package_type: Which kind of packages this metadata detector applies to. The package type is generally
479 defined by `Package-Type` field in the binary package. The default is to only run for regular `deb` packages
480 and ignore `udeb` packages.
481 """
482 raise NotImplementedError
484 def manifest_variable(
485 self,
486 variable_name: str,
487 value: str,
488 variable_reference_documentation: Optional[str] = None,
489 ) -> None:
490 """Provide a variable that can be used in the package manifest
492 >>> # Enable users to use "{{path:BASH_COMPLETION_DIR}}/foo" in their manifest.
493 >>> api.manifest_variable( # doctest: +SKIP
494 ... "path:BASH_COMPLETION_DIR",
495 ... "/usr/share/bash-completion/completions",
496 ... variable_reference_documentation="Directory to install bash completions into",
497 ... )
499 :param variable_name: The variable name.
500 :param value: The value the variable should resolve to.
501 :param variable_reference_documentation: A short snippet of reference documentation that explains
502 the purpose of the variable.
503 """
504 raise NotImplementedError
507class MaintscriptAccessor:
508 __slots__ = ()
510 def on_configure(
511 self,
512 run_snippet: str,
513 /,
514 indent: Optional[bool] = None,
515 perform_substitution: bool = True,
516 skip_on_rollback: bool = False,
517 ) -> None:
518 """Provide a snippet to be run when the package is about to be "configured"
520 This condition is the most common "post install" condition and covers the two
521 common cases:
522 * On initial install, OR
523 * On upgrade
525 In dpkg maintscript terms, this method roughly corresponds to postinst containing
526 `if [ "$1" = configure ]; then <snippet>; fi`
528 Additionally, the condition will by default also include rollback/abort scenarios such as "above-remove",
529 which is normally what you want but most people forget about.
531 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
532 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
533 snippet may contain '{{FOO}}' substitutions by default.
534 :param skip_on_rollback: By default, this condition will also cover common rollback scenarios. This
535 is normally what you want (or benign in most cases due to the idempotence requirement for maintscripts).
536 However, you can disable the rollback cases, leaving only "On initial install OR On upgrade".
537 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
538 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
539 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
540 You are recommended to do 4 spaces of indentation when indent is False for readability.
541 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
542 substitution is provided.
543 """
544 raise NotImplementedError
546 def on_initial_install(
547 self,
548 run_snippet: str,
549 /,
550 indent: Optional[bool] = None,
551 perform_substitution: bool = True,
552 ) -> None:
553 """Provide a snippet to be run when the package is about to be "configured" for the first time
555 The snippet will only be run on the first time the package is installed (ever or since last purge).
556 Note that "first" does not mean "exactly once" as dpkg does *not* provide such semantics. There are two
557 common cases where this can snippet can be run multiple times for the same system (and why the snippet
558 must still be idempotent):
560 1) The package is installed (1), then purged and then installed again (2). This can partly be mitigated
561 by having an `on_purge` script to do clean up.
563 2) As the package is installed, the `postinst` script terminates prematurely (Disk full, power loss, etc.).
564 The user resolves the problem and runs `dpkg --configure <pkg>`, which in turn restarts the script
565 from the beginning. This is why scripts must be idempotent in general.
567 In dpkg maintscript terms, this method roughly corresponds to postinst containing
568 `if [ "$1" = configure ] && [ -z "$2" ]; then <snippet>; fi`
570 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
571 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
572 snippet may contain '{{FOO}}' substitutions by default.
573 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
574 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
575 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
576 You are recommended to do 4 spaces of indentation when indent is False for readability.
577 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
578 substitution is provided.
579 """
580 raise NotImplementedError
582 def on_upgrade(
583 self,
584 run_snippet: str,
585 /,
586 indent: Optional[bool] = None,
587 perform_substitution: bool = True,
588 ) -> None:
589 """Provide a snippet to be run when the package is about to be "configured" after an upgrade
591 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).
593 In dpkg maintscript terms, this method roughly corresponds to postinst containing
594 `if [ "$1" = configure ] && [ -n "$2" ]; then <snippet>; fi`
596 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
597 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
598 snippet may contain '{{FOO}}' substitutions by default.
599 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
600 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
601 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
602 You are recommended to do 4 spaces of indentation when indent is False for readability.
603 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
604 substitution is provided.
605 """
606 raise NotImplementedError
608 def on_upgrade_from(
609 self,
610 version: str,
611 run_snippet: str,
612 /,
613 indent: Optional[bool] = None,
614 perform_substitution: bool = True,
615 ) -> None:
616 """Provide a snippet to be run when the package is about to be "configured" after an upgrade from a given version
618 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).
620 In dpkg maintscript terms, this method roughly corresponds to postinst containing
621 `if [ "$1" = configure ] && dpkg --compare-versions le-nl "$2" ; then <snippet>; fi`
623 :param version: The version to upgrade from
624 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
625 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
626 snippet may contain '{{FOO}}' substitutions by default.
627 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
628 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
629 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
630 You are recommended to do 4 spaces of indentation when indent is False for readability.
631 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
632 substitution is provided.
633 """
634 raise NotImplementedError
636 def on_before_removal(
637 self,
638 run_snippet: str,
639 /,
640 indent: Optional[bool] = None,
641 perform_substitution: bool = True,
642 ) -> None:
643 """Provide a snippet to be run when the package is about to be removed
645 The snippet will be run before dpkg removes any files.
647 In dpkg maintscript terms, this method roughly corresponds to prerm containing
648 `if [ "$1" = remove ] ; then <snippet>; fi`
650 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
651 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
652 snippet may contain '{{FOO}}' substitutions by default.
653 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
654 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
655 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
656 You are recommended to do 4 spaces of indentation when indent is False for readability.
657 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
658 substitution is provided.
659 """
660 raise NotImplementedError
662 def on_removed(
663 self,
664 run_snippet: str,
665 /,
666 indent: Optional[bool] = None,
667 perform_substitution: bool = True,
668 ) -> None:
669 """Provide a snippet to be run when the package has been removed
671 The snippet will be run after dpkg removes the package content from the file system.
673 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.
675 In dpkg maintscript terms, this method roughly corresponds to postrm containing
676 `if [ "$1" = remove ] ; then <snippet>; fi`
678 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
679 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
680 snippet may contain '{{FOO}}' substitutions by default.
681 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
682 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
683 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
684 You are recommended to do 4 spaces of indentation when indent is False for readability.
685 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
686 substitution is provided.
687 """
688 raise NotImplementedError
690 def on_purge(
691 self,
692 run_snippet: str,
693 /,
694 indent: Optional[bool] = None,
695 perform_substitution: bool = True,
696 ) -> None:
697 """Provide a snippet to be run when the package is being purged.
699 The snippet will when the package is purged from the system.
701 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.
703 In dpkg maintscript terms, this method roughly corresponds to postrm containing
704 `if [ "$1" = purge ] ; then <snippet>; fi`
706 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
707 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
708 snippet may contain '{{FOO}}' substitutions by default.
709 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
710 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
711 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
712 You are recommended to do 4 spaces of indentation when indent is False for readability.
713 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
714 substitution is provided.
715 """
716 raise NotImplementedError
718 def unconditionally_in_script(
719 self,
720 maintscript: Maintscript,
721 run_snippet: str,
722 /,
723 perform_substitution: bool = True,
724 ) -> None:
725 """Provide a snippet to be run in a given script
727 Run a given snippet unconditionally from a given script. The snippet must contain its own conditional
728 for when it should be run.
730 :param maintscript: The maintscript to insert the snippet into.
731 :param run_snippet: The actual shell snippet to be run. The snippet will be run unconditionally and should
732 contain its own conditions as necessary. The snippet must be idempotent. The snippet may contain newlines
733 as necessary, which will make the result more readable. Additionally, the snippet may contain '{{FOO}}'
734 substitutions by default.
735 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
736 substitution is provided.
737 """
738 raise NotImplementedError
740 def escape_shell_words(self, *args: str) -> str:
741 """Provide sh-shell escape of strings
743 `assert escape_shell("foo", "fu bar", "baz") == 'foo "fu bar" baz'`
745 This is useful for ensuring file names and other "input" are considered one parameter even when they
746 contain spaces or shell meta-characters.
748 :param args: The string(s) to be escaped.
749 :return: Each argument escaped so that each argument becomes a single "word" and then all these words are
750 joined by a single space.
751 """
752 return util.escape_shell(*args)
755class BinaryCtrlAccessor:
756 __slots__ = ()
758 def dpkg_trigger(self, trigger_type: DpkgTriggerType, trigger_target: str) -> None:
759 """Register a declarative dpkg level trigger
761 The provided trigger will be added to the package's metadata (the triggers file of the control.tar).
763 If the trigger has already been added previously, a second call with the same trigger data will be ignored.
764 """
765 raise NotImplementedError
767 @property
768 def maintscript(self) -> MaintscriptAccessor:
769 """Attribute for manipulating maintscripts"""
770 raise NotImplementedError
772 @property
773 def substvars(self) -> "FlushableSubstvars":
774 """Attribute for manipulating dpkg substvars (deb-substvars)"""
775 raise NotImplementedError
778class VirtualPath:
779 __slots__ = ()
781 @property
782 def name(self) -> str:
783 """Basename of the path a.k.a. last segment of the path
785 In a path "usr/share/doc/pkg/changelog.gz" the basename is "changelog.gz".
787 For a directory, the basename *never* ends with a `/`.
788 """
789 raise NotImplementedError
791 @property
792 def iterdir(self) -> Iterable["VirtualPath"]:
793 """Returns an iterable that iterates over all children of this path
795 For directories, this returns an iterable of all children. For non-directories,
796 the iterable is always empty.
797 """
798 raise NotImplementedError
800 def lookup(self, path: str) -> Optional["VirtualPath"]:
801 """Perform a path lookup relative to this path
803 As an example `doc_dir = fs_root.lookup('./usr/share/doc')`
805 If the provided path starts with `/`, then the lookup is performed relative to the
806 file system root. That is, you can assume the following to always be True:
808 `fs_root.lookup("usr") == any_path_beneath_fs_root.lookup('/usr')`
810 Note: This method requires the path to be attached (see `is_detached`) regardless of
811 whether the lookup is relative or absolute.
813 If the path traverse a symlink, the symlink will be resolved.
815 :param path: The path to look. Can contain "." and ".." segments. If starting with `/`,
816 look up is performed relative to the file system root, otherwise the lookup is relative
817 to this path.
818 :return: The path object for the desired path if it can be found. Otherwise, None.
819 """
820 raise NotImplementedError
822 def all_paths(self) -> Iterable["VirtualPath"]:
823 """Iterate over this path and all of its descendants (if any)
825 If used on the root path, then every path in the package is returned.
827 The iterable is ordered, so using the order in output will be produce
828 bit-for-bit reproducible output. Additionally, a directory will always
829 be seen before its descendants. Otherwise, the order is implementation
830 defined.
832 The iteration is lazy and as a side effect do account for some obvious
833 mutation. Like if the current path is removed, then none of its children
834 will be returned (provided mutation happens before the lazy iteration
835 was required to resolve it). Likewise, mutation of the directory will
836 also work (again, provided mutation happens before the lazy iteration order).
838 :return: An ordered iterable of this path followed by its descendants.
839 """
840 raise NotImplementedError
842 @property
843 def is_detached(self) -> bool:
844 """Returns True if this path is detached
846 Paths that are detached from the file system will not be present in the package and
847 most operations are unsafe on them. This usually only happens if the path or one of
848 its parent directories are unlinked (rm'ed) from the file system tree.
850 All paths are attached by default and will only become detached as a result of
851 an action to mutate the virtual file system. Note that the file system may not
852 always be manipulated.
854 :return: True if the entry is detached. Detached entries should be discarded, so they
855 can be garbage collected.
856 """
857 raise NotImplementedError
859 # The __getitem__ behaves like __getitem__ from Dict but __iter__ would ideally work like a Sequence.
860 # However, that does not feel compatible, so lets force people to use .children instead for the Sequence
861 # behavior to avoid surprises for now.
862 # (Maybe it is a non-issue, but it is easier to add the API later than to remove it once we have committed
863 # to using it)
864 __iter__ = None
866 def __getitem__(self, key: object) -> "VirtualPath":
867 """Lookup a (direct) child by name
869 Ignoring the possible `KeyError`, then the following are the same:
870 `fs_root["usr"] == fs_root.lookup('usr')`
872 Note that unlike `.lookup` this can only locate direct children.
873 """
874 raise NotImplementedError
876 def __delitem__(self, key) -> None:
877 """Remove a child from this node if it exists
879 If that child is a directory, then the entire tree is removed (like `rm -fr`).
880 """
881 raise NotImplementedError
883 def get(self, key: str) -> "Optional[VirtualPath]":
884 """Lookup a (direct) child by name
886 The following are the same:
887 `fs_root.get("usr") == fs_root.lookup('usr')`
889 Note that unlike `.lookup` this can only locate direct children.
890 """
891 try:
892 return self[key]
893 except KeyError:
894 return None
896 def __contains__(self, item: object) -> bool:
897 """Determine if this path includes a given child (either by object or string)
899 Examples:
901 if 'foo' in dir: ...
902 """
903 if isinstance(item, VirtualPath):
904 return item.parent_dir is self
905 if not isinstance(item, str):
906 return False
907 m = self.get(item)
908 return m is not None
910 @property
911 def path(self) -> str:
912 """Returns the "full" path for this file system entry
914 This is the path that debputy uses to refer to this file system entry. It is always
915 normalized. Use the `absolute` attribute for how the path looks
916 when the package is installed. Alternatively, there is also `fs_path`, which is the
917 path to the underlying file system object (assuming there is one). That is the one
918 you need if you want to read the file.
920 This is attribute is mostly useful for debugging or for looking up the path relative
921 to the "root" of the virtual file system that debputy maintains.
923 If the path is detached (see `is_detached`), then this method returns the path as it
924 was known prior to being detached.
925 """
926 raise NotImplementedError
928 @property
929 def absolute(self) -> str:
930 """Returns the absolute version of this path
932 This is how to refer to this path when the package is installed.
934 If the path is detached (see `is_detached`), then this method returns the last known location
935 of installation (prior to being detached).
937 :return: The absolute path of this file as it would be on the installed system.
938 """
939 p = self.path.lstrip(".")
940 if not p.startswith("/"):
941 return f"/{p}"
942 return p
944 @property
945 def parent_dir(self) -> Optional["VirtualPath"]:
946 """The parent directory of this path
948 Note this operation requires the path is "attached" (see `is_detached`). All paths are attached
949 by default but unlinking paths will cause them to become detached.
951 :return: The parent path or None for the root.
952 """
953 raise NotImplementedError
955 def stat(self) -> os.stat_result:
956 """Attempt to do stat of the underlying path (if it exists)
958 *Avoid* using `stat()` whenever possible where a more specialized attribute exist. The
959 `stat()` call returns the data from the file system and often, `debputy` does *not* track
960 its state in the file system. As an example, if you want to know the file system mode of
961 a path, please use the `mode` attribute instead.
963 This never follow symlinks (it behaves like `os.lstat`). It will raise an error
964 if the path is not backed by a file system object (that is, `has_fs_path` is False).
966 :return: The stat result or an error.
967 """
968 raise NotImplementedError()
970 @property
971 def size(self) -> int:
972 """Resolve the file size (`st_size`)
974 This may be using `stat()` and therefore `fs_path`.
976 :return: The size of the file in bytes
977 """
978 return self.stat().st_size
980 @property
981 def mode(self) -> int:
982 """Determine the mode bits of this path object
984 Note that:
985 * like with `stat` above, this never follows symlinks.
986 * the mode returned by this method is not always a 1:1 with the mode in the
987 physical file system. As an optimization, `debputy` skips unnecessary writes
988 to the underlying file system in many cases.
991 :return: The mode bits for the path.
992 """
993 raise NotImplementedError
995 @mode.setter
996 def mode(self, new_mode: int) -> None:
997 """Set the octal file mode of this path
999 Note that:
1000 * this operation will fail if `path.is_read_write` returns False.
1001 * this operation is generally *not* synced to the physical file system (as
1002 an optimization).
1004 :param new_mode: The new octal mode for this path. Note that `debputy` insists
1005 that all paths have the `user read bit` and, for directories also, the
1006 `user execute bit`. The absence of these minimal mode bits causes hard to
1007 debug errors.
1008 """
1009 raise NotImplementedError
1011 @property
1012 def is_executable(self) -> bool:
1013 """Determine whether a path is considered executable
1015 Generally, this means that at least one executable bit is set. This will
1016 basically always be true for directories as directories need the execute
1017 parameter to be traversable.
1019 :return: True if the path is considered executable with its current mode
1020 """
1021 return bool(self.mode & 0o0111)
1023 def chmod(self, new_mode: Union[int, str]) -> None:
1024 """Set the file mode of this path
1026 This is similar to setting the `mode` attribute. However, this method accepts
1027 a string argument, which will be parsed as a symbolic mode (example: `u+rX,go=rX`).
1029 Note that:
1030 * this operation will fail if `path.is_read_write` returns False.
1031 * this operation is generally *not* synced to the physical file system (as
1032 an optimization).
1034 :param new_mode: The new mode for this path.
1035 Note that `debputy` insists that all paths have the `user read bit` and, for
1036 directories also, the `user execute bit`. The absence of these minimal mode
1037 bits causes hard to debug errors.
1038 """
1039 if isinstance(new_mode, str):
1040 segments = parse_symbolic_mode(new_mode, None)
1041 final_mode = self.mode
1042 is_dir = self.is_dir
1043 for segment in segments:
1044 final_mode = segment.apply(final_mode, is_dir)
1045 self.mode = final_mode
1046 else:
1047 self.mode = new_mode
1049 def chown(
1050 self,
1051 owner: Optional["StaticFileSystemOwner"],
1052 group: Optional["StaticFileSystemGroup"],
1053 ) -> None:
1054 """Change the owner/group of this path
1056 :param owner: The desired owner definition for this path. If None, then no change of owner is performed.
1057 :param group: The desired group definition for this path. If None, then no change of group is performed.
1058 """
1059 raise NotImplementedError
1061 @property
1062 def mtime(self) -> float:
1063 """Determine the mtime of this path object
1065 Note that:
1066 * like with `stat` above, this never follows symlinks.
1067 * the mtime returned has *not* been clamped against ´SOURCE_DATE_EPOCH`. Timestamp
1068 normalization is handled later by `debputy`.
1069 * the mtime returned by this method is not always a 1:1 with the mtime in the
1070 physical file system. As an optimization, `debputy` skips unnecessary writes
1071 to the underlying file system in many cases.
1073 :return: The mtime for the path.
1074 """
1075 raise NotImplementedError
1077 @mtime.setter
1078 def mtime(self, new_mtime: float) -> None:
1079 """Set the mtime of this path
1081 Note that:
1082 * this operation will fail if `path.is_read_write` returns False.
1083 * this operation is generally *not* synced to the physical file system (as
1084 an optimization).
1086 :param new_mtime: The new mtime of this path. Note that the caller does not need to
1087 account for `SOURCE_DATE_EPOCH`. Timestamp normalization is handled later.
1088 """
1089 raise NotImplementedError
1091 def readlink(self) -> str:
1092 """Determine the link target of this path assuming it is a symlink
1094 For paths where `is_symlink` is True, this already returns a link target even when
1095 `has_fs_path` is False.
1097 :return: The link target of the path or an error is this is not a symlink
1098 """
1099 raise NotImplementedError()
1101 @overload
1102 def open( 1102 ↛ exitline 1102 didn't jump to the function exit
1103 self,
1104 *,
1105 byte_io: Literal[False] = False,
1106 buffering: int = -1,
1107 ) -> TextIO: ...
1109 @overload
1110 def open( 1110 ↛ exitline 1110 didn't jump to the function exit
1111 self,
1112 *,
1113 byte_io: Literal[True],
1114 buffering: int = -1,
1115 ) -> BinaryIO: ...
1117 @overload
1118 def open( 1118 ↛ exitline 1118 didn't jump to the function exit
1119 self,
1120 *,
1121 byte_io: bool,
1122 buffering: int = -1,
1123 ) -> Union[TextIO, BinaryIO]: ...
1125 def open(
1126 self,
1127 *,
1128 byte_io: bool = False,
1129 buffering: int = -1,
1130 ) -> Union[TextIO, BinaryIO]:
1131 """Open the file for reading. Usually used with a context manager
1133 By default, the file is opened in text mode (utf-8). Binary mode can be requested
1134 via the `byte_io` parameter. This operation is only valid for files (`is_file` returns
1135 `True`). Usage on symlinks and directories will raise exceptions.
1137 This method *often* requires the `fs_path` to be present. However, tests as a notable
1138 case can inject content without having the `fs_path` point to a real file. (To be clear,
1139 such tests are generally expected to ensure `has_fs_path` returns `True`).
1142 :param byte_io: If True, open the file in binary mode (like `rb` for `open`)
1143 :param buffering: Same as open(..., buffering=...) where supported. Notably during
1144 testing, the content may be purely in memory and use a BytesIO/StringIO
1145 (which does not accept that parameter, but then it is buffered in a different way)
1146 :return: The file handle.
1147 """
1149 if not self.is_file: 1149 ↛ 1150line 1149 didn't jump to line 1150 because the condition on line 1149 was never true
1150 raise TypeError(f"Cannot open {self.path} for reading: It is not a file")
1152 if byte_io:
1153 return open(self.fs_path, "rb", buffering=buffering)
1154 return open(self.fs_path, "rt", encoding="utf-8", buffering=buffering)
1156 @property
1157 def fs_path(self) -> str:
1158 """Request the underling fs_path of this path
1160 Only available when `has_fs_path` is True. Generally this should only be used for files to read
1161 the contents of the file and do some action based on the parsed result.
1163 The path should only be used for read-only purposes as debputy may assume that it is safe to have
1164 multiple paths pointing to the same file system path.
1166 Note that:
1167 * This is often *not* available for directories and symlinks.
1168 * The debputy in-memory file system overrules the physical file system. Attempting to "fix" things
1169 by using `os.chmod` or `os.unlink`'ing files, etc. will generally not do as you expect. Best case,
1170 your actions are ignored and worst case it will cause the build to fail as it violates debputy's
1171 internal invariants.
1173 :return: The path to the underlying file system object on the build system or an error if no such
1174 file exist (see `has_fs_path`).
1175 """
1176 raise NotImplementedError()
1178 @property
1179 def is_dir(self) -> bool:
1180 """Determine if this path is a directory
1182 Never follows symlinks.
1184 :return: True if this path is a directory. False otherwise.
1185 """
1186 raise NotImplementedError()
1188 @property
1189 def is_file(self) -> bool:
1190 """Determine if this path is a directory
1192 Never follows symlinks.
1194 :return: True if this path is a regular file. False otherwise.
1195 """
1196 raise NotImplementedError()
1198 @property
1199 def is_symlink(self) -> bool:
1200 """Determine if this path is a symlink
1202 :return: True if this path is a symlink. False otherwise.
1203 """
1204 raise NotImplementedError()
1206 @property
1207 def has_fs_path(self) -> bool:
1208 """Determine whether this path is backed by a file system path
1210 :return: True if this path is backed by a file system object on the build system.
1211 """
1212 raise NotImplementedError()
1214 @property
1215 def is_read_write(self) -> bool:
1216 """When true, the file system entry may be mutated
1218 Read-write rules are:
1220 +--------------------------+-------------------+------------------------+
1221 | File system | From / Inside | Read-Only / Read-Write |
1222 +--------------------------+-------------------+------------------------+
1223 | Source directory | Any context | Read-Only |
1224 | Binary staging directory | Package Processor | Read-Write |
1225 | Binary staging directory | Metadata Detector | Read-Only |
1226 +--------------------------+-------------------+------------------------+
1228 These rules apply to the virtual file system (`debputy` cannot enforce
1229 these rules in the underlying file system). The `debputy` code relies
1230 on these rules for its logic in multiple places to catch bugs and for
1231 optimizations.
1233 As an example, the reason why the file system is read-only when Metadata
1234 Detectors are run is based the contents of the file system has already
1235 been committed. New files will not be included, removals of existing
1236 files will trigger a hard error when the package is assembled, etc.
1237 To avoid people spending hours debugging why their code does not work
1238 as intended, `debputy` instead throws a hard error if you try to mutate
1239 the file system when it is read-only mode to "fail fast".
1241 :return: Whether file system mutations are permitted.
1242 """
1243 return False
1245 def mkdir(self, name: str) -> "VirtualPath":
1246 """Create a new subdirectory of the current path
1248 :param name: Basename of the new directory. The directory must not contain a path
1249 with this basename.
1250 :return: The new subdirectory
1251 """
1252 raise NotImplementedError
1254 def mkdirs(self, path: str) -> "VirtualPath":
1255 """Ensure a given path exists and is a directory.
1257 :param path: Path to the directory to create. Any parent directories will be
1258 created as needed. If the path already exists and is a directory, then it
1259 is returned. If any part of the path exists and that is not a directory,
1260 then the `mkdirs` call will raise an error.
1261 :return: The directory denoted by the given path
1262 """
1263 raise NotImplementedError
1265 def add_file(
1266 self,
1267 name: str,
1268 *,
1269 unlink_if_exists: bool = True,
1270 use_fs_path_mode: bool = False,
1271 mode: int = 0o0644,
1272 mtime: Optional[float] = None,
1273 ) -> ContextManager["VirtualPath"]:
1274 """Add a new regular file as a child of this path
1276 This method will insert a new file into the virtual file system as a child
1277 of the current path (which must be a directory). The caller must use the
1278 return value as a context manager (see example). During the life-cycle of
1279 the managed context, the caller can fill out the contents of the file
1280 from the new path's `fs_path` attribute. The `fs_path` will exist as an
1281 empty file when the context manager is entered.
1283 Once the context manager exits, mutation of the `fs_path` is no longer permitted.
1285 >>> import subprocess
1286 >>> path = ... # doctest: +SKIP
1287 >>> with path.add_file("foo") as new_file, open(new_file.fs_path, "w") as fd: # doctest: +SKIP
1288 ... fd.writelines(["Some", "Content", "Here"])
1290 The caller can replace the provided `fs_path` entirely provided at the end result
1291 (when the context manager exits) is a regular file with no hard links.
1293 Note that this operation will fail if `path.is_read_write` returns False.
1295 :param name: Basename of the new file
1296 :param unlink_if_exists: If the name was already in use, then either an exception is thrown
1297 (when `unlink_if_exists` is False) or the path will be removed via ´unlink(recursive=False)`
1298 (when `unlink_if_exists` is True)
1299 :param use_fs_path_mode: When True, the file created will have this mode in the physical file
1300 system. When the context manager exists, `debputy` will refresh its mode to match the mode
1301 in the physical file system. This is primarily useful if the caller uses a subprocess to
1302 mutate the path and the file mode is relevant for this tool (either as input or output).
1303 When the parameter is false, the new file is guaranteed to be readable and writable for
1304 the current user. However, no other guarantees are given (not even that it matches the
1305 `mode` parameter and any changes to the mode in the physical file system will be ignored.
1306 :param mode: This is the initial file mode. Note the `use_fs_path_mode` parameter for how
1307 this interacts with the physical file system.
1308 :param mtime: If the caller has a more accurate mtime than the mtime of the generated file,
1309 then it can be provided here. Note that all mtimes will later be clamped based on
1310 `SOURCE_DATE_EPOCH`. This parameter is only for when the conceptual mtime of this path
1311 should be earlier than `SOURCE_DATE_EPOCH`.
1312 :return: A Context manager that upon entering provides a `VirtualPath` instance for the
1313 new file. The instance remains valid after the context manager exits (assuming it exits
1314 successfully), but the file denoted by `fs_path` must not be changed after the context
1315 manager exits
1316 """
1317 raise NotImplementedError
1319 def replace_fs_path_content(
1320 self,
1321 *,
1322 use_fs_path_mode: bool = False,
1323 ) -> ContextManager[str]:
1324 """Replace the contents of this file via inline manipulation
1326 Used as a context manager to provide the fs path for manipulation.
1328 Example:
1329 >>> import subprocess
1330 >>> path = ... # doctest: +SKIP
1331 >>> with path.replace_fs_path_content() as fs_path: # doctest: +SKIP
1332 ... subprocess.check_call(['strip', fs_path]) # doctest: +SKIP
1334 The provided file system path should be manipulated inline. The debputy framework may
1335 copy it first as necessary and therefore the provided fs_path may be different from
1336 `path.fs_path` prior to entering the context manager.
1338 Note that this operation will fail if `path.is_read_write` returns False.
1340 If the mutation causes the returned `fs_path` to be a non-file or a hard-linked file
1341 when the context manager exits, `debputy` will raise an error at that point. To preserve
1342 the internal invariants of `debputy`, the path will be unlinked as `debputy` cannot
1343 reliably restore the path.
1345 :param use_fs_path_mode: If True, any changes to the mode on the physical FS path will be
1346 recorded as the desired mode of the file when the contextmanager ends. The provided FS path
1347 with start with the current mode when `use_fs_path_mode` is True. Otherwise, `debputy` will
1348 ignore the mode of the file system entry and reuse its own current mode
1349 definition.
1350 :return: A Context manager that upon entering provides the path to a muable (copy) of
1351 this path's `fs_path` attribute. The file on the underlying path may be mutated however
1352 the caller wishes until the context manager exits.
1353 """
1354 raise NotImplementedError
1356 def add_symlink(self, link_name: str, link_target: str) -> "VirtualPath":
1357 """Add a new regular file as a child of this path
1359 This will create a new symlink inside the current path. If the path already exists,
1360 the existing path will be unlinked via `unlink(recursive=False)`.
1362 Note that this operation will fail if `path.is_read_write` returns False.
1364 :param link_name: The basename of the link file entry.
1365 :param link_target: The target of the link. Link target normalization will
1366 be handled by `debputy`, so the caller can use relative or absolute paths.
1367 (At the time of writing, symlink target normalization happens late)
1368 :return: The newly created symlink.
1369 """
1370 raise NotImplementedError
1372 def unlink(self, *, recursive: bool = False) -> None:
1373 """Unlink a file or a directory
1375 This operation will remove the path from the file system (causing `is_detached` to return True).
1377 When the path is a:
1379 * symlink, then the symlink itself is removed. The target (if present) is not affected.
1380 * *non-empty* directory, then the `recursive` parameter decides the outcome. An empty
1381 directory will be removed regardless of the value of `recursive`.
1383 Note that:
1384 * the root directory cannot be deleted.
1385 * this operation will fail if `path.is_read_write` returns False.
1387 :param recursive: If True, then non-empty directories will be unlinked as well removing everything inside them
1388 as well. When False, an error is raised if the path is a non-empty directory
1389 """
1390 raise NotImplementedError
1392 def interpreter(self) -> Optional[Interpreter]:
1393 """Determine the interpreter of the file (`#!`-line details)
1395 Note: this method is only applicable for files (`is_file` is True).
1397 :return: The detected interpreter if present or None if no interpreter can be detected.
1398 """
1399 if not self.is_file:
1400 raise TypeError("Only files can have interpreters")
1401 try:
1402 with self.open(byte_io=True, buffering=4096) as fd:
1403 return extract_shebang_interpreter_from_file(fd)
1404 except (PureVirtualPathError, TestPathWithNonExistentFSPathError):
1405 return None
1407 def metadata(
1408 self,
1409 metadata_type: Type[PMT],
1410 ) -> PathMetadataReference[PMT]:
1411 """Fetch the path metadata reference to access the underlying metadata
1413 Calling this method returns a reference to an arbitrary piece of metadata associated
1414 with this path. Plugins can store any arbitrary data associated with a given path.
1415 Keep in mind that the metadata is stored in memory, so keep the size in moderation.
1417 To store / update the metadata, the path must be in read-write mode. However,
1418 already stored metadata remains accessible even if the path becomes read-only.
1420 Note this method is not applicable if the path is detached
1422 :param metadata_type: Type of the metadata being stored.
1423 :return: A reference to the metadata.
1424 """
1425 raise NotImplementedError
1428class FlushableSubstvars(Substvars):
1429 __slots__ = ()
1431 @contextlib.contextmanager
1432 def flush(self) -> Iterator[str]:
1433 """Temporarily write the substvars to a file and then re-read it again
1435 >>> s = FlushableSubstvars()
1436 >>> 'Test:Var' in s
1437 False
1438 >>> with s.flush() as name, open(name, 'wt', encoding='utf-8') as fobj:
1439 ... _ = fobj.write('Test:Var=bar\\n') # "_ = " is to ignore the return value of write
1440 >>> 'Test:Var' in s
1441 True
1443 Used as a context manager to define when the file is flushed and can be
1444 accessed via the file system. If the context terminates successfully, the
1445 file is read and its content replaces the current substvars.
1447 This is mostly useful if the plugin needs to interface with a third-party
1448 tool that requires a file as interprocess communication (IPC) for sharing
1449 the substvars.
1451 The file may be truncated or completed replaced (change inode) as long as
1452 the provided path points to a regular file when the context manager
1453 terminates successfully.
1455 Note that any manipulation of the substvars via the `Substvars` API while
1456 the file is flushed will silently be discarded if the context manager completes
1457 successfully.
1458 """
1459 with tempfile.NamedTemporaryFile(mode="w+t", encoding="utf-8") as tmp:
1460 self.write_substvars(tmp)
1461 tmp.flush() # Temping to use close, but then we have to manually delete the file.
1462 yield tmp.name
1463 # Re-open; seek did not work when I last tried (if I did it work, feel free to
1464 # convert back to seek - as long as it works!)
1465 with open(tmp.name, "rt", encoding="utf-8") as fd:
1466 self.read_substvars(fd)
1468 def save(self) -> None:
1469 # Promote the debputy extension over `save()` for the plugins.
1470 if self._substvars_path is None:
1471 raise TypeError(
1472 "Please use `flush()` extension to temporarily write the substvars to the file system"
1473 )
1474 super().save()
1477class ServiceRegistry(Generic[DSD]):
1478 __slots__ = ()
1480 def register_service(
1481 self,
1482 path: VirtualPath,
1483 name: Union[str, List[str]],
1484 *,
1485 type_of_service: str = "service", # "timer", etc.
1486 service_scope: str = "system",
1487 enable_by_default: bool = True,
1488 start_by_default: bool = True,
1489 default_upgrade_rule: ServiceUpgradeRule = "restart",
1490 service_context: Optional[DSD] = None,
1491 ) -> None:
1492 """Register a service detected in the package
1494 All the details will either be provided as-is or used as default when the plugin provided
1495 integration code is called.
1497 Two services from different service managers are considered related when:
1499 1) They are of the same type (`type_of_service`) and has the same scope (`service_scope`), AND
1500 2) Their plugin provided names has an overlap
1502 Related services can be covered by the same service definition in the manifest.
1504 :param path: The path defining this service.
1505 :param name: The name of the service. Multiple ones can be provided if the service has aliases.
1506 Note that when providing multiple names, `debputy` will use the first name in the list as the
1507 default name if it has to choose. Any alternative name provided can be used by the packager
1508 to identify this service.
1509 :param type_of_service: The type of service. By default, this is "service", but plugins can
1510 provide other types (such as "timer" for the systemd timer unit).
1511 :param service_scope: The scope for this service. By default, this is "system" meaning the
1512 service is a system-wide service. Service managers can define their own scopes such as
1513 "user" (which is used by systemd for "per-user" services).
1514 :param enable_by_default: Whether the service should be enabled by default, assuming the
1515 packager does not explicitly override this setting.
1516 :param start_by_default: Whether the service should be started by default on install, assuming
1517 the packager does not explicitly override this setting.
1518 :param default_upgrade_rule: The default value for how the service should be processed during
1519 upgrades. Options are:
1520 * `do-nothing`: The plugin should not interact with the running service (if any)
1521 (maintenance of the enabled start, start on install, etc. are still applicable)
1522 * `reload`: The plugin should attempt to reload the running service (if any).
1523 Note: In combination with `auto_start_in_install == False`, be careful to not
1524 start the service if not is not already running.
1525 * `restart`: The plugin should attempt to restart the running service (if any).
1526 Note: In combination with `auto_start_in_install == False`, be careful to not
1527 start the service if not is not already running.
1528 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`
1529 and start it against in the `postinst` script.
1531 :param service_context: Any custom data that the detector want to pass along to the
1532 integrator for this service.
1533 """
1534 raise NotImplementedError
1537@dataclasses.dataclass(slots=True, frozen=True)
1538class ParserAttributeDocumentation:
1539 attributes: FrozenSet[str]
1540 description: Optional[str]
1542 @property
1543 def is_hidden(self) -> bool:
1544 return False
1547@final
1548@dataclasses.dataclass(slots=True, frozen=True)
1549class StandardParserAttributeDocumentation(ParserAttributeDocumentation):
1550 sort_category: int = 0
1553def undocumented_attr(attr: str) -> ParserAttributeDocumentation:
1554 """Describe an attribute as undocumented
1556 If you for some reason do not want to document a particular attribute, you can mark it as
1557 undocumented. This is required if you are only documenting a subset of the attributes,
1558 because `debputy` assumes any omission to be a mistake.
1560 :param attr: Name of the attribute
1561 """
1562 return ParserAttributeDocumentation(
1563 frozenset({attr}),
1564 None,
1565 )
1568@dataclasses.dataclass(slots=True, frozen=True)
1569class ParserDocumentation:
1570 synopsis: Optional[str] = None
1571 title: Optional[str] = None
1572 description: Optional[str] = None
1573 attribute_doc: Optional[Sequence[ParserAttributeDocumentation]] = None
1574 alt_parser_description: Optional[str] = None
1575 documentation_reference_url: Optional[str] = None
1577 def replace(self, **changes: Any) -> "ParserDocumentation":
1578 return dataclasses.replace(self, **changes)
1580 @classmethod
1581 def from_ref_doc(cls, ref_doc: "ParserRefDocumentation") -> "ParserDocumentation":
1582 attr = [
1583 documented_attr(d["attr"], d["description"])
1584 for d in ref_doc.get("attributes", [])
1585 ]
1586 undoc_attr = ref_doc.get("undocumented_attributes")
1587 if undoc_attr: 1587 ↛ 1588line 1587 didn't jump to line 1588 because the condition on line 1587 was never true
1588 attr.extend(undocumented_attr(attr) for attr in undoc_attr)
1590 return reference_documentation(
1591 title=ref_doc["title"],
1592 description=ref_doc["description"],
1593 attributes=attr,
1594 non_mapping_description=ref_doc.get("non_mapping_description"),
1595 reference_documentation_url=ref_doc.get("ref_doc_url"),
1596 synopsis=ref_doc.get("synopsis"),
1597 )
1600@dataclasses.dataclass(slots=True, frozen=True)
1601class TypeMappingExample(Generic[S]):
1602 source_input: S
1605@dataclasses.dataclass(slots=True, frozen=True)
1606class TypeMappingDocumentation(Generic[S]):
1607 description: Optional[str] = None
1608 examples: Sequence[TypeMappingExample[S]] = tuple()
1611def type_mapping_example(source_input: S) -> TypeMappingExample[S]:
1612 return TypeMappingExample(source_input)
1615def type_mapping_reference_documentation(
1616 *,
1617 description: Optional[str] = None,
1618 examples: Union[TypeMappingExample[S], Iterable[TypeMappingExample[S]]] = tuple(),
1619) -> TypeMappingDocumentation[S]:
1620 e = (
1621 tuple([examples])
1622 if isinstance(examples, TypeMappingExample)
1623 else tuple(examples)
1624 )
1625 return TypeMappingDocumentation(
1626 description=description,
1627 examples=e,
1628 )
1631def documented_attr(
1632 attr: Union[str, Iterable[str]],
1633 description: str,
1634) -> ParserAttributeDocumentation:
1635 """Describe an attribute or a group of attributes
1637 :param attr: A single attribute or a sequence of attributes. The attribute must be the
1638 attribute name as used in the source format version of the TypedDict.
1640 If multiple attributes are provided, they will be documented together. This is often
1641 useful if these attributes are strongly related (such as different names for the same
1642 target attribute).
1643 :param description: The description the user should see for this attribute / these
1644 attributes. This parameter can be a Python format string with variables listed in
1645 the description of `reference_documentation`.
1646 :return: An opaque representation of the documentation,
1647 """
1648 attributes = [attr] if isinstance(attr, str) else attr
1649 return ParserAttributeDocumentation(
1650 frozenset(attributes),
1651 description,
1652 )
1655def reference_documentation(
1656 title: str = "Auto-generated reference documentation for ${RULE_NAME}",
1657 description: Optional[str] = textwrap.dedent(
1658 """\
1659 This is an automatically generated reference documentation for ${RULE_NAME}. It is generated
1660 from input provided by ${PLUGIN_NAME} via the debputy API.
1662 (If you are the provider of the ${PLUGIN_NAME} plugin, you can replace this text with
1663 your own documentation by providing the `inline_reference_documentation` when registering
1664 the manifest rule.)
1665 """
1666 ),
1667 attributes: Optional[Sequence[ParserAttributeDocumentation]] = None,
1668 non_mapping_description: Optional[str] = None,
1669 reference_documentation_url: Optional[str] = None,
1670 synopsis: Optional[str] = None,
1671) -> ParserDocumentation:
1672 """Provide inline reference documentation for the manifest snippet
1674 For parameters that mention that they are a Python format, the following template variables
1675 are available (`${FOO}`):
1677 * RULE_NAME: Name of the rule. If manifest snippet has aliases, this will be the name of
1678 the alias provided by the user.
1679 * MANIFEST_FORMAT_DOC: Path OR URL to the "MANIFEST-FORMAT" reference documentation from
1680 `debputy`. By using the MANIFEST_FORMAT_DOC variable, you ensure that you point to the
1681 file that matches the version of `debputy` itself.
1682 * PLUGIN_NAME: Name of the plugin providing this rule.
1684 :param title: The text you want the user to see as for your rule. A placeholder is provided by default.
1685 This parameter can be a Python format string with the above listed variables.
1686 :param description: The text you want the user to see as a description for the rule. An auto-generated
1687 placeholder is provided by default saying that no human written documentation was provided.
1688 This parameter can be a Python format string with the above listed variables.
1689 :param synopsis: One-line plain-text description used for describing the feature during completion.
1690 :param attributes: A sequence of attribute-related documentation. Each element of the sequence should
1691 be the result of `documented_attr` or `undocumented_attr`. The sequence must cover all source
1692 attributes exactly once.
1693 :param non_mapping_description: The text you want the user to see as the description for your rule when
1694 `debputy` describes its non-mapping format. Must not be provided for rules that do not have an
1695 (optional) non-mapping format as source format. This parameter can be a Python format string with
1696 the above listed variables.
1697 :param reference_documentation_url: A URL to the reference documentation.
1698 :return: An opaque representation of the documentation,
1699 """
1700 return ParserDocumentation(
1701 synopsis,
1702 title,
1703 description,
1704 attributes,
1705 non_mapping_description,
1706 reference_documentation_url,
1707 )
1710class ServiceDefinition(Generic[DSD]):
1711 __slots__ = ()
1713 @property
1714 def name(self) -> str:
1715 """Name of the service registered by the plugin
1717 This is always a plugin provided name for this service (that is, `x.name in x.names`
1718 will always be `True`). Where possible, this will be the same as the one that the
1719 packager provided when they provided any configuration related to this service.
1720 When not possible, this will be the first name provided by the plugin (`x.names[0]`).
1722 If all the aliases are equal, then using this attribute will provide traceability
1723 between the manifest and the generated maintscript snippets. When the exact name
1724 used is important, the plugin should ignore this attribute and pick the name that
1725 is needed.
1726 """
1727 raise NotImplementedError
1729 @property
1730 def names(self) -> Sequence[str]:
1731 """All *plugin provided* names and aliases of the service
1733 This is the name/sequence of names that the plugin provided when it registered
1734 the service earlier.
1735 """
1736 raise NotImplementedError
1738 @property
1739 def path(self) -> VirtualPath:
1740 """The registered path for this service
1742 :return: The path that was associated with this service when it was registered
1743 earlier.
1744 """
1745 raise NotImplementedError
1747 @property
1748 def type_of_service(self) -> str:
1749 """Type of the service such as "service" (daemon), "timer", etc.
1751 :return: The type of service scope. It is the same value as the one as the plugin provided
1752 when registering the service (if not explicitly provided, it defaults to "service").
1753 """
1754 raise NotImplementedError
1756 @property
1757 def service_scope(self) -> str:
1758 """Service scope such as "system" or "user"
1760 :return: The service scope. It is the same value as the one as the plugin provided
1761 when registering the service (if not explicitly provided, it defaults to "system")
1762 """
1763 raise NotImplementedError
1765 @property
1766 def auto_enable_on_install(self) -> bool:
1767 """Whether the service should be auto-enabled on install
1769 :return: True if the service should be enabled automatically, false if not.
1770 """
1771 raise NotImplementedError
1773 @property
1774 def auto_start_on_install(self) -> bool:
1775 """Whether the service should be auto-started on install
1777 :return: True if the service should be started automatically, false if not.
1778 """
1779 raise NotImplementedError
1781 @property
1782 def on_upgrade(self) -> ServiceUpgradeRule:
1783 """How to handle the service during an upgrade
1785 Options are:
1786 * `do-nothing`: The plugin should not interact with the running service (if any)
1787 (maintenance of the enabled start, start on install, etc. are still applicable)
1788 * `reload`: The plugin should attempt to reload the running service (if any).
1789 Note: In combination with `auto_start_in_install == False`, be careful to not
1790 start the service if not is not already running.
1791 * `restart`: The plugin should attempt to restart the running service (if any).
1792 Note: In combination with `auto_start_in_install == False`, be careful to not
1793 start the service if not is not already running.
1794 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`
1795 and start it against in the `postinst` script.
1797 Note: In all cases, the plugin should still consider what to do in
1798 `prerm remove`, which is the last point in time where the plugin can rely on the
1799 service definitions in the file systems to stop the services when the package is
1800 being uninstalled.
1802 :return: The service restart rule
1803 """
1804 raise NotImplementedError
1806 @property
1807 def definition_source(self) -> str:
1808 """Describes where this definition came from
1810 If the definition is provided by the packager, then this will reference the part
1811 of the manifest that made this definition. Otherwise, this will be a reference
1812 to the plugin providing this definition.
1814 :return: The source of this definition
1815 """
1816 raise NotImplementedError
1818 @property
1819 def is_plugin_provided_definition(self) -> bool:
1820 """Whether the definition source points to the plugin or a package provided definition
1822 :return: True if definition is 100% from the plugin. False if the definition is partially
1823 or fully from another source (usually, the packager via the manifest).
1824 """
1825 raise NotImplementedError
1827 @property
1828 def service_context(self) -> Optional[DSD]:
1829 """Custom service context (if any) provided by the detector code of the plugin
1831 :return: If the detection code provided a custom data when registering the
1832 service, this attribute will reference that data. If nothing was provided,
1833 then this attribute will be None.
1834 """
1835 raise NotImplementedError