Coverage for src/debputy/maintscript_snippet.py: 64%
238 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-17 19:43 +0000
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-17 19:43 +0000
1import collections
2import dataclasses
3import enum
4import typing
5from typing import Self, TYPE_CHECKING
6from collections.abc import Sequence, Iterable
8from debputy.manifest_parser.tagging_types import DebputyDispatchableType
9from debputy.manifest_parser.util import AttributePath
10from debputy.substitution import Substitution
11from debputy.util import POSTINST_DEFAULT_CONDITION
13if TYPE_CHECKING:
14 from debputy.plugins.debputy.binary_package_rules import (
15 MaintainerProvidedMaintscriptSnippetContainer,
16 )
18# The standard ctrl scripts from `dpkg` (`config` is from `debconf`)
19DPKG_DEB_CONTROL_SCRIPTS = frozenset(
20 {
21 "preinst",
22 "prerm",
23 "postinst",
24 "postrm",
25 }
26)
27# all control scripts for debs and udebs respectively.
28ALL_DEB_CONTROL_SCRIPTS = DPKG_DEB_CONTROL_SCRIPTS | {"config"}
29ALL_UDEB_CONTROL_SCRIPTS = frozenset(
30 {
31 "postinst",
32 "menutest",
33 "isinstallable",
34 "config",
35 }
36)
37ALL_CONTROL_SCRIPTS = DPKG_DEB_CONTROL_SCRIPTS | ALL_UDEB_CONTROL_SCRIPTS
38# TODO: Unclear what is missing to support the remaining `udeb` scripts.
39#
40# Note depending on support, this might change into scripts that we do something useful
41# for vs. scripts we just handle as a PPF.
42SUPPORTED_UDEB_SCRIPTS = frozenset({"postinst"})
45class SnippetAnchor(enum.StrEnum):
46 CUSTOM_SCRIPT_PLACEHOLDER = "custom-script-placeholder"
47 """
48 Placeholder.
50 Custom scripts should not have a specific place but rather a "before X/after X" logic.
51 """
52 USER_CREATION = "user-creation"
53 CONFIGURATION_MANAGEMENT = "configuration-management"
54 # TODO: Rename before it becomes public
55 _BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE = (
56 "between-configuration-maagement-and-service"
57 )
58 SERVICE = "service"
61class SnippetResolver:
63 def _resolve_snippet(self) -> str:
64 raise NotImplementedError
66 def resolve(self) -> str:
67 snippet = self._resolve_snippet()
68 if not snippet.endswith("\n"): 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 return snippet + "\n"
70 return snippet
72 def apply_manifest_substitution(
73 self,
74 definition_source: str,
75 substitution: Substitution,
76 ) -> "SnippetResolver":
77 return SubstitutingSnippetResolver(self, definition_source, substitution)
79 @classmethod
80 def snippet(cls, snippet: str) -> "SnippetResolver":
81 return ConstantSnippetResolver(snippet, "IGNORED", None)
83 @classmethod
84 def lazy_snippet_no_manifest_substitution(
85 cls,
86 impl: collections.abc.Callable[[], str],
87 ) -> "SnippetResolver":
88 return LazySnippetResolver(impl)
90 @classmethod
91 def lazy_snippet(
92 cls,
93 impl: collections.abc.Callable[[], str],
94 ) -> "SnippetResolver":
95 return LazySnippetResolver(impl)
98class ConstantSnippetResolver(SnippetResolver):
99 def __init__(
100 self, snippet: str, definition_source: str, substitution: Substitution | None
101 ) -> None:
102 self._snippet = snippet
103 self._definition_source = definition_source
104 self._substitution = substitution
106 @property
107 def provides_substitution(self) -> bool:
108 return self._substitution is not None
110 def _resolve_snippet(self) -> str:
111 template = self._snippet
112 substitution = self._substitution
113 if substitution: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 return substitution.substitute(
115 template,
116 self._definition_source,
117 )
118 return template
121class CachingSnippetResolver(SnippetResolver):
123 def __init__(self) -> None:
124 self._resolved: str | None = None
126 def _resolve_snippet_uncached(self) -> str:
127 raise NotImplementedError
129 def _resolve_snippet(self) -> str:
130 resolved = self._resolved
131 if resolved is not None:
132 return resolved
133 resolved = self._resolve_snippet_uncached()
134 self._resolved = resolved
135 return resolved
138class LazySnippetResolver(CachingSnippetResolver):
140 def __init__(self, resolver: collections.abc.Callable[[], str]) -> None:
141 super().__init__()
142 self._resolver = resolver
144 def _resolve_snippet_uncached(self) -> str:
145 return self._resolver()
148class SubstitutingSnippetResolver(CachingSnippetResolver):
150 def __init__(
151 self,
152 wrapped_resolver: SnippetResolver,
153 definition_source: str,
154 substitution: Substitution,
155 ) -> None:
156 super().__init__()
157 self._wrapped_resolver = wrapped_resolver
158 self._definition_source = definition_source
159 self._substitution = substitution
161 def _resolve_snippet_uncached(self) -> str:
162 template = self._wrapped_resolver.resolve()
163 return self._substitution.substitute(template, self._definition_source)
166@dataclasses.dataclass(slots=True, frozen=True)
167class UnboundMaintscriptSnippet:
168 definition_source: str
169 snippet: SnippetResolver
170 snippet_anchor: SnippetAnchor = (
171 SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE
172 )
173 uses_debconf: bool = False
175 def script_content(self) -> str:
176 snippet = self.snippet.resolve()
177 lines = [
178 f"# Snippet source: {self.definition_source}\n",
179 snippet,
180 ]
181 return "".join(lines)
184@dataclasses.dataclass(slots=True, frozen=True)
185class MaintscriptSnippetForNamedScript:
186 definition_source: str
187 maintscript_name: str
188 snippet: SnippetResolver
189 snippet_anchor: SnippetAnchor = (
190 SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE
191 )
192 uses_debconf: bool = False
194 def apply_to_container(
195 self,
196 maintscript_container: "PackageMaintscriptSnippetContainer",
197 ) -> None:
198 maintscript_container[self.maintscript_name].append(self)
200 def script_content(self) -> str:
201 snippet = self.snippet.resolve()
202 lines = [
203 f"# Snippet source: {self.definition_source}\n",
204 snippet,
205 ]
206 return "".join(lines)
209MaintscriptSnippet: typing.TypeAlias = (
210 UnboundMaintscriptSnippet | MaintscriptSnippetForNamedScript
211)
214class MaintscriptSnippetContainer:
215 def __init__(self, script_name: str) -> None:
216 self._script_name = script_name
217 self._snippets_by_order: dict[SnippetAnchor, list[MaintscriptSnippet]] = {}
219 def copy(self) -> "MaintscriptSnippetContainer":
220 instance = self.__class__(self._script_name)
221 instance._snippets_by_order = self._snippets_by_order.copy()
222 return instance
224 def append(self, maintscript_snippet: MaintscriptSnippet) -> None:
225 assert maintscript_snippet.snippet_anchor
226 if ( 226 ↛ 230line 226 didn't jump to line 230 because the condition on line 226 was never true
227 isinstance(maintscript_snippet, MaintscriptSnippetForNamedScript)
228 and maintscript_snippet.maintscript_name != self._script_name
229 ):
230 raise ValueError(
231 f"Cannot apply snippet for {maintscript_snippet.maintscript_name}"
232 f" to {self._script_name}"
233 )
234 if maintscript_snippet.snippet_anchor not in self._snippets_by_order:
235 self._snippets_by_order[maintscript_snippet.snippet_anchor] = []
236 self._snippets_by_order[maintscript_snippet.snippet_anchor].append(
237 maintscript_snippet
238 )
240 def all_snippets(self) -> Iterable[MaintscriptSnippet]:
241 for sa in SnippetAnchor:
242 if sa not in self._snippets_by_order:
243 continue
244 yield from self._snippets_by_order[sa]
246 def needs_debconf(self) -> bool:
247 return any(
248 s.uses_debconf
249 for snippet_lists in self._snippets_by_order.values()
250 for s in snippet_lists
251 )
253 def generate_snippet(
254 self,
255 *,
256 tool_with_version: str | None = None,
257 snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE,
258 reverse: bool = False,
259 ) -> str | None:
260 snippets = self._snippets_by_order.get(snippet_anchor)
261 if not snippets:
262 return None
263 if reverse:
264 resulting_snippets = reversed(snippets)
265 else:
266 resulting_snippets = snippets
267 inner_content = "".join(s.script_content() for s in resulting_snippets)
269 if not inner_content: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 return None
272 if tool_with_version: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 return (
274 f"# Automatically added by {tool_with_version}\n"
275 + inner_content
276 + "# End automatically added section"
277 )
278 return inner_content
281class PackageMaintscriptSnippetContainer(
282 collections.abc.Mapping[str, MaintscriptSnippetContainer]
283):
285 def __init__(self, supported_scripts: frozenset[str] | None = None) -> None:
286 self._supported_scripts = (
287 supported_scripts if supported_scripts is not None else ALL_CONTROL_SCRIPTS
288 )
289 self._script_containers: dict[str, MaintscriptSnippetContainer] = {}
291 def __len__(self):
292 return len(self._script_containers)
294 def __iter__(self) -> collections.abc.Iterator[str]:
295 yield from self._script_containers
297 def __getitem__(self, key: str, /) -> MaintscriptSnippetContainer:
298 if key not in self._supported_scripts: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 raise KeyError(key)
300 try:
301 return self._script_containers[key]
302 except KeyError:
303 pass
304 c = MaintscriptSnippetContainer(key)
305 self._script_containers[key] = c
306 return c
308 def apply_maintainer_provided_snippets(
309 self,
310 maintainer_provided_maintscript_snippets: "MaintainerProvidedMaintscriptSnippetContainer",
311 ) -> None:
312 for snippet in maintainer_provided_maintscript_snippets.snippets:
313 snippet.apply_to_container(self)
316class DpkgMaintscriptHelperCommand(DebputyDispatchableType):
317 __slots__ = ("cmdline", "definition_source")
319 def __init__(self, cmdline: Sequence[str], definition_source: str) -> None:
320 super().__init__()
321 self.cmdline = cmdline
322 self.definition_source = definition_source
324 @classmethod
325 def _finish_cmd(
326 cls,
327 definition_source: str,
328 cmdline: list[str],
329 prior_version: str | None,
330 owning_package: str | None,
331 ) -> Self:
332 if prior_version is not None:
333 cmdline.append(prior_version)
334 if owning_package is not None:
335 if prior_version is None: 335 ↛ 337line 335 didn't jump to line 337 because the condition on line 335 was never true
336 # Empty is allowed according to `man dpkg-maintscript-helper`
337 cmdline.append("")
338 cmdline.append(owning_package)
339 return cls(
340 tuple(cmdline),
341 definition_source,
342 )
344 @classmethod
345 def rm_conffile(
346 cls,
347 definition_source: AttributePath,
348 conffile: str,
349 prior_version: str | None = None,
350 owning_package: str | None = None,
351 ) -> Self:
352 cmdline = ["rm_conffile", conffile]
353 return cls._finish_cmd(
354 definition_source.path, cmdline, prior_version, owning_package
355 )
357 @classmethod
358 def mv_conffile(
359 cls,
360 definition_source: AttributePath,
361 old_conffile: str,
362 new_confile: str,
363 prior_version: str | None = None,
364 owning_package: str | None = None,
365 ) -> Self:
366 cmdline = ["mv_conffile", old_conffile, new_confile]
367 return cls._finish_cmd(
368 definition_source.path, cmdline, prior_version, owning_package
369 )
371 @classmethod
372 def symlink_to_dir(
373 cls,
374 definition_source: AttributePath,
375 pathname: str,
376 old_target: str,
377 prior_version: str | None = None,
378 owning_package: str | None = None,
379 ) -> Self:
380 cmdline = ["symlink_to_dir", pathname, old_target]
381 return cls._finish_cmd(
382 definition_source.path, cmdline, prior_version, owning_package
383 )
385 @classmethod
386 def dir_to_symlink(
387 cls,
388 definition_source: AttributePath,
389 pathname: str,
390 new_target: str,
391 prior_version: str | None = None,
392 owning_package: str | None = None,
393 ) -> Self:
394 cmdline = ["dir_to_symlink", pathname, new_target]
395 return cls._finish_cmd(
396 definition_source.path, cmdline, prior_version, owning_package
397 )
400class MaintscriptCondition(DebputyDispatchableType):
401 __slots__ = ("_maintscript_name", "_condition")
403 def __init__(self, maintscript_name: str, condition: str) -> None:
404 super().__init__()
405 if maintscript_name not in ALL_CONTROL_SCRIPTS:
406 raise ValueError(f"Invalid control script name: {maintscript_name}")
407 self._maintscript_name = maintscript_name
408 self._condition = condition
410 @property
411 def maintscript_name(self) -> str:
412 return self._maintscript_name
414 @classmethod
415 def on_purge(cls) -> typing.Self:
416 condition = '[ "$1" = "purge" ]'
417 return cls("postrm", condition)
419 @classmethod
420 def on_configure(cls) -> typing.Self:
421 condition = POSTINST_DEFAULT_CONDITION
422 return cls("postinst", condition)
424 @classmethod
425 def on_initial_install(cls) -> typing.Self:
426 condition = '[ "$1" = "configure" -a -z "$2" ]'
427 return cls("postinst", condition)
429 @classmethod
430 def on_upgrade(cls) -> typing.Self:
431 condition = '[ "$1" = "configure" -a -n "$2" ]'
432 return cls("postinst", condition)
434 @classmethod
435 def on_before_removal(cls) -> typing.Self:
436 condition = '[ "$1" = "remove" ]'
437 return cls("prerm", condition)
439 @classmethod
440 def on_after_removal(cls) -> typing.Self:
441 condition = '[ "$1" = "remove" ]'
442 return cls("postrm", condition)
444 def _apply_condition_to_script(
445 self,
446 condition: str,
447 run_snippet: str,
448 /,
449 indent: bool | None = None,
450 ) -> str:
451 if indent is None:
452 # We auto-determine this based on heredocs currently
453 indent = "<<" not in run_snippet
455 if indent:
456 run_snippet = "".join(" " + x for x in run_snippet.splitlines(True))
457 if not run_snippet.endswith("\n"):
458 run_snippet += "\n"
459 condition_line = f"if {condition}; then\n"
460 end_line = "fi\n"
461 return "".join((condition_line, run_snippet, end_line))
463 def _wrap_resolver(
464 self,
465 snippet_resolver: SnippetResolver,
466 *,
467 indent: bool | None = None,
468 ) -> SnippetResolver:
469 def _wrap() -> str:
470 unconditional_snippet = snippet_resolver.resolve()
471 return self._apply_condition_to_script(
472 self._condition,
473 unconditional_snippet,
474 indent=indent,
475 )
477 return SnippetResolver.lazy_snippet(_wrap)
479 def as_maintscript_snippet(
480 self,
481 definition_source: str,
482 snippet_resolver,
483 *,
484 snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE,
485 uses_debconf: bool = False,
486 indent: bool | None = None,
487 ) -> MaintscriptSnippetForNamedScript:
488 return MaintscriptSnippetForNamedScript(
489 definition_source,
490 self.maintscript_name,
491 self._wrap_resolver(snippet_resolver, indent=indent),
492 snippet_anchor=snippet_anchor,
493 uses_debconf=uses_debconf,
494 )