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

1import collections 

2import dataclasses 

3import enum 

4import typing 

5from typing import Self, TYPE_CHECKING 

6from collections.abc import Sequence, Iterable 

7 

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 

12 

13if TYPE_CHECKING: 

14 from debputy.plugins.debputy.binary_package_rules import ( 

15 MaintainerProvidedMaintscriptSnippetContainer, 

16 ) 

17 

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"}) 

43 

44 

45class SnippetAnchor(enum.StrEnum): 

46 CUSTOM_SCRIPT_PLACEHOLDER = "custom-script-placeholder" 

47 """ 

48 Placeholder. 

49 

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" 

59 

60 

61class SnippetResolver: 

62 

63 def _resolve_snippet(self) -> str: 

64 raise NotImplementedError 

65 

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 

71 

72 def apply_manifest_substitution( 

73 self, 

74 definition_source: str, 

75 substitution: Substitution, 

76 ) -> "SnippetResolver": 

77 return SubstitutingSnippetResolver(self, definition_source, substitution) 

78 

79 @classmethod 

80 def snippet(cls, snippet: str) -> "SnippetResolver": 

81 return ConstantSnippetResolver(snippet, "IGNORED", None) 

82 

83 @classmethod 

84 def lazy_snippet_no_manifest_substitution( 

85 cls, 

86 impl: collections.abc.Callable[[], str], 

87 ) -> "SnippetResolver": 

88 return LazySnippetResolver(impl) 

89 

90 @classmethod 

91 def lazy_snippet( 

92 cls, 

93 impl: collections.abc.Callable[[], str], 

94 ) -> "SnippetResolver": 

95 return LazySnippetResolver(impl) 

96 

97 

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 

105 

106 @property 

107 def provides_substitution(self) -> bool: 

108 return self._substitution is not None 

109 

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 

119 

120 

121class CachingSnippetResolver(SnippetResolver): 

122 

123 def __init__(self) -> None: 

124 self._resolved: str | None = None 

125 

126 def _resolve_snippet_uncached(self) -> str: 

127 raise NotImplementedError 

128 

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 

136 

137 

138class LazySnippetResolver(CachingSnippetResolver): 

139 

140 def __init__(self, resolver: collections.abc.Callable[[], str]) -> None: 

141 super().__init__() 

142 self._resolver = resolver 

143 

144 def _resolve_snippet_uncached(self) -> str: 

145 return self._resolver() 

146 

147 

148class SubstitutingSnippetResolver(CachingSnippetResolver): 

149 

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 

160 

161 def _resolve_snippet_uncached(self) -> str: 

162 template = self._wrapped_resolver.resolve() 

163 return self._substitution.substitute(template, self._definition_source) 

164 

165 

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 

174 

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) 

182 

183 

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 

193 

194 def apply_to_container( 

195 self, 

196 maintscript_container: "PackageMaintscriptSnippetContainer", 

197 ) -> None: 

198 maintscript_container[self.maintscript_name].append(self) 

199 

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) 

207 

208 

209MaintscriptSnippet: typing.TypeAlias = ( 

210 UnboundMaintscriptSnippet | MaintscriptSnippetForNamedScript 

211) 

212 

213 

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]] = {} 

218 

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 

223 

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 ) 

239 

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] 

245 

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 ) 

252 

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) 

268 

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 

271 

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 

279 

280 

281class PackageMaintscriptSnippetContainer( 

282 collections.abc.Mapping[str, MaintscriptSnippetContainer] 

283): 

284 

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] = {} 

290 

291 def __len__(self): 

292 return len(self._script_containers) 

293 

294 def __iter__(self) -> collections.abc.Iterator[str]: 

295 yield from self._script_containers 

296 

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 

307 

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) 

314 

315 

316class DpkgMaintscriptHelperCommand(DebputyDispatchableType): 

317 __slots__ = ("cmdline", "definition_source") 

318 

319 def __init__(self, cmdline: Sequence[str], definition_source: str) -> None: 

320 super().__init__() 

321 self.cmdline = cmdline 

322 self.definition_source = definition_source 

323 

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 ) 

343 

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 ) 

356 

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 ) 

370 

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 ) 

384 

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 ) 

398 

399 

400class MaintscriptCondition(DebputyDispatchableType): 

401 __slots__ = ("_maintscript_name", "_condition") 

402 

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 

409 

410 @property 

411 def maintscript_name(self) -> str: 

412 return self._maintscript_name 

413 

414 @classmethod 

415 def on_purge(cls) -> typing.Self: 

416 condition = '[ "$1" = "purge" ]' 

417 return cls("postrm", condition) 

418 

419 @classmethod 

420 def on_configure(cls) -> typing.Self: 

421 condition = POSTINST_DEFAULT_CONDITION 

422 return cls("postinst", condition) 

423 

424 @classmethod 

425 def on_initial_install(cls) -> typing.Self: 

426 condition = '[ "$1" = "configure" -a -z "$2" ]' 

427 return cls("postinst", condition) 

428 

429 @classmethod 

430 def on_upgrade(cls) -> typing.Self: 

431 condition = '[ "$1" = "configure" -a -n "$2" ]' 

432 return cls("postinst", condition) 

433 

434 @classmethod 

435 def on_before_removal(cls) -> typing.Self: 

436 condition = '[ "$1" = "remove" ]' 

437 return cls("prerm", condition) 

438 

439 @classmethod 

440 def on_after_removal(cls) -> typing.Self: 

441 condition = '[ "$1" = "remove" ]' 

442 return cls("postrm", condition) 

443 

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 

454 

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)) 

462 

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 ) 

476 

477 return SnippetResolver.lazy_snippet(_wrap) 

478 

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 )