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

1import collections 

2import dataclasses 

3import functools 

4from typing import Literal, Self 

5from collections.abc import Sequence, Iterable 

6 

7from debputy.manifest_parser.tagging_types import DebputyDispatchableType 

8from debputy.manifest_parser.util import AttributePath 

9 

10STD_CONTROL_SCRIPTS = frozenset( 

11 { 

12 "preinst", 

13 "prerm", 

14 "postinst", 

15 "postrm", 

16 } 

17) 

18UDEB_CONTROL_SCRIPTS = frozenset( 

19 { 

20 "postinst", 

21 "menutest", 

22 "isinstallable", 

23 } 

24) 

25ALL_CONTROL_SCRIPTS = STD_CONTROL_SCRIPTS | UDEB_CONTROL_SCRIPTS | {"config"} 

26 

27 

28class SnippetResolver: 

29 

30 def _resolve_snippet(self) -> str: 

31 raise NotImplementedError 

32 

33 def resolve(self) -> str: 

34 snippet = self._resolve_snippet() 

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

36 return snippet + "\n" 

37 return snippet 

38 

39 @classmethod 

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

41 return ConstantSnippetResolver(snippet) 

42 

43 @classmethod 

44 def lazy_snippet_template( 

45 cls, impl: collections.abc.Callable[[], str] 

46 ) -> "SnippetResolver": 

47 return LazySnippetResolver(impl) 

48 

49 

50class ConstantSnippetResolver(SnippetResolver): 

51 def __init__(self, snippet: str) -> None: 

52 self._snippet = snippet 

53 

54 def _resolve_snippet(self) -> str: 

55 return self._snippet 

56 

57 

58class LazySnippetResolver(SnippetResolver): 

59 

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

61 self._resolver = resolver 

62 

63 @functools.lru_cache 

64 def _resolve_snippet(self) -> str: 

65 return self._resolver() 

66 

67 

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

69class MaintscriptSnippet: 

70 definition_source: str 

71 snippet: SnippetResolver 

72 snippet_order: Literal["service"] | None = None 

73 uses_debconf: bool = False 

74 

75 def script_content(self) -> str: 

76 snippet = self.snippet.resolve() 

77 lines = [ 

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

79 snippet, 

80 ] 

81 return "".join(lines) 

82 

83 

84class MaintscriptSnippetContainer: 

85 def __init__(self) -> None: 

86 self._generic_snippets: list[MaintscriptSnippet] = [] 

87 self._snippets_by_order: dict[Literal["service"], list[MaintscriptSnippet]] = {} 

88 

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

90 instance = self.__class__() 

91 instance._generic_snippets = self._generic_snippets.copy() 

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

93 return instance 

94 

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

96 if maintscript_snippet.snippet_order is None: 

97 self._generic_snippets.append(maintscript_snippet) 

98 else: 

99 if maintscript_snippet.snippet_order not in self._snippets_by_order: 99 ↛ 101line 99 didn't jump to line 101 because the condition on line 99 was always true

100 self._snippets_by_order[maintscript_snippet.snippet_order] = [] 

101 self._snippets_by_order[maintscript_snippet.snippet_order].append( 

102 maintscript_snippet 

103 ) 

104 

105 def has_content(self, snippet_order: Literal["service"] | None = None) -> bool: 

106 if snippet_order is None: 

107 return bool(self._generic_snippets) 

108 if snippet_order not in self._snippets_by_order: 

109 return False 

110 return bool(self._snippets_by_order[snippet_order]) 

111 

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

113 yield from self._generic_snippets 

114 for snippets in self._snippets_by_order.values(): 

115 yield from snippets 

116 

117 def needs_debconf(self) -> bool: 

118 return any(s.uses_debconf for s in self._generic_snippets) or any( 

119 s.uses_debconf 

120 for snippet_lists in self._snippets_by_order.values() 

121 for s in snippet_lists 

122 ) 

123 

124 def generate_snippet( 

125 self, 

126 tool_with_version: str | None = None, 

127 snippet_order: Literal["service"] | None = None, 

128 reverse: bool = False, 

129 ) -> str | None: 

130 inner_content = "" 

131 if snippet_order is None: 

132 snippets = ( 

133 reversed(self._generic_snippets) if reverse else self._generic_snippets 

134 ) 

135 inner_content = "".join(s.script_content() for s in snippets) 

136 elif snippet_order in self._snippets_by_order: 

137 snippets = self._snippets_by_order[snippet_order] 

138 if reverse: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 snippets = reversed(snippets) 

140 inner_content = "".join(s.script_content() for s in snippets) 

141 

142 if not inner_content: 

143 return None 

144 

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

146 return ( 

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

148 + inner_content 

149 + "# End automatically added section" 

150 ) 

151 return inner_content 

152 

153 

154class DpkgMaintscriptHelperCommand(DebputyDispatchableType): 

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

156 

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

158 super().__init__() 

159 self.cmdline = cmdline 

160 self.definition_source = definition_source 

161 

162 @classmethod 

163 def _finish_cmd( 

164 cls, 

165 definition_source: str, 

166 cmdline: list[str], 

167 prior_version: str | None, 

168 owning_package: str | None, 

169 ) -> Self: 

170 if prior_version is not None: 

171 cmdline.append(prior_version) 

172 if owning_package is not None: 

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

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

175 cmdline.append("") 

176 cmdline.append(owning_package) 

177 return cls( 

178 tuple(cmdline), 

179 definition_source, 

180 ) 

181 

182 @classmethod 

183 def rm_conffile( 

184 cls, 

185 definition_source: AttributePath, 

186 conffile: str, 

187 prior_version: str | None = None, 

188 owning_package: str | None = None, 

189 ) -> Self: 

190 cmdline = ["rm_conffile", conffile] 

191 return cls._finish_cmd( 

192 definition_source.path, cmdline, prior_version, owning_package 

193 ) 

194 

195 @classmethod 

196 def mv_conffile( 

197 cls, 

198 definition_source: AttributePath, 

199 old_conffile: str, 

200 new_confile: str, 

201 prior_version: str | None = None, 

202 owning_package: str | None = None, 

203 ) -> Self: 

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

205 return cls._finish_cmd( 

206 definition_source.path, cmdline, prior_version, owning_package 

207 ) 

208 

209 @classmethod 

210 def symlink_to_dir( 

211 cls, 

212 definition_source: AttributePath, 

213 pathname: str, 

214 old_target: str, 

215 prior_version: str | None = None, 

216 owning_package: str | None = None, 

217 ) -> Self: 

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

219 return cls._finish_cmd( 

220 definition_source.path, cmdline, prior_version, owning_package 

221 ) 

222 

223 @classmethod 

224 def dir_to_symlink( 

225 cls, 

226 definition_source: AttributePath, 

227 pathname: str, 

228 new_target: str, 

229 prior_version: str | None = None, 

230 owning_package: str | None = None, 

231 ) -> Self: 

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

233 return cls._finish_cmd( 

234 definition_source.path, cmdline, prior_version, owning_package 

235 )