Coverage for src/hods/path.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 

19import os 

20import shutil 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25# TODO: use pathlib.Path # noqa: T101 

26class Path: 

27 """A path on disk.""" 

28 

29 @classmethod 

30 def _get_path(cls, path): 

31 """Get the path string from the given `Path` or string.""" 

32 if isinstance(path, Path): 

33 path = path.path 

34 return os.path.normpath(path) 

35 

36 def __init__(self, path): 

37 """Initialize path. 

38 

39 :param path: `str` - Absolute path 

40 """ 

41 logger.debug('Initializing %s %s', self.__class__.__name__, path) 

42 if not os.path.isabs(path): 

43 raise ValueError('{}: path argument must be an absolute path!'.format(self.__class__.__name__)) 

44 self.path = path 

45 

46 @property 

47 def dirname(self): 

48 """Shortcut for `os.path.dirname`.""" 

49 return os.path.dirname(self.path) 

50 

51 @property 

52 def basename(self): 

53 """Shortcut for `os.path.basename`. 

54 

55 Set this property to update the `path` attribute. 

56 """ 

57 return os.path.basename(self.path) 

58 

59 @basename.setter 

60 def basename(self, value): 

61 self.path = os.path.join(self.dirname, value) 

62 

63 def __str__(self): 

64 """String representation of the object.""" 

65 return str(self.path) 

66 

67 def __repr__(self): 

68 """Representation of the object.""" 

69 return '<{} {}>'.format(self.__class__.__name__, self.path) 

70 

71 def __hash__(self): 

72 """Hash of the object.""" 

73 return hash((self.__class__, self.path)) 

74 

75 def __eq__(self, other): 

76 """Return a `bool` whether the other object equals this one.""" 

77 if not isinstance(other, Path): 

78 return False 

79 return self.path == other.path 

80 

81 def __ne__(self, other): 

82 """Return a `bool` whether the other object does *not* equal this one.""" 

83 return not self.__eq__(other) 

84 

85 def exists(self): 

86 """Shortcut: Call and return `os.path.exists` with ``self.path``.""" 

87 return os.path.exists(self.path) 

88 

89 def islink(self): 

90 """Shortcut: Call and return `os.path.islink` with ``self.path``.""" 

91 return os.path.islink(self.path) 

92 

93 def isdir(self): 

94 """Shortcut: Call and return `os.path.isdir` with ``self.path``.""" 

95 return os.path.isdir(self.path) 

96 

97 def relpath(self, start): 

98 """Shortcut: Call and return `os.path.relpath` with ``self.path``.""" 

99 return os.path.relpath(self.path, self._get_path(start)) 

100 

101 def realpath(self): 

102 """Shortcut: Call and return `os.path.realpath` with ``self.path``.""" 

103 return os.path.realpath(self.path) 

104 

105 def move(self, destination): 

106 """Move this file or directory to the given destination. 

107 

108 :param destination: `str` - absolute path to move the file to 

109 :return: ``None`` 

110 """ 

111 destination_path = self._get_path(destination) 

112 if not os.path.isabs(destination_path): 

113 raise ValueError('{}.move() requires an absolute path.'.format(self.__class__.__name__)) 

114 logger.info('mv "%s" "%s"', self.path, destination_path) 

115 shutil.move(self.path, destination_path) 

116 

117 def _clean_copy_destination(self, destination): 

118 destination = self._get_path(destination) 

119 if not os.path.isabs(destination): 

120 raise ValueError('{}.copy() requires an absolute path.'.format(self.__class__.__name__)) 

121 if os.path.exists(destination): 

122 raise FileExistsError('{}.copy() path must not exist.'.format(self.__class__.__name__)) 

123 logger.info('cp "%s" "%s"', self.path, destination) 

124 return destination 

125 

126 

127class Directory(Path): 

128 """A path representing a directory on disk.""" 

129 

130 #: filenames to ignore in `listdir` 

131 ignore_basenames = [] 

132 

133 def rmtree(self): 

134 """Shortcut: Call `shutil.rmtree` with ``self.path``.""" 

135 logger.info('rm -r "%s"', self.path) 

136 shutil.rmtree(self.path) 

137 

138 def delete(self): 

139 """Call `rmtree` if `isdir`, otherwise :func`unlink`.""" 

140 self.rmtree() 

141 

142 def join(self, *parts): 

143 """Shortcut: Call and return `os.path.join` with ``self.path`` and the given parts.""" 

144 parts = [self._get_path(p) for p in parts] 

145 return os.path.join(self.path, *parts) 

146 

147 def listdir(self, dirs_only=False): 

148 """List files in this directory. 

149 

150 Call `os.listdir`, filter the basenames by `ignore_basenames` and return them. 

151 Return an empty list if the file does not exist or if its not a directory. 

152 

153 :param dirs_only: `bool` - Return only directories. Default: ``False`` 

154 """ 

155 try: 

156 listdir = os.listdir(self.path) 

157 except (FileNotFoundError, NotADirectoryError): 

158 return [] 

159 

160 basenames = [n for n in listdir if n not in self.ignore_basenames] 

161 if dirs_only: 

162 basenames = [n for n in listdir if os.path.isdir(self.join(n))] 

163 return basenames 

164 

165 def mkdir(self, mode=0o750): 

166 """Shortcut: Call `os.mkdir` and `os.chown` with ``self.path``.""" 

167 logger.info('mkdir "%s" with mode %s', self.path, mode) 

168 os.mkdir(self.path) 

169 os.chmod(self.path, mode) 

170 

171 def makedirs(self, mode=0o750): 

172 """Shortcut for `os.makedirs`.""" 

173 os.makedirs(self.path, mode=mode) 

174 

175 def copy(self, destination): 

176 """Copy this directory recursively to the given destination.""" 

177 destination = self._clean_copy_destination(destination) 

178 shutil.copytree(self.path, destination, ignore_dangling_symlinks=True) 

179 

180 

181class File(Path): 

182 """A path representing a file on disk.""" 

183 

184 def unlink(self): 

185 """Shortcut for `os.unlink`.""" 

186 logger.info('rm "%s"', self.path) 

187 os.unlink(self.path) 

188 

189 def symlink(self, destination): 

190 """Create a symlink at the given destination pointing to this file. 

191 

192 :param destination: `str` - Absolute path to point the symlink to 

193 """ 

194 destination_path = self._get_path(destination) 

195 os.symlink(self.path, destination_path) 

196 

197 def delete(self): 

198 """Call `rmtree` if `isdir`, otherwise :func`unlink`.""" 

199 self.unlink() 

200 

201 def _read(self, path): 

202 """Read and return the given paths contents. 

203 

204 :param path: `str` - The path to read from. 

205 :return: `str` - The files contents. 

206 """ 

207 path = self._get_path(path) 

208 logger.debug('reading "%s"', path) 

209 with open(path) as f: 

210 return f.read() 

211 

212 def read(self): 

213 """Read and return the file contents. 

214 

215 :return: `str` - The file contents as string. 

216 """ 

217 return self._read(self) 

218 

219 def _write(self, path, content): 

220 """Write the given content to the given path. 

221 

222 :param path: `str` - The path to write to 

223 :param content: `str` - The content to write 

224 """ 

225 path = self._get_path(path) 

226 logger.debug('writing "%s"', path) 

227 with open(path, 'w') as f: 

228 f.write(content) 

229 

230 def write(self, content): 

231 """Write the given content into the file. 

232 

233 :param content: `str` - The string to write to the file. 

234 """ 

235 self._write(self, content) 

236 

237 def copy(self, destination): 

238 """Copy this file to the given destination.""" 

239 destination = self._clean_copy_destination(destination) 

240 os.makedirs(os.path.dirname(destination), exist_ok=True) 

241 shutil.copy2(self.path, destination, follow_symlinks=True)