Coverage for src/debputy/maintscript_snippet.py: 64%

1import collections 

2import dataclasses 

3import enum 

4import typing 

5from typing import Self, TYPE_CHECKING 

6from collections.abc import Sequence, Iterable 

7 

8import debian.debian_support 

9 

10from debputy.manifest_parser.tagging_types import DebputyDispatchableType 

11from debputy.manifest_parser.util import AttributePath 

12from debputy.substitution import Substitution 

13from debputy.util import POSTINST_DEFAULT_CONDITION 

14 

15if TYPE_CHECKING: 

16 from debputy.plugins.debputy.binary_package_rules import ( 

17 MaintainerProvidedMaintscriptSnippetContainer, 

18 ) 

19 

20# The standard ctrl scripts from `dpkg` (`config` is from `debconf`) 

21DPKG_DEB_CONTROL_SCRIPTS = frozenset( 

22 { 

23 "preinst", 

24 "prerm", 

25 "postinst", 

26 "postrm", 

27 } 

28) 

29# all control scripts for debs and udebs respectively. 

30ALL_DEB_CONTROL_SCRIPTS = DPKG_DEB_CONTROL_SCRIPTS | {"config"} 

31ALL_UDEB_CONTROL_SCRIPTS = frozenset( 

32 { 

33 "postinst", 

34 "menutest", 

35 "isinstallable", 

36 "config", 

37 } 

38) 

39ALL_CONTROL_SCRIPTS = DPKG_DEB_CONTROL_SCRIPTS | ALL_UDEB_CONTROL_SCRIPTS 

40# TODO: Unclear what is missing to support the remaining `udeb` scripts. 

41# 

42# Note depending on support, this might change into scripts that we do something useful 

43# for vs. scripts we just handle as a PPF. 

44SUPPORTED_UDEB_SCRIPTS = frozenset({"postinst"}) 

45 

46 

47@dataclasses.dataclass(slots=True, frozen=True) 

48class MaintscriptForBinary: 

49 name: str 

50 

51 

52class SnippetAnchor(enum.StrEnum): 

53 CUSTOM_SCRIPT_PLACEHOLDER = "custom-script-placeholder" 

54 """ 

55 Placeholder. 

56 

57 Custom scripts should not have a specific place but rather a "before X/after X" logic.  

58 """ 

59 USER_CREATION = "user-creation" 

60 CONFIGURATION_MANAGEMENT = "configuration-management" 

61 # TODO: Rename before it becomes public 

62 _BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE = ( 

63 "between-configuration-maagement-and-service" 

64 ) 

65 SERVICE = "service" 

66 

67 

68class SnippetResolver: 

69 

70 def _resolve_snippet(self) -> str: 

71 raise NotImplementedError 

72 

73 def resolve(self) -> str: 

74 snippet = self._resolve_snippet() 

75 if not snippet.endswith("\n"): 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 return snippet + "\n" 

77 return snippet 

78 

79 def apply_manifest_substitution( 

80 self, 

81 definition_source: str, 

82 substitution: Substitution, 

83 ) -> "SnippetResolver": 

84 return SubstitutingSnippetResolver(self, definition_source, substitution) 

85 

86 @classmethod 

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

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

89 

90 @classmethod 

91 def lazy_snippet_no_manifest_substitution( 

92 cls, 

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

94 ) -> "SnippetResolver": 

95 return LazySnippetResolver(impl) 

96 

97 @classmethod 

98 def lazy_snippet( 

99 cls, 

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

101 ) -> "SnippetResolver": 

102 return LazySnippetResolver(impl) 

103 

104 

105class ConstantSnippetResolver(SnippetResolver): 

106 def __init__( 

107 self, snippet: str, definition_source: str, substitution: Substitution | None 

108 ) -> None: 

109 self._snippet = snippet 

110 self._definition_source = definition_source 

111 self._substitution = substitution 

112 

113 @property 

114 def provides_substitution(self) -> bool: 

115 return self._substitution is not None 

116 

117 def _resolve_snippet(self) -> str: 

118 template = self._snippet 

119 substitution = self._substitution 

120 if substitution: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 return substitution.substitute( 

122 template, 

123 self._definition_source, 

124 ) 

125 return template 

126 

127 

128class CachingSnippetResolver(SnippetResolver): 

129 

130 def __init__(self) -> None: 

131 self._resolved: str | None = None 

132 

133 def _resolve_snippet_uncached(self) -> str: 

134 raise NotImplementedError 

135 

136 def _resolve_snippet(self) -> str: 

137 resolved = self._resolved 

138 if resolved is not None: 

139 return resolved 

140 resolved = self._resolve_snippet_uncached() 

141 self._resolved = resolved 

142 return resolved 

143 

144 

145class LazySnippetResolver(CachingSnippetResolver): 

146 

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

148 super().__init__() 

149 self._resolver = resolver 

150 

151 def _resolve_snippet_uncached(self) -> str: 

152 return self._resolver() 

153 

154 

155class SubstitutingSnippetResolver(CachingSnippetResolver): 

156 

157 def __init__( 

158 self, 

159 wrapped_resolver: SnippetResolver, 

160 definition_source: str, 

161 substitution: Substitution, 

162 ) -> None: 

163 super().__init__() 

164 self._wrapped_resolver = wrapped_resolver 

165 self._definition_source = definition_source 

166 self._substitution = substitution 

167 

168 def _resolve_snippet_uncached(self) -> str: 

169 template = self._wrapped_resolver.resolve() 

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

171 

172 

173@dataclasses.dataclass(slots=True, frozen=True) 

174class UnboundMaintscriptSnippet: 

175 definition_source: str 

176 snippet: SnippetResolver 

177 snippet_anchor: SnippetAnchor = ( 

178 SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE 

179 ) 

180 uses_debconf: bool = False 

181 

182 def script_content(self) -> str: 

183 snippet = self.snippet.resolve() 

184 lines = [ 

185 f"# Snippet source: {self.definition_source}\n", 

186 snippet, 

187 ] 

188 return "".join(lines) 

189 

190 

191@dataclasses.dataclass(slots=True, frozen=True) 

192class MaintscriptSnippetForNamedScript: 

193 definition_source: str 

194 maintscript_name: str 

195 snippet: SnippetResolver 

196 snippet_anchor: SnippetAnchor = ( 

197 SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE 

198 ) 

199 uses_debconf: bool = False 

200 

201 def apply_to_container( 

202 self, 

203 maintscript_container: "PackageMaintscriptSnippetContainer", 

204 ) -> None: 

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

206 

207 def script_content(self) -> str: 

208 snippet = self.snippet.resolve() 

209 lines = [ 

210 f"# Snippet source: {self.definition_source}\n", 

211 snippet, 

212 ] 

213 return "".join(lines) 

214 

215 

216MaintscriptSnippet: typing.TypeAlias = ( 

217 UnboundMaintscriptSnippet | MaintscriptSnippetForNamedScript 

218) 

219 

220 

221class MaintscriptSnippetContainer: 

222 def __init__(self, script_name: str) -> None: 

223 self._script_name = script_name 

224 self._snippets_by_order: dict[SnippetAnchor, list[MaintscriptSnippet]] = {} 

225 

226 def copy(self) -> "MaintscriptSnippetContainer": 

227 instance = self.__class__(self._script_name) 

228 instance._snippets_by_order = self._snippets_by_order.copy() 

229 return instance 

230 

231 def append(self, maintscript_snippet: MaintscriptSnippet) -> None: 

232 assert maintscript_snippet.snippet_anchor 

233 if ( 233 ↛ 237line 233 didn't jump to line 237 because the condition on line 233 was never true

234 isinstance(maintscript_snippet, MaintscriptSnippetForNamedScript) 

235 and maintscript_snippet.maintscript_name != self._script_name 

236 ): 

237 raise ValueError( 

238 f"Cannot apply snippet for {maintscript_snippet.maintscript_name}" 

239 f" to {self._script_name}" 

240 ) 

241 if maintscript_snippet.snippet_anchor not in self._snippets_by_order: 

242 self._snippets_by_order[maintscript_snippet.snippet_anchor] = [] 

243 self._snippets_by_order[maintscript_snippet.snippet_anchor].append( 

244 maintscript_snippet 

245 ) 

246 

247 def all_snippets(self) -> Iterable[MaintscriptSnippet]: 

248 for sa in SnippetAnchor: 

249 if sa not in self._snippets_by_order: 

250 continue 

251 yield from self._snippets_by_order[sa] 

252 

253 def needs_debconf(self) -> bool: 

254 return any( 

255 s.uses_debconf 

256 for snippet_lists in self._snippets_by_order.values() 

257 for s in snippet_lists 

258 ) 

259 

260 def generate_snippet( 

261 self, 

262 *, 

263 tool_with_version: str | None = None, 

264 snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE, 

265 reverse: bool = False, 

266 ) -> str | None: 

267 snippets = self._snippets_by_order.get(snippet_anchor) 

268 if not snippets: 

269 return None 

270 if reverse: 

271 resulting_snippets: collections.abc.Iterable[MaintscriptSnippet] = reversed( 

272 snippets 

273 ) 

274 else: 

275 resulting_snippets = snippets 

276 inner_content = "".join(s.script_content() for s in resulting_snippets) 

277 

278 if not inner_content: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true

279 return None 

280 

281 if tool_with_version: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true

282 return ( 

283 f"# Automatically added by {tool_with_version}\n" 

284 + inner_content 

285 + "# End automatically added section" 

286 ) 

287 return inner_content 

288 

289 

290class PackageMaintscriptSnippetContainer( 

291 collections.abc.Mapping[str, MaintscriptSnippetContainer] 

292): 

293 

294 def __init__(self, supported_scripts: frozenset[str] | None = None) -> None: 

295 self._supported_scripts = ( 

296 supported_scripts if supported_scripts is not None else ALL_CONTROL_SCRIPTS 

297 ) 

298 self._script_containers: dict[str, MaintscriptSnippetContainer] = {} 

299 

300 def __len__(self): 

301 return len(self._script_containers) 

302 

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

304 yield from self._script_containers 

305 

306 def __getitem__(self, key: str, /) -> MaintscriptSnippetContainer: 

307 if key not in self._supported_scripts: 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true

308 raise KeyError(key) 

309 try: 

310 return self._script_containers[key] 

311 except KeyError: 

312 pass 

313 c = MaintscriptSnippetContainer(key) 

314 self._script_containers[key] = c 

315 return c 

316 

317 def apply_maintainer_provided_snippets( 

318 self, 

319 maintainer_provided_maintscript_snippets: "MaintainerProvidedMaintscriptSnippetContainer", 

320 ) -> None: 

321 for snippet in maintainer_provided_maintscript_snippets.snippets: 

322 snippet.apply_to_container(self) 

323 

324 

325class DpkgMaintscriptHelperCommand(DebputyDispatchableType): 

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

327 

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

329 super().__init__() 

330 self.cmdline = cmdline 

331 self.definition_source = definition_source 

332 

333 @classmethod 

334 def _finish_cmd( 

335 cls, 

336 definition_source: str, 

337 cmdline: list[str], 

338 prior_version: str | None, 

339 owning_package: str | None, 

340 ) -> Self: 

341 if prior_version is not None: 

342 cmdline.append(prior_version) 

343 if owning_package is not None: 

344 if prior_version is None: 344 ↛ 346line 344 didn't jump to line 346 because the condition on line 344 was never true

345 # Empty is allowed according to `man dpkg-maintscript-helper` 

346 cmdline.append("") 

347 cmdline.append(owning_package) 

348 return cls( 

349 tuple(cmdline), 

350 definition_source, 

351 ) 

352 

353 @classmethod 

354 def rm_conffile( 

355 cls, 

356 definition_source: AttributePath, 

357 conffile: str, 

358 prior_version: str | None = None, 

359 owning_package: str | None = None, 

360 ) -> Self: 

361 cmdline = ["rm_conffile", conffile] 

362 return cls._finish_cmd( 

363 definition_source.path, cmdline, prior_version, owning_package 

364 ) 

365 

366 @classmethod 

367 def mv_conffile( 

368 cls, 

369 definition_source: AttributePath, 

370 old_conffile: str, 

371 new_confile: str, 

372 prior_version: str | None = None, 

373 owning_package: str | None = None, 

374 ) -> Self: 

375 cmdline = ["mv_conffile", old_conffile, new_confile] 

376 return cls._finish_cmd( 

377 definition_source.path, cmdline, prior_version, owning_package 

378 ) 

379 

380 @classmethod 

381 def symlink_to_dir( 

382 cls, 

383 definition_source: AttributePath, 

384 pathname: str, 

385 old_target: str, 

386 prior_version: str | None = None, 

387 owning_package: str | None = None, 

388 ) -> Self: 

389 cmdline = ["symlink_to_dir", pathname, old_target] 

390 return cls._finish_cmd( 

391 definition_source.path, cmdline, prior_version, owning_package 

392 ) 

393 

394 @classmethod 

395 def dir_to_symlink( 

396 cls, 

397 definition_source: AttributePath, 

398 pathname: str, 

399 new_target: str, 

400 prior_version: str | None = None, 

401 owning_package: str | None = None, 

402 ) -> Self: 

403 cmdline = ["dir_to_symlink", pathname, new_target] 

404 return cls._finish_cmd( 

405 definition_source.path, cmdline, prior_version, owning_package 

406 ) 

407 

408 

409class MaintscriptCondition(DebputyDispatchableType): 

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

411 

412 def __init__(self, maintscript_name: str, condition: str) -> None: 

413 super().__init__() 

414 if maintscript_name not in ALL_CONTROL_SCRIPTS: 

415 raise ValueError(f"Invalid control script name: {maintscript_name}") 

416 self._maintscript_name = maintscript_name 

417 self._condition = condition 

418 

419 @property 

420 def maintscript_name(self) -> str: 

421 return self._maintscript_name 

422 

423 @classmethod 

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

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

426 return cls("postrm", condition) 

427 

428 @classmethod 

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

430 condition = POSTINST_DEFAULT_CONDITION 

431 return cls("postinst", condition) 

432 

433 @classmethod 

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

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

436 return cls("postinst", condition) 

437 

438 @classmethod 

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

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

441 return cls("postinst", condition) 

442 

443 @classmethod 

444 def on_upgrade_from( 

445 cls, 

446 from_version: debian.debian_support.Version, 

447 ) -> typing.Self: 

448 condition = ( 

449 f'[ "$1" = "configure" ] && dpkg --compare-versions le-nl "{from_version}"' 

450 ) 

451 return cls("postinst", condition) 

452 

453 @classmethod 

454 def on_before_upgrade(cls) -> typing.Self: 

455 condition = '[ "$1" = "upgrade" ]' 

456 return cls("preinst", condition) 

457 

458 @classmethod 

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

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

461 return cls("prerm", condition) 

462 

463 @classmethod 

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

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

466 return cls("postrm", condition) 

467 

468 @classmethod 

469 def on_unconditionally_in_script(cls, script_name: str) -> typing.Self: 

470 return cls(script_name, "true") 

471 

472 def _apply_condition_to_script( 

473 self, 

474 condition: str, 

475 run_snippet: str, 

476 /, 

477 indent: bool | None = None, 

478 ) -> str: 

479 if indent is None: 

480 # We auto-determine this based on heredocs currently 

481 indent = "<<" not in run_snippet 

482 

483 if indent: 

484 run_snippet = "".join(" " + x for x in run_snippet.splitlines(True)) 

485 if not run_snippet.endswith("\n"): 

486 run_snippet += "\n" 

487 condition_line = f"if {condition}; then\n" 

488 end_line = "fi\n" 

489 return "".join((condition_line, run_snippet, end_line)) 

490 

491 def _wrap_resolver( 

492 self, 

493 snippet_resolver: SnippetResolver, 

494 *, 

495 indent: bool | None = None, 

496 ) -> SnippetResolver: 

497 def _wrap() -> str: 

498 unconditional_snippet = snippet_resolver.resolve() 

499 return self._apply_condition_to_script( 

500 self._condition, 

501 unconditional_snippet, 

502 indent=indent, 

503 ) 

504 

505 return SnippetResolver.lazy_snippet(_wrap) 

506 

507 def as_maintscript_snippet( 

508 self, 

509 definition_source: str, 

510 snippet_resolver, 

511 *, 

512 snippet_anchor: SnippetAnchor = SnippetAnchor._BETWEEN_CONFIGURATION_MANAGEMENT_AND_SERVICE, 

513 uses_debconf: bool = False, 

514 indent: bool | None = None, 

515 ) -> MaintscriptSnippetForNamedScript: 

516 return MaintscriptSnippetForNamedScript( 

517 definition_source, 

518 self.maintscript_name, 

519 self._wrap_resolver(snippet_resolver, indent=indent), 

520 snippet_anchor=snippet_anchor, 

521 uses_debconf=uses_debconf, 

522 )