Coverage for src/debputy/plugin/api/spec.py: 87%
282 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-07 12:14 +0200
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)
28from debian.substvars import Substvars
30from debputy import util
31from debputy.exceptions import TestPathWithNonExistentFSPathError, PureVirtualPathError
32from debputy.interpreter import Interpreter, extract_shebang_interpreter_from_file
33from debputy.manifest_parser.util import parse_symbolic_mode
34from debputy.packages import BinaryPackage
35from debputy.types import S
37if TYPE_CHECKING:
38 from debputy.manifest_parser.base_types import (
39 StaticFileSystemOwner,
40 StaticFileSystemGroup,
41 )
44PluginInitializationEntryPoint = Callable[["DebputyPluginInitializer"], None]
45MetadataAutoDetector = Callable[
46 ["VirtualPath", "BinaryCtrlAccessor", "PackageProcessingContext"], None
47]
48PackageProcessor = Callable[["VirtualPath", None, "PackageProcessingContext"], None]
49DpkgTriggerType = Literal[
50 "activate",
51 "activate-await",
52 "activate-noawait",
53 "interest",
54 "interest-await",
55 "interest-noawait",
56]
57Maintscript = Literal["postinst", "preinst", "prerm", "postrm"]
58PackageTypeSelector = Union[Literal["deb", "udeb"], Iterable[Literal["deb", "udeb"]]]
59ServiceUpgradeRule = Literal[
60 "do-nothing",
61 "reload",
62 "restart",
63 "stop-then-start",
64]
66DSD = TypeVar("DSD")
67ServiceDetector = Callable[
68 ["VirtualPath", "ServiceRegistry[DSD]", "PackageProcessingContext"],
69 None,
70]
71ServiceIntegrator = Callable[
72 [
73 Sequence["ServiceDefinition[DSD]"],
74 "BinaryCtrlAccessor",
75 "PackageProcessingContext",
76 ],
77 None,
78]
80PMT = TypeVar("PMT")
83@dataclasses.dataclass(slots=True, frozen=True)
84class PackagerProvidedFileReferenceDocumentation:
85 description: Optional[str] = None
86 format_documentation_uris: Sequence[str] = tuple()
88 def replace(self, **changes: Any) -> "PackagerProvidedFileReferenceDocumentation":
89 return dataclasses.replace(self, **changes)
92def packager_provided_file_reference_documentation(
93 *,
94 description: Optional[str] = None,
95 format_documentation_uris: Optional[Sequence[str]] = tuple(),
96) -> PackagerProvidedFileReferenceDocumentation:
97 """Provide documentation for a given packager provided file.
99 :param description: Textual description presented to the user.
100 :param format_documentation_uris: A sequence of URIs to documentation that describes
101 the format of the file. Most relevant first.
102 :return:
103 """
104 uris = tuple(format_documentation_uris) if format_documentation_uris else tuple()
105 return PackagerProvidedFileReferenceDocumentation(
106 description=description,
107 format_documentation_uris=uris,
108 )
111class PathMetadataReference(Generic[PMT]):
112 """An accessor to plugin provided metadata
114 This is a *short-lived* reference to a piece of metadata. It should *not* be stored beyond
115 the boundaries of the current plugin execution context as it can be become invalid (as an
116 example, if the path associated with this path is removed, then this reference become invalid)
117 """
119 @property
120 def is_present(self) -> bool:
121 """Determine whether the value has been set
123 If the current plugin cannot access the value, then this method unconditionally returns
124 `False` regardless of whether the value is there.
126 :return: `True` if the value has been set to a not None value (and not been deleted).
127 Otherwise, this property is `False`.
128 """
129 raise NotImplementedError
131 @property
132 def can_read(self) -> bool:
133 """Test whether it is possible to read the metadata
135 Note: That the metadata being readable does *not* imply that the metadata is present.
137 :return: True if it is possible to read the metadata. This is always True for the
138 owning plugin.
139 """
140 raise NotImplementedError
142 @property
143 def can_write(self) -> bool:
144 """Test whether it is possible to update the metadata
146 :return: True if it is possible to update the metadata.
147 """
148 raise NotImplementedError
150 @property
151 def value(self) -> Optional[PMT]:
152 """Fetch the currently stored value if present.
154 :return: The value previously stored if any. Returns `None` if the value was never
155 stored, explicitly set to `None` or was deleted.
156 """
157 raise NotImplementedError
159 @value.setter
160 def value(self, value: Optional[PMT]) -> None:
161 """Replace any current value with the provided value
163 This operation is only possible if the path is writable *and* the caller is from
164 the owning plugin OR the owning plugin made the reference read-write.
165 """
166 raise NotImplementedError
168 @value.deleter
169 def value(self) -> None:
170 """Delete any current value.
172 This has the same effect as setting the value to `None`. It has the same restrictions
173 as the value setter.
174 """
175 self.value = None
178@dataclasses.dataclass(slots=True)
179class PathDef:
180 path_name: str
181 mode: Optional[int] = None
182 mtime: Optional[int] = None
183 has_fs_path: Optional[bool] = None
184 fs_path: Optional[str] = None
185 link_target: Optional[str] = None
186 content: Optional[str] = None
187 materialized_content: Optional[str] = None
190def virtual_path_def(
191 path_name: str,
192 /,
193 mode: Optional[int] = None,
194 mtime: Optional[int] = None,
195 fs_path: Optional[str] = None,
196 link_target: Optional[str] = None,
197 content: Optional[str] = None,
198 materialized_content: Optional[str] = None,
199) -> PathDef:
200 """Define a virtual path for use with examples or, in tests, `build_virtual_file_system`
202 :param path_name: The full path. Must start with "./". If it ends with "/", the path will be interpreted
203 as a directory (the `is_dir` attribute will be True). Otherwise, it will be a symlink or file depending
204 on whether a `link_target` is provided.
205 :param mode: The mode to use for this path. Defaults to 0644 for files and 0755 for directories. The mode
206 should be None for symlinks.
207 :param mtime: Define the last modified time for this path. If not provided, debputy will provide a default
208 if the mtime attribute is accessed.
209 :param fs_path: Define a file system path for this path. This causes `has_fs_path` to return True and the
210 `fs_path` attribute will return this value. The test is required to make this path available to the extent
211 required. Note that the virtual file system will *not* examine the provided path in any way nor attempt
212 to resolve defaults from the path.
213 :param link_target: A target for the symlink. Providing a not None value for this parameter will make the
214 path a symlink.
215 :param content: The content of the path (if opened). The path must be a file.
216 :param materialized_content: Same as `content` except `debputy` will put the contents into a physical file
217 as needed. Cannot be used with `content` or `fs_path`.
218 :return: An *opaque* object to be passed to `build_virtual_file_system`. While the exact type is provided
219 to aid with typing, the type name and its behaviour is not part of the API.
220 """
222 is_dir = path_name.endswith("/")
223 is_symlink = link_target is not None
225 if is_symlink:
226 if mode is not None:
227 raise ValueError(
228 f'Please do not provide mode for symlinks. Triggered by "{path_name}"'
229 )
230 if is_dir:
231 raise ValueError(
232 "Path name looks like a directory, but a symlink target was also provided."
233 f' Please remove the trailing slash OR the symlink_target. Triggered by "{path_name}"'
234 )
236 if content and (is_dir or is_symlink): 236 ↛ 237line 236 didn't jump to line 237, because the condition on line 236 was never true
237 raise ValueError(
238 "Content was defined however, the path appears to be a directory a or a symlink"
239 f' Please remove the content, the trailing slash OR the symlink_target. Triggered by "{path_name}"'
240 )
242 if materialized_content is not None:
243 if content is not None: 243 ↛ 244line 243 didn't jump to line 244, because the condition on line 243 was never true
244 raise ValueError(
245 "The materialized_content keyword is mutually exclusive with the content keyword."
246 f' Triggered by "{path_name}"'
247 )
248 if fs_path is not None: 248 ↛ 249line 248 didn't jump to line 249, because the condition on line 248 was never true
249 raise ValueError(
250 "The materialized_content keyword is mutually exclusive with the fs_path keyword."
251 f' Triggered by "{path_name}"'
252 )
253 return PathDef(
254 path_name,
255 mode=mode,
256 mtime=mtime,
257 has_fs_path=bool(fs_path) or materialized_content is not None,
258 fs_path=fs_path,
259 link_target=link_target,
260 content=content,
261 materialized_content=materialized_content,
262 )
265class PackageProcessingContext:
266 """Context for auto-detectors of metadata and package processors (no instantiation)
268 This object holds some context related data for the metadata detector or/and package
269 processors. It may receive new attributes in the future.
270 """
272 __slots__ = ()
274 @property
275 def binary_package(self) -> BinaryPackage:
276 """The binary package stanza from `debian/control`"""
277 raise NotImplementedError
279 @property
280 def binary_package_version(self) -> str:
281 """The version of the binary package
283 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.
284 """
285 raise NotImplementedError
287 @property
288 def related_udeb_package(self) -> Optional[BinaryPackage]:
289 """An udeb related to this binary package (if any)"""
290 raise NotImplementedError
292 @property
293 def related_udeb_package_version(self) -> Optional[str]:
294 """The version of the related udeb package (if present)
296 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.
297 """
298 raise NotImplementedError
300 def accessible_package_roots(self) -> Iterable[Tuple[BinaryPackage, "VirtualPath"]]:
301 raise NotImplementedError
303 # """The source package stanza from `debian/control`"""
304 # source_package: SourcePackage
307class DebputyPluginInitializer:
308 __slots__ = ()
310 def packager_provided_file(
311 self,
312 stem: str,
313 installed_path: str,
314 *,
315 default_mode: int = 0o0644,
316 default_priority: Optional[int] = None,
317 allow_name_segment: bool = True,
318 allow_architecture_segment: bool = False,
319 post_formatting_rewrite: Optional[Callable[[str], str]] = None,
320 packageless_is_fallback_for_all_packages: bool = False,
321 reservation_only: bool = False,
322 reference_documentation: Optional[
323 PackagerProvidedFileReferenceDocumentation
324 ] = None,
325 ) -> None:
326 """Register a packager provided file (debian/<pkg>.foo)
328 Register a packager provided file that debputy should automatically detect and install for the
329 packager (example `debian/foo.tmpfiles` -> `debian/foo/usr/lib/tmpfiles.d/foo.conf`). A packager
330 provided file typically identified by a package prefix and a "stem" and by convention placed
331 in the `debian/` directory.
333 Like debhelper, debputy also supports the `foo.bar.tmpfiles` variant where the file is to be
334 installed into the `foo` package but be named after the `bar` segment rather than the package name.
335 This feature can be controlled via the `allow_name_segment` parameter.
337 :param stem: The "stem" of the file. This would be the `tmpfiles` part of `debian/foo.tmpfiles`.
338 Note that this value must be unique across all registered packager provided files.
339 :param installed_path: A format string describing where the file should be installed. Would be
340 `/usr/lib/tmpfiles.d/{name}.conf` from the example above.
342 The caller should provide a string with one or more of the placeholders listed below (usually `{name}`
343 should be one of them). The format affect the entire path.
345 The following placeholders are supported:
346 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)
347 * `{priority}` / `{priority:02}` - The priority of the file. Only provided priorities are used (that
348 is, default_priority is not None). The latter variant ensuring that the priority takes at least
349 two characters and the `0` character is left-padded for priorities that takes less than two
350 characters.
351 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.
352 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.
354 The path is always interpreted as relative to the binary package root.
356 :param default_mode: The mode the installed file should have by default. Common options are 0o0644 (the default)
357 or 0o0755 (for files that must be executable).
358 :param allow_architecture_segment: If True, the file may have an optional "architecture" segment at the end
359 (`foo.tmpfiles.amd64`), which marks it architecture specific. When False, debputy will detect the
360 "architecture" segment and report the use as an error. Note the architecture segment is only allowed for
361 arch:any packages. If a file targeting an arch:all package uses an architecture specific file it will
362 always result in an error.
363 :param allow_name_segment: If True, the file may have an optional "name" segment after the package name prefix.
364 (`foo.<name-here>.tmpfiles`). When False, debputy will detect the "name" segment and report the use as an
365 error.
366 :param default_priority: Special-case option for packager files that are installed into directories that have
367 "parse ordering" or "priority". These files will generally be installed as something like `20-foo.conf`
368 where the `20-` denotes their "priority". If the plugin is registering such a file type, then it should
369 provide a default priority.
371 The following placeholders are supported:
372 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)
373 * `{priority}` - The priority of the file. Only provided priorities are used (that is, default_priority
374 is not None)
375 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.
376 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.
377 :param post_formatting_rewrite: An optional "name correcting" callback. It receives the formatted name and can
378 do any transformation required. The primary use-case for this is to replace "forbidden" characters. The most
379 common case for debputy itself is to replace "." with "_" for tools that refuse to work with files containing
380 "." (`lambda x: x.replace(".", "_")`). The callback operates on basename of formatted version of the
381 `installed_path` and the callback should return the basename.
382 :param packageless_is_fallback_for_all_packages: If True, the packageless variant (such as, `debian/changelog`)
383 is a fallback for every package.
384 :param reference_documentation: Reference documentation for the packager provided file. Use the
385 packager_provided_file_reference_documentation function to provide the value for this parameter.
386 :param reservation_only: When True, tell debputy that the plugin reserves this packager provided file, but that
387 debputy should not actually install it automatically. This is useful in the cases, where the plugin
388 needs to process the file before installing it. The file will be marked as provided by this plugin. This
389 enables introspection and detects conflicts if other plugins attempts to claim the file.
390 """
391 raise NotImplementedError
393 def metadata_or_maintscript_detector(
394 self,
395 auto_detector_id: str,
396 auto_detector: MetadataAutoDetector,
397 *,
398 package_type: PackageTypeSelector = "deb",
399 ) -> None:
400 """Provide a pre-assembly hook that can affect the metadata/maintscript of binary ("deb") packages
402 The provided hook will be run once per binary package to be assembled, and it can see all the content
403 ("data.tar") planned to be included in the deb. The hook may do any *read-only* analysis of this content
404 and provide metadata, alter substvars or inject maintscript snippets. However, the hook must *not*
405 change the content ("data.tar") part of the deb.
407 The hook will be run unconditionally for all binary packages built. When the hook does not apply to all
408 packages, it must provide its own (internal) logic for detecting whether it is relevant and reduced itself
409 to a no-op if it should not apply to the current package.
411 Hooks are run in "some implementation defined order" and should not rely on being run before or after
412 any other hook.
414 The hooks are only applied to packages defined in `debian/control`. Notably, the metadata detector will
415 not apply to auto-generated `-dbgsym` packages (as those are not listed explicitly in `debian/control`).
417 :param auto_detector_id: A plugin-wide unique ID for this detector. Packagers may use this ID for disabling
418 the detector and accordingly the ID is part of the plugin's API toward the packager.
419 :param auto_detector: The code to be called that will be run at the metadata generation state (once for each
420 binary package).
421 :param package_type: Which kind of packages this metadata detector applies to. The package type is generally
422 defined by `Package-Type` field in the binary package. The default is to only run for regular `deb` packages
423 and ignore `udeb` packages.
424 """
425 raise NotImplementedError
427 def manifest_variable(
428 self,
429 variable_name: str,
430 value: str,
431 variable_reference_documentation: Optional[str] = None,
432 ) -> None:
433 """Provide a variable that can be used in the package manifest
435 >>> # Enable users to use "{{path:BASH_COMPLETION_DIR}}/foo" in their manifest.
436 >>> api.manifest_variable( # doctest: +SKIP
437 ... "path:BASH_COMPLETION_DIR",
438 ... "/usr/share/bash-completion/completions",
439 ... variable_reference_documentation="Directory to install bash completions into",
440 ... )
442 :param variable_name: The variable name.
443 :param value: The value the variable should resolve to.
444 :param variable_reference_documentation: A short snippet of reference documentation that explains
445 the purpose of the variable.
446 """
447 raise NotImplementedError
450class MaintscriptAccessor:
451 __slots__ = ()
453 def on_configure(
454 self,
455 run_snippet: str,
456 /,
457 indent: Optional[bool] = None,
458 perform_substitution: bool = True,
459 skip_on_rollback: bool = False,
460 ) -> None:
461 """Provide a snippet to be run when the package is about to be "configured"
463 This condition is the most common "post install" condition and covers the two
464 common cases:
465 * On initial install, OR
466 * On upgrade
468 In dpkg maintscript terms, this method roughly corresponds to postinst containing
469 `if [ "$1" = configure ]; then <snippet>; fi`
471 Additionally, the condition will by default also include rollback/abort scenarios such as "above-remove",
472 which is normally what you want but most people forget about.
474 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
475 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
476 snippet may contain '{{FOO}}' substitutions by default.
477 :param skip_on_rollback: By default, this condition will also cover common rollback scenarios. This
478 is normally what you want (or benign in most cases due to the idempotence requirement for maintscripts).
479 However, you can disable the rollback cases, leaving only "On initial install OR On upgrade".
480 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
481 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
482 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
483 You are recommended to do 4 spaces of indentation when indent is False for readability.
484 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
485 substitution is provided.
486 """
487 raise NotImplementedError
489 def on_initial_install(
490 self,
491 run_snippet: str,
492 /,
493 indent: Optional[bool] = None,
494 perform_substitution: bool = True,
495 ) -> None:
496 """Provide a snippet to be run when the package is about to be "configured" for the first time
498 The snippet will only be run on the first time the package is installed (ever or since last purge).
499 Note that "first" does not mean "exactly once" as dpkg does *not* provide such semantics. There are two
500 common cases where this can snippet can be run multiple times for the same system (and why the snippet
501 must still be idempotent):
503 1) The package is installed (1), then purged and then installed again (2). This can partly be mitigated
504 by having an `on_purge` script to do clean up.
506 2) As the package is installed, the `postinst` script terminates prematurely (Disk full, power loss, etc.).
507 The user resolves the problem and runs `dpkg --configure <pkg>`, which in turn restarts the script
508 from the beginning. This is why scripts must be idempotent in general.
510 In dpkg maintscript terms, this method roughly corresponds to postinst containing
511 `if [ "$1" = configure ] && [ -z "$2" ]; then <snippet>; fi`
513 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
514 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
515 snippet may contain '{{FOO}}' substitutions by default.
516 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
517 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
518 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
519 You are recommended to do 4 spaces of indentation when indent is False for readability.
520 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
521 substitution is provided.
522 """
523 raise NotImplementedError
525 def on_upgrade(
526 self,
527 run_snippet: str,
528 /,
529 indent: Optional[bool] = None,
530 perform_substitution: bool = True,
531 ) -> None:
532 """Provide a snippet to be run when the package is about to be "configured" after an upgrade
534 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).
536 In dpkg maintscript terms, this method roughly corresponds to postinst containing
537 `if [ "$1" = configure ] && [ -n "$2" ]; then <snippet>; fi`
539 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
540 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
541 snippet may contain '{{FOO}}' substitutions by default.
542 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
543 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
544 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
545 You are recommended to do 4 spaces of indentation when indent is False for readability.
546 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
547 substitution is provided.
548 """
549 raise NotImplementedError
551 def on_upgrade_from(
552 self,
553 version: str,
554 run_snippet: str,
555 /,
556 indent: Optional[bool] = None,
557 perform_substitution: bool = True,
558 ) -> None:
559 """Provide a snippet to be run when the package is about to be "configured" after an upgrade from a given version
561 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).
563 In dpkg maintscript terms, this method roughly corresponds to postinst containing
564 `if [ "$1" = configure ] && dpkg --compare-versions le-nl "$2" ; then <snippet>; fi`
566 :param version: The version to upgrade from
567 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
568 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
569 snippet may contain '{{FOO}}' substitutions by default.
570 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
571 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
572 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
573 You are recommended to do 4 spaces of indentation when indent is False for readability.
574 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
575 substitution is provided.
576 """
577 raise NotImplementedError
579 def on_before_removal(
580 self,
581 run_snippet: str,
582 /,
583 indent: Optional[bool] = None,
584 perform_substitution: bool = True,
585 ) -> None:
586 """Provide a snippet to be run when the package is about to be removed
588 The snippet will be run before dpkg removes any files.
590 In dpkg maintscript terms, this method roughly corresponds to prerm containing
591 `if [ "$1" = remove ] ; then <snippet>; fi`
593 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
594 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
595 snippet may contain '{{FOO}}' substitutions by default.
596 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
597 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
598 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
599 You are recommended to do 4 spaces of indentation when indent is False for readability.
600 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
601 substitution is provided.
602 """
603 raise NotImplementedError
605 def on_removed(
606 self,
607 run_snippet: str,
608 /,
609 indent: Optional[bool] = None,
610 perform_substitution: bool = True,
611 ) -> None:
612 """Provide a snippet to be run when the package has been removed
614 The snippet will be run after dpkg removes the package content from the file system.
616 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.
618 In dpkg maintscript terms, this method roughly corresponds to postrm containing
619 `if [ "$1" = remove ] ; then <snippet>; fi`
621 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
622 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
623 snippet may contain '{{FOO}}' substitutions by default.
624 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
625 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
626 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
627 You are recommended to do 4 spaces of indentation when indent is False for readability.
628 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
629 substitution is provided.
630 """
631 raise NotImplementedError
633 def on_purge(
634 self,
635 run_snippet: str,
636 /,
637 indent: Optional[bool] = None,
638 perform_substitution: bool = True,
639 ) -> None:
640 """Provide a snippet to be run when the package is being purged.
642 The snippet will when the package is purged from the system.
644 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.
646 In dpkg maintscript terms, this method roughly corresponds to postrm containing
647 `if [ "$1" = purge ] ; then <snippet>; fi`
649 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.
650 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the
651 snippet may contain '{{FOO}}' substitutions by default.
652 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.
653 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues
654 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.
655 You are recommended to do 4 spaces of indentation when indent is False for readability.
656 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
657 substitution is provided.
658 """
659 raise NotImplementedError
661 def unconditionally_in_script(
662 self,
663 maintscript: Maintscript,
664 run_snippet: str,
665 /,
666 perform_substitution: bool = True,
667 ) -> None:
668 """Provide a snippet to be run in a given script
670 Run a given snippet unconditionally from a given script. The snippet must contain its own conditional
671 for when it should be run.
673 :param maintscript: The maintscript to insert the snippet into.
674 :param run_snippet: The actual shell snippet to be run. The snippet will be run unconditionally and should
675 contain its own conditions as necessary. The snippet must be idempotent. The snippet may contain newlines
676 as necessary, which will make the result more readable. Additionally, the snippet may contain '{{FOO}}'
677 substitutions by default.
678 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no
679 substitution is provided.
680 """
681 raise NotImplementedError
683 def escape_shell_words(self, *args: str) -> str:
684 """Provide sh-shell escape of strings
686 `assert escape_shell("foo", "fu bar", "baz") == 'foo "fu bar" baz'`
688 This is useful for ensuring file names and other "input" are considered one parameter even when they
689 contain spaces or shell meta-characters.
691 :param args: The string(s) to be escaped.
692 :return: Each argument escaped such that each argument becomes a single "word" and then all these words are
693 joined by a single space.
694 """
695 return util.escape_shell(*args)
698class BinaryCtrlAccessor:
699 __slots__ = ()
701 def dpkg_trigger(self, trigger_type: DpkgTriggerType, trigger_target: str) -> None:
702 """Register a declarative dpkg level trigger
704 The provided trigger will be added to the package's metadata (the triggers file of the control.tar).
706 If the trigger has already been added previously, a second call with the same trigger data will be ignored.
707 """
708 raise NotImplementedError
710 @property
711 def maintscript(self) -> MaintscriptAccessor:
712 """Attribute for manipulating maintscripts"""
713 raise NotImplementedError
715 @property
716 def substvars(self) -> "FlushableSubstvars":
717 """Attribute for manipulating dpkg substvars (deb-substvars)"""
718 raise NotImplementedError
721class VirtualPath:
722 __slots__ = ()
724 @property
725 def name(self) -> str:
726 """Basename of the path a.k.a. last segment of the path
728 In a path "usr/share/doc/pkg/changelog.gz" the basename is "changelog.gz".
730 For a directory, the basename *never* ends with a `/`.
731 """
732 raise NotImplementedError
734 @property
735 def iterdir(self) -> Iterable["VirtualPath"]:
736 """Returns an iterable that iterates over all children of this path
738 For directories, this returns an iterable of all children. For non-directories,
739 the iterable is always empty.
740 """
741 raise NotImplementedError
743 def lookup(self, path: str) -> Optional["VirtualPath"]:
744 """Perform a path lookup relative to this path
746 As an example `doc_dir = fs_root.lookup('./usr/share/doc')`
748 If the provided path starts with `/`, then the lookup is performed relative to the
749 file system root. That is, you can assume the following to always be True:
751 `fs_root.lookup("usr") == any_path_beneath_fs_root.lookup('/usr')`
753 Note: This method requires the path to be attached (see `is_detached`) regardless of
754 whether the lookup is relative or absolute.
756 If the path traverse a symlink, the symlink will be resolved.
758 :param path: The path to look. Can contain "." and ".." segments. If starting with `/`,
759 look up is performed relative to the file system root, otherwise the lookup is relative
760 to this path.
761 :return: The path object for the desired path if it can be found. Otherwise, None.
762 """
763 raise NotImplementedError
765 def all_paths(self) -> Iterable["VirtualPath"]:
766 """Iterate over this path and all of its descendants (if any)
768 If used on the root path, then every path in the package is returned.
770 The iterable is ordered, so using the order in output will be produce
771 bit-for-bit reproducible output. Additionally, a directory will always
772 be seen before its descendants. Otherwise, the order is implementation
773 defined.
775 The iteration is lazy and as a side effect do account for some obvious
776 mutation. Like if the current path is removed, then none of its children
777 will be returned (provided mutation happens before the lazy iteration
778 was required to resolve it). Likewise, mutation of the directory will
779 also work (again, provided mutation happens before the lazy iteration order).
781 :return: An ordered iterable of this path followed by its descendants.
782 """
783 raise NotImplementedError
785 @property
786 def is_detached(self) -> bool:
787 """Returns True if this path is detached
789 Paths that are detached from the file system will not be present in the package and
790 most operations are unsafe on them. This usually only happens if the path or one of
791 its parent directories are unlinked (rm'ed) from the file system tree.
793 All paths are attached by default and will only become detached as a result of
794 an action to mutate the virtual file system. Note that the file system may not
795 always be manipulated.
797 :return: True if the entry is detached. Detached entries should be discarded, so they
798 can be garbage collected.
799 """
800 raise NotImplementedError
802 # The __getitem__ behaves like __getitem__ from Dict but __iter__ would ideally work like a Sequence.
803 # However, that does not feel compatible, so lets force people to use .children instead for the Sequence
804 # behaviour to avoid surprises for now.
805 # (Maybe it is a non-issue, but it is easier to add the API later than to remove it once we have committed
806 # to using it)
807 __iter__ = None
809 def __getitem__(self, key: object) -> "VirtualPath":
810 """Lookup a (direct) child by name
812 Ignoring the possible `KeyError`, then the following are the same:
813 `fs_root["usr"] == fs_root.lookup('usr')`
815 Note that unlike `.lookup` this can only locate direct children.
816 """
817 raise NotImplementedError
819 def __delitem__(self, key) -> None:
820 """Remove a child from this node if it exists
822 If that child is a directory, then the entire tree is removed (like `rm -fr`).
823 """
824 raise NotImplementedError
826 def get(self, key: str) -> "Optional[VirtualPath]":
827 """Lookup a (direct) child by name
829 The following are the same:
830 `fs_root.get("usr") == fs_root.lookup('usr')`
832 Note that unlike `.lookup` this can only locate direct children.
833 """
834 try:
835 return self[key]
836 except KeyError:
837 return None
839 def __contains__(self, item: object) -> bool:
840 """Determine if this path includes a given child (either by object or string)
842 Examples:
844 if 'foo' in dir: ...
845 """
846 if isinstance(item, VirtualPath):
847 return item.parent_dir is self
848 if not isinstance(item, str):
849 return False
850 m = self.get(item)
851 return m is not None
853 @property
854 def path(self) -> str:
855 """Returns the "full" path for this file system entry
857 This is the path that debputy uses to refer to this file system entry. It is always
858 normalized. Use the `absolute` attribute for how the path looks
859 when the package is installed. Alternatively, there is also `fs_path`, which is the
860 path to the underlying file system object (assuming there is one). That is the one
861 you need if you want to read the file.
863 This is attribute is mostly useful for debugging or for looking up the path relative
864 to the "root" of the virtual file system that debputy maintains.
866 If the path is detached (see `is_detached`), then this method returns the path as it
867 was known prior to being detached.
868 """
869 raise NotImplementedError
871 @property
872 def absolute(self) -> str:
873 """Returns the absolute version of this path
875 This is how to refer to this path when the package is installed.
877 If the path is detached (see `is_detached`), then this method returns the last known location
878 of installation (prior to being detached).
880 :return: The absolute path of this file as it would be on the installed system.
881 """
882 p = self.path.lstrip(".")
883 if not p.startswith("/"):
884 return f"/{p}"
885 return p
887 @property
888 def parent_dir(self) -> Optional["VirtualPath"]:
889 """The parent directory of this path
891 Note this operation requires the path is "attached" (see `is_detached`). All paths are attached
892 by default but unlinking paths will cause them to become detached.
894 :return: The parent path or None for the root.
895 """
896 raise NotImplementedError
898 def stat(self) -> os.stat_result:
899 """Attempt to do stat of the underlying path (if it exists)
901 *Avoid* using `stat()` whenever possible where a more specialized attribute exist. The
902 `stat()` call returns the data from the file system and often, `debputy` does *not* track
903 its state in the file system. As an example, if you want to know the file system mode of
904 a path, please use the `mode` attribute instead.
906 This never follow symlinks (it behaves like `os.lstat`). It will raise an error
907 if the path is not backed by a file system object (that is, `has_fs_path` is False).
909 :return: The stat result or an error.
910 """
911 raise NotImplementedError()
913 @property
914 def size(self) -> int:
915 """Resolve the file size (`st_size`)
917 This may be using `stat()` and therefore `fs_path`.
919 :return: The size of the file in bytes
920 """
921 return self.stat().st_size
923 @property
924 def mode(self) -> int:
925 """Determine the mode bits of this path object
927 Note that:
928 * like with `stat` above, this never follows symlinks.
929 * the mode returned by this method is not always a 1:1 with the mode in the
930 physical file system. As an optimization, `debputy` skips unnecessary writes
931 to the underlying file system in many cases.
934 :return: The mode bits for the path.
935 """
936 raise NotImplementedError
938 @mode.setter
939 def mode(self, new_mode: int) -> None:
940 """Set the octal file mode of this path
942 Note that:
943 * this operation will fail if `path.is_read_write` returns False.
944 * this operation is generally *not* synced to the physical file system (as
945 an optimization).
947 :param new_mode: The new octal mode for this path. Note that `debputy` insists
948 that all paths have the `user read bit` and, for directories also, the
949 `user execute bit`. The absence of these minimal mode bits causes hard to
950 debug errors.
951 """
952 raise NotImplementedError
954 @property
955 def is_executable(self) -> bool:
956 """Determine whether a path is considered executable
958 Generally, this means that at least one executable bit is set. This will
959 basically always be true for directories as directories need the execute
960 parameter to be traversable.
962 :return: True if the path is considered executable with its current mode
963 """
964 return bool(self.mode & 0o0111)
966 def chmod(self, new_mode: Union[int, str]) -> None:
967 """Set the file mode of this path
969 This is similar to setting the `mode` attribute. However, this method accepts
970 a string argument, which will be parsed as a symbolic mode (example: `u+rX,go=rX`).
972 Note that:
973 * this operation will fail if `path.is_read_write` returns False.
974 * this operation is generally *not* synced to the physical file system (as
975 an optimization).
977 :param new_mode: The new mode for this path.
978 Note that `debputy` insists that all paths have the `user read bit` and, for
979 directories also, the `user execute bit`. The absence of these minimal mode
980 bits causes hard to debug errors.
981 """
982 if isinstance(new_mode, str):
983 segments = parse_symbolic_mode(new_mode, None)
984 final_mode = self.mode
985 is_dir = self.is_dir
986 for segment in segments:
987 final_mode = segment.apply(final_mode, is_dir)
988 self.mode = final_mode
989 else:
990 self.mode = new_mode
992 def chown(
993 self,
994 owner: Optional["StaticFileSystemOwner"],
995 group: Optional["StaticFileSystemGroup"],
996 ) -> None:
997 """Change the owner/group of this path
999 :param owner: The desired owner definition for this path. If None, then no change of owner is performed.
1000 :param group: The desired group definition for this path. If None, then no change of group is performed.
1001 """
1002 raise NotImplementedError
1004 @property
1005 def mtime(self) -> float:
1006 """Determine the mtime of this path object
1008 Note that:
1009 * like with `stat` above, this never follows symlinks.
1010 * the mtime returned has *not* been clamped against ´SOURCE_DATE_EPOCH`. Timestamp
1011 normalization is handled later by `debputy`.
1012 * the mtime returned by this method is not always a 1:1 with the mtime in the
1013 physical file system. As an optimization, `debputy` skips unnecessary writes
1014 to the underlying file system in many cases.
1016 :return: The mtime for the path.
1017 """
1018 raise NotImplementedError
1020 @mtime.setter
1021 def mtime(self, new_mtime: float) -> None:
1022 """Set the mtime of this path
1024 Note that:
1025 * this operation will fail if `path.is_read_write` returns False.
1026 * this operation is generally *not* synced to the physical file system (as
1027 an optimization).
1029 :param new_mtime: The new mtime of this path. Note that the caller does not need to
1030 account for `SOURCE_DATE_EPOCH`. Timestamp normalization is handled later.
1031 """
1032 raise NotImplementedError
1034 def readlink(self) -> str:
1035 """Determine the link target of this path assuming it is a symlink
1037 For paths where `is_symlink` is True, this already returns a link target even when
1038 `has_fs_path` is False.
1040 :return: The link target of the path or an error is this is not a symlink
1041 """
1042 raise NotImplementedError()
1044 @overload
1045 def open( 1045 ↛ exitline 1045 didn't jump to the function exit
1046 self,
1047 *,
1048 byte_io: Literal[False] = False,
1049 buffering: Optional[int] = ...,
1050 ) -> TextIO: ...
1052 @overload
1053 def open( 1053 ↛ exitline 1053 didn't jump to the function exit
1054 self,
1055 *,
1056 byte_io: Literal[True],
1057 buffering: Optional[int] = ...,
1058 ) -> BinaryIO: ...
1060 @overload
1061 def open( 1061 ↛ exitline 1061 didn't jump to the function exit
1062 self,
1063 *,
1064 byte_io: bool,
1065 buffering: Optional[int] = ...,
1066 ) -> Union[TextIO, BinaryIO]: ...
1068 def open(
1069 self,
1070 *,
1071 byte_io: bool = False,
1072 buffering: int = -1,
1073 ) -> Union[TextIO, BinaryIO]:
1074 """Open the file for reading. Usually used with a context manager
1076 By default, the file is opened in text mode (utf-8). Binary mode can be requested
1077 via the `byte_io` parameter. This operation is only valid for files (`is_file` returns
1078 `True`). Usage on symlinks and directories will raise exceptions.
1080 This method *often* requires the `fs_path` to be present. However, tests as a notable
1081 case can inject content without having the `fs_path` point to a real file. (To be clear,
1082 such tests are generally expected to ensure `has_fs_path` returns `True`).
1085 :param byte_io: If True, open the file in binary mode (like `rb` for `open`)
1086 :param buffering: Same as open(..., buffering=...) where supported. Notably during
1087 testing, the content may be purely in memory and use a BytesIO/StringIO
1088 (which does not accept that parameter, but then is buffered in a different way)
1089 :return: The file handle.
1090 """
1092 if not self.is_file: 1092 ↛ 1093line 1092 didn't jump to line 1093, because the condition on line 1092 was never true
1093 raise TypeError(f"Cannot open {self.path} for reading: It is not a file")
1095 if byte_io:
1096 return open(self.fs_path, "rb", buffering=buffering)
1097 return open(self.fs_path, "rt", encoding="utf-8", buffering=buffering)
1099 @property
1100 def fs_path(self) -> str:
1101 """Request the underling fs_path of this path
1103 Only available when `has_fs_path` is True. Generally this should only be used for files to read
1104 the contents of the file and do some action based on the parsed result.
1106 The path should only be used for read-only purposes as debputy may assume that it is safe to have
1107 multiple paths pointing to the same file system path.
1109 Note that:
1110 * This is often *not* available for directories and symlinks.
1111 * The debputy in-memory file system overrules the physical file system. Attempting to "fix" things
1112 by using `os.chmod` or `os.unlink`'ing files, etc. will generally not do as you expect. Best case,
1113 your actions are ignored and worst case it will cause the build to fail as it violates debputy's
1114 internal invariants.
1116 :return: The path to the underlying file system object on the build system or an error if no such
1117 file exist (see `has_fs_path`).
1118 """
1119 raise NotImplementedError()
1121 @property
1122 def is_dir(self) -> bool:
1123 """Determine if this path is a directory
1125 Never follows symlinks.
1127 :return: True if this path is a directory. False otherwise.
1128 """
1129 raise NotImplementedError()
1131 @property
1132 def is_file(self) -> bool:
1133 """Determine if this path is a directory
1135 Never follows symlinks.
1137 :return: True if this path is a regular file. False otherwise.
1138 """
1139 raise NotImplementedError()
1141 @property
1142 def is_symlink(self) -> bool:
1143 """Determine if this path is a symlink
1145 :return: True if this path is a symlink. False otherwise.
1146 """
1147 raise NotImplementedError()
1149 @property
1150 def has_fs_path(self) -> bool:
1151 """Determine whether this path is backed by a file system path
1153 :return: True if this path is backed by a file system object on the build system.
1154 """
1155 raise NotImplementedError()
1157 @property
1158 def is_read_write(self) -> bool:
1159 """When true, the file system entry may be mutated
1161 Read-write rules are:
1163 +--------------------------+-------------------+------------------------+
1164 | File system | From / Inside | Read-Only / Read-Write |
1165 +--------------------------+-------------------+------------------------+
1166 | Source directory | Any context | Read-Only |
1167 | Binary staging directory | Package Processor | Read-Write |
1168 | Binary staging directory | Metadata Detector | Read-Only |
1169 +--------------------------+-------------------+------------------------+
1171 These rules apply to the virtual file system (`debputy` cannot enforce
1172 these rules in the underlying file system). The `debputy` code relies
1173 on these rules for its logic in multiple places to catch bugs and for
1174 optimizations.
1176 As an example, the reason why the file system is read-only when Metadata
1177 Detectors are run is based the contents of the file system has already
1178 been committed. New files will not be included, removals of existing
1179 files will trigger a hard error when the package is assembled, etc.
1180 To avoid people spending hours debugging why their code does not work
1181 as intended, `debputy` instead throws a hard error if you try to mutate
1182 the file system when it is read-only mode to "fail fast".
1184 :return: Whether file system mutations are permitted.
1185 """
1186 return False
1188 def mkdir(self, name: str) -> "VirtualPath":
1189 """Create a new subdirectory of the current path
1191 :param name: Basename of the new directory. The directory must not contain a path
1192 with this basename.
1193 :return: The new subdirectory
1194 """
1195 raise NotImplementedError
1197 def mkdirs(self, path: str) -> "VirtualPath":
1198 """Ensure a given path exists and is a directory.
1200 :param path: Path to the directory to create. Any parent directories will be
1201 created as needed. If the path already exists and is a directory, then it
1202 is returned. If any part of the path exists and that is not a directory,
1203 then the `mkdirs` call will raise an error.
1204 :return: The directory denoted by the given path
1205 """
1206 raise NotImplementedError
1208 def add_file(
1209 self,
1210 name: str,
1211 *,
1212 unlink_if_exists: bool = True,
1213 use_fs_path_mode: bool = False,
1214 mode: int = 0o0644,
1215 mtime: Optional[float] = None,
1216 ) -> ContextManager["VirtualPath"]:
1217 """Add a new regular file as a child of this path
1219 This method will insert a new file into the virtual file system as a child
1220 of the current path (which must be a directory). The caller must use the
1221 return value as a context manager (see example). During the life-cycle of
1222 the managed context, the caller can fill out the contents of the file
1223 from the new path's `fs_path` attribute. The `fs_path` will exist as an
1224 empty file when the context manager is entered.
1226 Once the context manager exits, mutation of the `fs_path` is no longer permitted.
1228 >>> import subprocess
1229 >>> path = ... # doctest: +SKIP
1230 >>> with path.add_file("foo") as new_file, open(new_file.fs_path, "w") as fd: # doctest: +SKIP
1231 ... fd.writelines(["Some", "Content", "Here"])
1233 The caller can replace the provided `fs_path` entirely provided at the end result
1234 (when the context manager exits) is a regular file with no hard links.
1236 Note that this operation will fail if `path.is_read_write` returns False.
1238 :param name: Basename of the new file
1239 :param unlink_if_exists: If the name was already in use, then either an exception is thrown
1240 (when `unlink_if_exists` is False) or the path will be removed via ´unlink(recursive=False)`
1241 (when `unlink_if_exists` is True)
1242 :param use_fs_path_mode: When True, the file created will have this mode in the physical file
1243 system. When the context manager exists, `debputy` will refresh its mode to match the mode
1244 in the physical file system. This is primarily useful if the caller uses a subprocess to
1245 mutate the path and the file mode is relevant for this tool (either as input or output).
1246 When the parameter is false, the new file is guaranteed to be readable and writable for
1247 the current user. However, no other guarantees are given (not even that it matches the
1248 `mode` parameter and any changes to the mode in the physical file system will be ignored.
1249 :param mode: This is the initial file mode. Note the `use_fs_path_mode` parameter for how
1250 this interacts with the physical file system.
1251 :param mtime: If the caller has a more accurate mtime than the mtime of the generated file,
1252 then it can be provided here. Note that all mtimes will later be clamped based on
1253 `SOURCE_DATE_EPOCH`. This parameter is only for when the conceptual mtime of this path
1254 should be earlier than `SOURCE_DATE_EPOCH`.
1255 :return: A Context manager that upon entering provides a `VirtualPath` instance for the
1256 new file. The instance remains valid after the context manager exits (assuming it exits
1257 successfully), but the file denoted by `fs_path` must not be changed after the context
1258 manager exits
1259 """
1260 raise NotImplementedError
1262 def replace_fs_path_content(
1263 self,
1264 *,
1265 use_fs_path_mode: bool = False,
1266 ) -> ContextManager[str]:
1267 """Replace the contents of this file via inline manipulation
1269 Used as a context manager to provide the fs path for manipulation.
1271 Example:
1272 >>> import subprocess
1273 >>> path = ... # doctest: +SKIP
1274 >>> with path.replace_fs_path_content() as fs_path: # doctest: +SKIP
1275 ... subprocess.check_call(['strip', fs_path]) # doctest: +SKIP
1277 The provided file system path should be manipulated inline. The debputy framework may
1278 copy it first as necessary and therefore the provided fs_path may be different from
1279 `path.fs_path` prior to entering the context manager.
1281 Note that this operation will fail if `path.is_read_write` returns False.
1283 If the mutation causes the returned `fs_path` to be a non-file or a hard-linked file
1284 when the context manager exits, `debputy` will raise an error at that point. To preserve
1285 the internal invariants of `debputy`, the path will be unlinked as `debputy` cannot
1286 reliably restore the path.
1288 :param use_fs_path_mode: If True, any changes to the mode on the physical FS path will be
1289 recorded as the desired mode of the file when the contextmanager ends. The provided FS path
1290 with start with the current mode when `use_fs_path_mode` is True. Otherwise, `debputy` will
1291 ignore the mode of the file system entry and reuse its own current mode
1292 definition.
1293 :return: A Context manager that upon entering provides the path to a muable (copy) of
1294 this path's `fs_path` attribute. The file on the underlying path may be mutated however
1295 the caller wishes until the context manager exits.
1296 """
1297 raise NotImplementedError
1299 def add_symlink(self, link_name: str, link_target: str) -> "VirtualPath":
1300 """Add a new regular file as a child of this path
1302 This will create a new symlink inside the current path. If the path already exists,
1303 the existing path will be unlinked via `unlink(recursive=False)`.
1305 Note that this operation will fail if `path.is_read_write` returns False.
1307 :param link_name: The basename of the link file entry.
1308 :param link_target: The target of the link. Link target normalization will
1309 be handled by `debputy`, so the caller can use relative or absolute paths.
1310 (At the time of writing, symlink target normalization happens late)
1311 :return: The newly created symlink.
1312 """
1313 raise NotImplementedError
1315 def unlink(self, *, recursive: bool = False) -> None:
1316 """Unlink a file or a directory
1318 This operation will remove the path from the file system (causing `is_detached` to return True).
1320 When the path is a:
1322 * symlink, then the symlink itself is removed. The target (if present) is not affected.
1323 * *non-empty* directory, then the `recursive` parameter decides the outcome. An empty
1324 directory will be removed regardless of the value of `recursive`.
1326 Note that:
1327 * the root directory cannot be deleted.
1328 * this operation will fail if `path.is_read_write` returns False.
1330 :param recursive: If True, then non-empty directories will be unlinked as well removing everything inside them
1331 as well. When False, an error is raised if the path is a non-empty directory
1332 """
1333 raise NotImplementedError
1335 def interpreter(self) -> Optional[Interpreter]:
1336 """Determine the interpreter of the file (`#!`-line details)
1338 Note: this method is only applicable for files (`is_file` is True).
1340 :return: The detected interpreter if present or None if no interpreter can be detected.
1341 """
1342 if not self.is_file:
1343 raise TypeError("Only files can have interpreters")
1344 try:
1345 with self.open(byte_io=True, buffering=4096) as fd:
1346 return extract_shebang_interpreter_from_file(fd)
1347 except (PureVirtualPathError, TestPathWithNonExistentFSPathError):
1348 return None
1350 def metadata(
1351 self,
1352 metadata_type: Type[PMT],
1353 ) -> PathMetadataReference[PMT]:
1354 """Fetch the path metadata reference to access the underlying metadata
1356 Calling this method returns a reference to an arbitrary piece of metadata associated
1357 with this path. Plugins can store any arbitrary data associated with a given path.
1358 Keep in mind that the metadata is stored in memory, so keep the size in moderation.
1360 To store / update the metadata, the path must be in read-write mode. However,
1361 already stored metadata remains accessible even if the path becomes read-only.
1363 Note this method is not applicable if the path is detached
1365 :param metadata_type: Type of the metadata being stored.
1366 :return: A reference to the metadata.
1367 """
1368 raise NotImplementedError
1371class FlushableSubstvars(Substvars):
1372 __slots__ = ()
1374 @contextlib.contextmanager
1375 def flush(self) -> Iterator[str]:
1376 """Temporarily write the substvars to a file and then re-read it again
1378 >>> s = FlushableSubstvars()
1379 >>> 'Test:Var' in s
1380 False
1381 >>> with s.flush() as name, open(name, 'wt', encoding='utf-8') as fobj:
1382 ... _ = fobj.write('Test:Var=bar\\n') # "_ = " is to ignore the return value of write
1383 >>> 'Test:Var' in s
1384 True
1386 Used as a context manager to define when the file is flushed and can be
1387 accessed via the file system. If the context terminates successfully, the
1388 file is read and its content replaces the current substvars.
1390 This is mostly useful if the plugin needs to interface with a third-party
1391 tool that requires a file as interprocess communication (IPC) for sharing
1392 the substvars.
1394 The file may be truncated or completed replaced (change inode) as long as
1395 the provided path points to a regular file when the context manager
1396 terminates successfully.
1398 Note that any manipulation of the substvars via the `Substvars` API while
1399 the file is flushed will silently be discarded if the context manager completes
1400 successfully.
1401 """
1402 with tempfile.NamedTemporaryFile(mode="w+t", encoding="utf-8") as tmp:
1403 self.write_substvars(tmp)
1404 tmp.flush() # Temping to use close, but then we have to manually delete the file.
1405 yield tmp.name
1406 # Re-open; seek did not work when I last tried (if I did it work, feel free to
1407 # convert back to seek - as long as it works!)
1408 with open(tmp.name, "rt", encoding="utf-8") as fd:
1409 self.read_substvars(fd)
1411 def save(self) -> None:
1412 # Promote the debputy extension over `save()` for the plugins.
1413 if self._substvars_path is None:
1414 raise TypeError(
1415 "Please use `flush()` extension to temporarily write the substvars to the file system"
1416 )
1417 super().save()
1420class ServiceRegistry(Generic[DSD]):
1421 __slots__ = ()
1423 def register_service(
1424 self,
1425 path: VirtualPath,
1426 name: Union[str, List[str]],
1427 *,
1428 type_of_service: str = "service", # "timer", etc.
1429 service_scope: str = "system",
1430 enable_by_default: bool = True,
1431 start_by_default: bool = True,
1432 default_upgrade_rule: ServiceUpgradeRule = "restart",
1433 service_context: Optional[DSD] = None,
1434 ) -> None:
1435 """Register a service detected in the package
1437 All the details will either be provided as-is or used as default when the plugin provided
1438 integration code is called.
1440 Two services from different service managers are considered related when:
1442 1) They are of the same type (`type_of_service`) and has the same scope (`service_scope`), AND
1443 2) Their plugin provided names has an overlap
1445 Related services can be covered by the same service definition in the manifest.
1447 :param path: The path defining this service.
1448 :param name: The name of the service. Multiple ones can be provided if the service has aliases.
1449 Note that when providing multiple names, `debputy` will use the first name in the list as the
1450 default name if it has to choose. Any alternative name provided can be used by the packager
1451 to identify this service.
1452 :param type_of_service: The type of service. By default, this is "service", but plugins can
1453 provide other types (such as "timer" for the systemd timer unit).
1454 :param service_scope: The scope for this service. By default, this is "system" meaning the
1455 service is a system-wide service. Service managers can define their own scopes such as
1456 "user" (which is used by systemd for "per-user" services).
1457 :param enable_by_default: Whether the service should be enabled by default, assuming the
1458 packager does not explicitly override this setting.
1459 :param start_by_default: Whether the service should be started by default on install, assuming
1460 the packager does not explicitly override this setting.
1461 :param default_upgrade_rule: The default value for how the service should be processed during
1462 upgrades. Options are:
1463 * `do-nothing`: The plugin should not interact with the running service (if any)
1464 (maintenance of the enabled start, start on install, etc. are still applicable)
1465 * `reload`: The plugin should attempt to reload the running service (if any).
1466 Note: In combination with `auto_start_in_install == False`, be careful to not
1467 start the service if not is not already running.
1468 * `restart`: The plugin should attempt to restart the running service (if any).
1469 Note: In combination with `auto_start_in_install == False`, be careful to not
1470 start the service if not is not already running.
1471 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`
1472 and start it against in the `postinst` script.
1474 :param service_context: Any custom data that the detector want to pass along to the
1475 integrator for this service.
1476 """
1477 raise NotImplementedError
1480@dataclasses.dataclass(slots=True, frozen=True)
1481class ParserAttributeDocumentation:
1482 attributes: FrozenSet[str]
1483 description: Optional[str]
1486def undocumented_attr(attr: str) -> ParserAttributeDocumentation:
1487 """Describe an attribute as undocumented
1489 If you for some reason do not want to document a particular attribute, you can mark it as
1490 undocumented. This is required if you are only documenting a subset of the attributes,
1491 because `debputy` assumes any omission to be a mistake.
1492 """
1493 return ParserAttributeDocumentation(
1494 frozenset({attr}),
1495 None,
1496 )
1499@dataclasses.dataclass(slots=True, frozen=True)
1500class ParserDocumentation:
1501 title: Optional[str] = None
1502 description: Optional[str] = None
1503 attribute_doc: Optional[Sequence[ParserAttributeDocumentation]] = None
1504 alt_parser_description: Optional[str] = None
1505 documentation_reference_url: Optional[str] = None
1507 def replace(self, **changes: Any) -> "ParserDocumentation":
1508 return dataclasses.replace(self, **changes)
1511@dataclasses.dataclass(slots=True, frozen=True)
1512class TypeMappingExample(Generic[S]):
1513 source_input: S
1516@dataclasses.dataclass(slots=True, frozen=True)
1517class TypeMappingDocumentation(Generic[S]):
1518 description: Optional[str] = None
1519 examples: Sequence[TypeMappingExample[S]] = tuple()
1522def type_mapping_example(source_input: S) -> TypeMappingExample[S]:
1523 return TypeMappingExample(source_input)
1526def type_mapping_reference_documentation(
1527 *,
1528 description: Optional[str] = None,
1529 examples: Union[TypeMappingExample[S], Iterable[TypeMappingExample[S]]] = tuple(),
1530) -> TypeMappingDocumentation[S]:
1531 e = (
1532 tuple([examples])
1533 if isinstance(examples, TypeMappingExample)
1534 else tuple(examples)
1535 )
1536 return TypeMappingDocumentation(
1537 description=description,
1538 examples=e,
1539 )
1542def documented_attr(
1543 attr: Union[str, Iterable[str]],
1544 description: str,
1545) -> ParserAttributeDocumentation:
1546 """Describe an attribute or a group of attributes
1548 :param attr: A single attribute or a sequence of attributes. The attribute must be the
1549 attribute name as used in the source format version of the TypedDict.
1551 If multiple attributes are provided, they will be documented together. This is often
1552 useful if these attributes are strongly related (such as different names for the same
1553 target attribute).
1554 :param description: The description the user should see for this attribute / these
1555 attributes. This parameter can be a Python format string with variables listed in
1556 the description of `reference_documentation`.
1557 :return: An opaque representation of the documentation,
1558 """
1559 attributes = [attr] if isinstance(attr, str) else attr
1560 return ParserAttributeDocumentation(
1561 frozenset(attributes),
1562 description,
1563 )
1566def reference_documentation(
1567 title: str = "Auto-generated reference documentation for {RULE_NAME}",
1568 description: Optional[str] = textwrap.dedent(
1569 """\
1570 This is an automatically generated reference documentation for {RULE_NAME}. It is generated
1571 from input provided by {PLUGIN_NAME} via the debputy API.
1573 (If you are the provider of the {PLUGIN_NAME} plugin, you can replace this text with
1574 your own documentation by providing the `inline_reference_documentation` when registering
1575 the manifest rule.)
1576 """
1577 ),
1578 attributes: Optional[Sequence[ParserAttributeDocumentation]] = None,
1579 non_mapping_description: Optional[str] = None,
1580 reference_documentation_url: Optional[str] = None,
1581) -> ParserDocumentation:
1582 """Provide inline reference documentation for the manifest snippet
1584 For parameters that mention that they are a Python format, the following format variables
1585 are available:
1587 * RULE_NAME: Name of the rule. If manifest snippet has aliases, this will be the name of
1588 the alias provided by the user.
1589 * MANIFEST_FORMAT_DOC: Path OR URL to the "MANIFEST-FORMAT" reference documentation from
1590 `debputy`. By using the MANIFEST_FORMAT_DOC variable, you ensure that you point to the
1591 file that matches the version of `debputy` itself.
1592 * PLUGIN_NAME: Name of the plugin providing this rule.
1594 :param title: The text you want the user to see as for your rule. A placeholder is provided by default.
1595 This parameter can be a Python format string with the above listed variables.
1596 :param description: The text you want the user to see as a description for the rule. An auto-generated
1597 placeholder is provided by default saying that no human written documentation was provided.
1598 This parameter can be a Python format string with the above listed variables.
1599 :param attributes: A sequence of attribute-related documentation. Each element of the sequence should
1600 be the result of `documented_attr` or `undocumented_attr`. The sequence must cover all source
1601 attributes exactly once.
1602 :param non_mapping_description: The text you want the user to see as the description for your rule when
1603 `debputy` describes its non-mapping format. Must not be provided for rules that do not have an
1604 (optional) non-mapping format as source format. This parameter can be a Python format string with
1605 the above listed variables.
1606 :param reference_documentation_url: A URL to the reference documentation.
1607 :return: An opaque representation of the documentation,
1608 """
1609 return ParserDocumentation(
1610 title,
1611 description,
1612 attributes,
1613 non_mapping_description,
1614 reference_documentation_url,
1615 )
1618class ServiceDefinition(Generic[DSD]):
1619 __slots__ = ()
1621 @property
1622 def name(self) -> str:
1623 """Name of the service registered by the plugin
1625 This is always a plugin provided name for this service (that is, `x.name in x.names`
1626 will always be `True`). Where possible, this will be the same as the one that the
1627 packager provided when they provided any configuration related to this service.
1628 When not possible, this will be the first name provided by the plugin (`x.names[0]`).
1630 If all the aliases are equal, then using this attribute will provide traceability
1631 between the manifest and the generated maintscript snippets. When the exact name
1632 used is important, the plugin should ignore this attribute and pick the name that
1633 is needed.
1634 """
1635 raise NotImplementedError
1637 @property
1638 def names(self) -> Sequence[str]:
1639 """All *plugin provided* names and aliases of the service
1641 This is the name/sequence of names that the plugin provided when it registered
1642 the service earlier.
1643 """
1644 raise NotImplementedError
1646 @property
1647 def path(self) -> VirtualPath:
1648 """The registered path for this service
1650 :return: The path that was associated with this service when it was registered
1651 earlier.
1652 """
1653 raise NotImplementedError
1655 @property
1656 def type_of_service(self) -> str:
1657 """Type of the service such as "service" (daemon), "timer", etc.
1659 :return: The type of service scope. It is the same value as the one as the plugin provided
1660 when registering the service (if not explicitly provided, it defaults to "service").
1661 """
1662 raise NotImplementedError
1664 @property
1665 def service_scope(self) -> str:
1666 """Service scope such as "system" or "user"
1668 :return: The service scope. It is the same value as the one as the plugin provided
1669 when registering the service (if not explicitly provided, it defaults to "system")
1670 """
1671 raise NotImplementedError
1673 @property
1674 def auto_enable_on_install(self) -> bool:
1675 """Whether the service should be auto-enabled on install
1677 :return: True if the service should be enabled automatically, false if not.
1678 """
1679 raise NotImplementedError
1681 @property
1682 def auto_start_on_install(self) -> bool:
1683 """Whether the service should be auto-started on install
1685 :return: True if the service should be started automatically, false if not.
1686 """
1687 raise NotImplementedError
1689 @property
1690 def on_upgrade(self) -> ServiceUpgradeRule:
1691 """How to handle the service during an upgrade
1693 Options are:
1694 * `do-nothing`: The plugin should not interact with the running service (if any)
1695 (maintenance of the enabled start, start on install, etc. are still applicable)
1696 * `reload`: The plugin should attempt to reload the running service (if any).
1697 Note: In combination with `auto_start_in_install == False`, be careful to not
1698 start the service if not is not already running.
1699 * `restart`: The plugin should attempt to restart the running service (if any).
1700 Note: In combination with `auto_start_in_install == False`, be careful to not
1701 start the service if not is not already running.
1702 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`
1703 and start it against in the `postinst` script.
1705 Note: In all cases, the plugin should still consider what to do in
1706 `prerm remove`, which is the last point in time where the plugin can rely on the
1707 service definitions in the file systems to stop the services when the package is
1708 being uninstalled.
1710 :return: The service restart rule
1711 """
1712 raise NotImplementedError
1714 @property
1715 def definition_source(self) -> str:
1716 """Describes where this definition came from
1718 If the definition is provided by the packager, then this will reference the part
1719 of the manifest that made this definition. Otherwise, this will be a reference
1720 to the plugin providing this definition.
1722 :return: The source of this definition
1723 """
1724 raise NotImplementedError
1726 @property
1727 def is_plugin_provided_definition(self) -> bool:
1728 """Whether the definition source points to the plugin or a package provided definition
1730 :return: True if definition is 100% from the plugin. False if the definition is partially
1731 or fully from another source (usually, the packager via the manifest).
1732 """
1733 raise NotImplementedError
1735 @property
1736 def service_context(self) -> Optional[DSD]:
1737 """Custom service context (if any) provided by the detector code of the plugin
1739 :return: If the detection code provided a custom data when registering the
1740 service, this attribute will reference that data. If nothing was provided,
1741 then this attribute will be None.
1742 """
1743 raise NotImplementedError