Coverage for src/hods/delta.py: 100.00%

1"""hods - home directory synchronization. 

2 

3Copyright (C) 2016-2020 Mathias Stelzer <knoppo@rolln.de> 

4 

5hods is free software: you can redistribute it and/or modify 

6it under the terms of the GNU General Public License as published by 

7the Free Software Foundation, either version 3 of the License, or 

8(at your option) any later version. 

9 

10hods is distributed in the hope that it will be useful, 

11but WITHOUT ANY WARRANTY; without even the implied warranty of 

12MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13GNU General Public License for more details. 

14 

15You should have received a copy of the GNU General Public License 

16along with this program. If not, see <http://www.gnu.org/licenses/>. 

17""" 

18import logging 

19 

20from hods.node import DirectoryNode, FileNode, RootDirectoryNode 

21from hods.path import Directory, File 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class SameDestinationError(Exception): 

27 """Exception for 2 SourceFile instances with the same destination.""" 

28 

29 def __init__(self, first, second, *args, **kwargs): 

30 """Initialize error. 

31 

32 :param first: first sourcefile object 

33 :param second: second sourcefile object 

34 :param args: more Exception args 

35 :param kwargs: more Exception kwargs 

36 """ 

37 self.first = first 

38 self.second = second 

39 msg = '"{first}" and "{second}" are both trying to create a symlink at "{destination}".'.format( 

40 first=first.relative_config_path, 

41 second=second.relative_config_path, 

42 destination=first.destination_path) 

43 super().__init__(msg, *args, **kwargs) 

44 

45 

46class DeltaBuilder: 

47 """Compare a tree with the home directory.""" 

48 

49 def __init__(self, config): 

50 """Initialize home directory update.Initialize home directory update. 

51 

52 :param config: `hods.config.base.Config` instance 

53 """ 

54 self.config = config 

55 

56 def build(self): 

57 """Build a new `Delta` instance and return it. 

58 

59 This compares the source destination tree with the current home directory and stores 

60 all changes and necessary actions in a `Delta` instance. 

61 """ 

62 delta = Delta(self.config) 

63 

64 if not delta.exists(): 

65 msg = 'Your home directory does not exist! ({})'.format(delta.path) 

66 logger.error(msg) 

67 raise FileNotFoundError(msg) 

68 

69 for sourcefile in self.config.tree.collect_sourcefiles(): 

70 self.add_sourcefile(delta, sourcefile) 

71 

72 for sourcefile in self.config.settings.tree.collect_sourcefiles(): 

73 if delta.find(sourcefile.destination_path) is None: 

74 # file is not in the config anymore -> it was removed. 

75 self.delete_sourcefile(delta, sourcefile) 

76 return delta 

77 

78 def add_sourcefile(self, delta, sourcefile): 

79 """Add a source file to this delta.""" 

80 # check if there already is a file or directory for the destination 

81 existing = delta.find(sourcefile.destination_path) 

82 if existing: 

83 # If it has a sourcefile assigned we have a collision! This is 

84 # a configuration error and should be forwarded to the user. 

85 existing_sourcefile = getattr(existing, 'sourcefile', None) 

86 if existing_sourcefile is not None: 

87 raise SameDestinationError(sourcefile, existing_sourcefile) 

88 # otherwise replace it (detach and use its parent) 

89 existing.detach() 

90 parent = existing.parent 

91 else: 

92 parent = delta.find(sourcefile.destination_dirname, create=True, cls='dir') 

93 

94 if sourcefile.mode == 'link': 

95 Symlink(parent, sourcefile) 

96 elif sourcefile.mode == 'template': 

97 Template(parent, sourcefile) 

98 else: 

99 m = 'Unable to add {} with invalid mode {}'.format(repr(sourcefile), repr(sourcefile.mode)) 

100 raise ValueError(m) 

101 

102 def delete_sourcefile(self, delta, sourcefile): 

103 """Add a deleted file to this delta.""" 

104 parent = delta.find(sourcefile.destination_dirname, create=True, 

105 node_cls=DeletedDirectory, cls=DeletedDirectory) 

106 

107 if sourcefile.mode == 'link': 

108 DeletedSymlink(parent, sourcefile) 

109 elif sourcefile.mode == 'template': 

110 DeletedTemplate(parent, sourcefile) 

111 else: 

112 m = 'Unable to delete {} with invalid mode "{}"'.format(repr(sourcefile), sourcefile.mode) 

113 raise ValueError(m) 

114 

115 

116class ApplyError(Exception): 

117 """Raised when applying a file or directory fails.""" 

118 

119 def __init__(self, node, action, message): 

120 """Initialize exception.""" 

121 self.node = node 

122 self.action = action 

123 self.message = message 

124 super().__init__('Failed to {} "{}": {}'.format(action, node.path, message)) 

125 

126 

127class Delta(RootDirectoryNode): 

128 """A home directory update. 

129 

130 Instances are created by `DeltaBuilder` and contain directories and files representing 

131 the destinations and necessary actions in a source tree. 

132 """ 

133 

134 ignore_basenames = ['.hods'] 

135 

136 def __init__(self, config): 

137 """Initialize the home directory tree.""" 

138 super().__init__(config.settings.home_path) 

139 self.config = config 

140 

141 def get_child_directory_class(self): 

142 """Return `DeltaDirectory` to instantiate child directories.""" 

143 return DeltaDirectory 

144 

145 def get_child_file_class(self): 

146 """Return `DeltaFile` to instantiate child files.""" 

147 return DeltaFile 

148 

149 def get_file_action(self, file): 

150 """Get the action for the given file and return it. 

151 

152 Overwrite this to intercept the update process of individual files. 

153 

154 :param file: `hods.update.BaseDeltaPath` - The file that will be updated 

155 :return: Callable or None - The action to update the given file 

156 """ 

157 return file.get_action() 

158 

159 def apply(self): 

160 """Apply this delta to the home directory.""" 

161 files = list(self.collect_children()) 

162 while files: # and directories 

163 f = files.pop(0) 

164 

165 action = self.get_file_action(f) 

166 if action is None: 

167 logger.debug('update %s: skip', f) 

168 continue 

169 

170 logger.info('update %s: %s', f, action) 

171 try: 

172 action() 

173 except OSError as e: 

174 raise ApplyError(f, action.__name__, str(e)) 

175 self.config.settings.tree.load_data(self.config.tree.as_dict()) 

176 logger.debug(str(self.config.settings.tree.features)) 

177 self.config.settings.save() 

178 

179 

180class DeltaNodeMixin: 

181 """Base file in the home directory.""" 

182 

183 label = None 

184 

185 def __init__(self, parent, path, attach=True): 

186 """Initialize HomeFile. 

187 

188 :param parent: `DeltaPath` instance - Parent directory 

189 :param path: `str` - basename 

190 :param attach: `bool` - attach the file to the parent (Default: ``True``) 

191 """ 

192 super().__init__(parent, path, attach=attach) 

193 self.home = self.root 

194 if self.label is None: 

195 self.label = self.__class__.__name__.lower() 

196 

197 @property 

198 def relative_home_path(self): 

199 """Return the relative path to the home directory.""" 

200 return self.relpath(self.home.path) 

201 

202 def get_action(self): 

203 """Get the required method to apply the delta.""" 

204 if self.is_applied(): 

205 return 

206 if self.islink() or self.exists(): 

207 return self.replace 

208 return self.create 

209 

210 def is_applied(self): 

211 """Overwrite in subclass and return the required action.""" 

212 raise NotImplementedError 

213 

214 def create(self): 

215 """Overwrite in subclass and return the required action.""" 

216 raise NotImplementedError 

217 

218 def replace(self): 

219 """Overwrite in subclass and return the required action.""" 

220 self.delete() 

221 self.create() 

222 

223 def delete(self): 

224 """Delete the file or directory at this path.""" 

225 if not self.islink() and self.isdir(): 

226 Directory(self.path).delete() 

227 else: 

228 File(self.path).delete() 

229 

230 def get_label_for_action(self, action): 

231 """Get descriptive label for given action.""" 

232 if action == 'create': 

233 return 'create {}'.format(self.label) 

234 if action == 'replace': 

235 return 'replace with {}'.format(self.label) 

236 return action 

237 

238 def get_action_label(self): 

239 """Get descriptive label for current action.""" 

240 action = self.get_action() 

241 if action is None: 

242 return '' 

243 return self.get_label_for_action(action.__name__) 

244 

245 

246class DeltaFile(DeltaNodeMixin, FileNode): 

247 """A simple file in the home directory.""" 

248 

249 pass 

250 

251 

252class DeltaDirectory(DeltaNodeMixin, DirectoryNode): 

253 """A directory in the home directory.""" 

254 

255 child_file_class = DeltaFile 

256 

257 label = 'directory' 

258 

259 def get_child_directory_class(self): 

260 """Return the child class for the given basename.""" 

261 return DeltaDirectory 

262 

263 def is_applied(self): 

264 """Check whether the directory exists.""" 

265 # if the parent gets re-created we need to re-create this child, too 

266 if not self.parent.is_root and not self.parent.is_applied(): 

267 return False 

268 return not self.islink() and self.isdir() 

269 

270 def create(self): 

271 """Create the directory node.""" 

272 self.mkdir() 

273 

274 

275class DestinationMixin(DeltaNodeMixin): 

276 """A file in the home directory with an assigned source file.""" 

277 

278 def __init__(self, parent, sourcefile, attach=True): 

279 """Initialize source file delta node. node. 

280 

281 :param parent: `BaseDeltaPath` object - Parent node. 

282 :param sourcefile: `hods.config.sourcefile.SourceFile` - The 

283 source file to apply. 

284 :param attach: `bool` - Attach the node to the tree. 

285 (Default: True) 

286 """ 

287 super().__init__(parent, sourcefile.destination_basename, attach=attach) 

288 self.sourcefile = sourcefile 

289 

290 

291class Symlink(DestinationMixin, DeltaFile): 

292 """Symlink in the home directory.""" 

293 

294 def is_linked(self): 

295 """Whether this file is a symlink pointing to the assigned `SourceFile`.""" 

296 if not self.islink(): 

297 return False 

298 return self.realpath() == self.sourcefile.path 

299 

300 def is_applied(self): 

301 """Check whether the symlink is already created.""" 

302 return self.is_linked() 

303 

304 def create(self): 

305 """Symlink the source files destination to it.""" 

306 self.sourcefile.symlink(self.path) 

307 

308 

309class Template(DestinationMixin, DeltaFile): 

310 """Template in the home directory.""" 

311 

312 def __init__(self, parent, sourcefile, attach=True): 

313 """Initialize template.""" 

314 super().__init__(parent, sourcefile, attach=attach) 

315 self.rendered = sourcefile.render() 

316 

317 def is_applied(self): 

318 """Check whether the template is rendered.""" 

319 if self.isdir() or self.islink(): 

320 return False 

321 try: 

322 text = self.read() 

323 except FileNotFoundError: 

324 return False 

325 return self.rendered == text 

326 

327 def create(self): 

328 """Render and write the template.""" 

329 self.write(self.rendered) 

330 

331 

332class DeletedDeltaNodeMixin: 

333 """Mixin for destination file classes of a deleted source file.""" 

334 

335 def get_action(self): 

336 """Return 'delete' action if file has not changed since last update.""" 

337 if self.is_applied(): 

338 return self.delete 

339 

340 

341class DeletedDirectory(DeletedDeltaNodeMixin, DeltaDirectory): 

342 """Deleted directory in the home directory.""" 

343 

344 def will_be_empty(self): 

345 """Check whether the directory can be safely deleted.""" 

346 filenames = self.listdir() 

347 for child in self.children[:]: 

348 if child.get_action() == child.delete: 

349 filenames.remove(child.basename) 

350 return not bool(filenames) 

351 

352 def get_action(self): 

353 """Get the required method to apply the delta.""" 

354 action = super().get_action() 

355 if action is None: 

356 return 

357 if self.will_be_empty(): 

358 return action 

359 

360 

361class DeletedSymlink(DeletedDeltaNodeMixin, Symlink): 

362 """Deleted symlink in the home directory.""" 

363 

364 pass 

365 

366 

367class DeletedTemplate(DeletedDeltaNodeMixin, Template): 

368 """Deleted template in the home directory.""" 

369 

370 pass