Coverage for src/hods/tui/config/settings.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 

20 

21import urwid 

22 

23from hods.config.template import load_context 

24from hods.tui.base.edit import EditWindow, ListBoxWindow, ObjectEditWindow, radio_buttons 

25from hods.tui.base.widgets import FocusMap 

26from hods.utils import clean_server, get_hostname, pw_user 

27 

28logger = logging.getLogger(__name__) 

29 

30 

31class VariableAddWindow(EditWindow): 

32 """A window widget to create sources (and features).""" 

33 

34 def __init__(self, app, existing_names, **kwargs): 

35 """Initialize window. 

36 

37 :param app: `hods.app.App` instance 

38 """ 

39 super().__init__( 

40 app, 

41 title='New Variable', 

42 height=4, 

43 **kwargs) 

44 

45 self.existing_names = existing_names 

46 self.edit_name = urwid.Edit(edit_text='') 

47 

48 def get_items(self): 

49 """Collect and return form widgets.""" 

50 yield urwid.Columns([ 

51 ('pack', urwid.Text('Name:')), 

52 urwid.AttrMap(self.edit_name, 'edit'), 

53 ], dividechars=1) 

54 

55 def clean(self): 

56 """Clean and return form data.""" 

57 name = self.edit_name.edit_text.strip() 

58 if not name: 

59 raise self.ValidationError('A name is required!') 

60 if name in self.existing_names: 

61 raise self.ValidationError('A variable with that name already exists!') 

62 return {'name': name} 

63 

64 

65class ContextWindow(ListBoxWindow): 

66 """Window to display the template context.""" 

67 

68 def __init__(self, app, overwrite, **kwargs): 

69 """Initialize window.""" 

70 super().__init__(app, title='Template Context', **kwargs) 

71 

72 context = dict(load_context(app.config.tree)) 

73 context.update(dict(overwrite)) 

74 max_key_width = max(map(len, context.keys())) 

75 self.key_width = min([20, max_key_width]) 

76 self.context = context 

77 

78 def get_items(self): 

79 """Collect the widgets to display.""" 

80 for index, key in enumerate(sorted(self.context.keys())): 

81 value = self.context[key] 

82 attr = 'frame' if index % 2 else 'frame header' 

83 cols = urwid.Columns([ 

84 (self.key_width, urwid.Text(key)), 

85 urwid.Text(str(value)), 

86 ], dividechars=1) 

87 yield urwid.AttrMap(cols, attr) 

88 

89 

90class SettingsWindow(ObjectEditWindow): 

91 """The main settings window.""" 

92 

93 def __init__(self, app, **kwargs): 

94 """Initialize the window with available edit widgets. 

95 

96 :param app: `hods.app.App` instance 

97 """ 

98 super().__init__(app, app.settings, title='Settings', **kwargs) 

99 

100 self.user = pw_user() 

101 self.hostname = get_hostname() 

102 

103 # server edit widget 

104 self.edit_server = urwid.Edit(edit_text=app.settings.server or '') 

105 urwid.connect_signal(self.edit_server, 'change', self.on_edit_server_change) 

106 self.edit_server_help = urwid.Text('') 

107 

108 self.checkbox_is_proxy = urwid.CheckBox( 

109 'This is a proxy / Sync all features', 

110 state=app.settings.is_proxy, on_state_change=self.on_change_is_proxy) 

111 self.is_proxy = app.settings.is_proxy 

112 

113 self.checkbox_server_is_proxy = urwid.CheckBox( 

114 'Server is a proxy / Use rsync to sync git repositories with the server', 

115 state=app.settings.server_is_proxy, on_state_change=self.on_change_server_is_proxy) 

116 self.server_is_proxy = app.settings.server_is_proxy 

117 

118 self.checkbox_cascade = urwid.CheckBox( 

119 'Cascade / SSH to proxy to synchronize it before pull and after push.', 

120 state=app.settings.cascade) 

121 

122 # ssh_agent_mode radio buttons 

123 self.ssh_agent_mode = app.settings.ssh_agent_mode 

124 choices = [ 

125 ('Start if not running', 'start'), 

126 ('Ask to start if not running', 'ask'), 

127 ('Warn if not running', 'warn'), 

128 ('Do nothing', 'ignore'), 

129 ] 

130 self.ssh_agent_mode_buttons = list(radio_buttons( 

131 choices, 

132 getset=(self, 'ssh_agent_mode'), 

133 on_change=lambda a, b, user_data=None: self.refresh(), 

134 )) 

135 

136 # ssh_agent_key edit widget 

137 self.edit_ssh_agent_key = urwid.Edit(edit_text=app.settings.ssh_agent_key or '') 

138 urwid.connect_signal(self.edit_ssh_agent_key, 'change', self.on_edit_ssh_agent_key_change) 

139 self.edit_ssh_agent_key_help = urwid.Text('') 

140 

141 self.variable_edits = {} 

142 for key, default in app.config.variables.items(): 

143 value = app.settings.variables.get(key, '') 

144 self.add_variable_edit(key, default, value) 

145 

146 self.template_header_columns = urwid.Columns([ 

147 ('pack', urwid.Text('Template Variables:')), 

148 (7, FocusMap( 

149 urwid.Button('Add', on_press=self.add_variable), 

150 'success', 

151 )), 

152 (12, FocusMap( 

153 urwid.Button('Show All', on_press=self.show_variables), 

154 'button', 

155 )), 

156 ], dividechars=1) 

157 

158 # initial help values 

159 self.on_edit_server_change(value=self.edit_server.edit_text) 

160 self.on_edit_ssh_agent_key_change(value=self.app.settings.ssh_agent_key) 

161 

162 def add_variable_edit(self, key, default='', value=''): 

163 """Add edit widgets for default and local values.""" 

164 self.variable_edits[key] = ( 

165 urwid.Edit(edit_text=default, wrap=urwid.CLIP), 

166 urwid.Edit(edit_text=value, wrap=urwid.CLIP), 

167 ) 

168 

169 def on_change_is_proxy(self, btn, value): 

170 """Store *is_proxy* value and refresh widgets.""" 

171 self.is_proxy = value 

172 self.refresh() 

173 

174 def on_change_server_is_proxy(self, btn, value): 

175 """Store *server_is_proxy* value and refresh widgets.""" 

176 self.server_is_proxy = value 

177 self.refresh() 

178 

179 def get_items(self): 

180 """Collect the widgets to display.""" 

181 yield urwid.Columns([(7, urwid.Text('Server:')), FocusMap(self.edit_server, 'edit')], dividechars=1) 

182 yield urwid.Columns([(8, urwid.Text('')), self.edit_server_help]) 

183 

184 if self.server: # this is not the master server 

185 yield urwid.Divider() 

186 yield self.checkbox_is_proxy 

187 if not self.is_proxy: 

188 yield self.checkbox_server_is_proxy 

189 if self.server_is_proxy: 

190 yield urwid.Columns([(4, urwid.Text('')), self.checkbox_cascade]) 

191 

192 yield urwid.Divider() 

193 

194 yield urwid.Text('SSH Agent:') 

195 

196 for button in self.ssh_agent_mode_buttons: 

197 yield button 

198 

199 if self.ssh_agent_mode in ('start', 'ask'): 

200 yield urwid.Columns([ 

201 (4, urwid.Text('Key:')), 

202 FocusMap(self.edit_ssh_agent_key, 'edit'), 

203 ], dividechars=1) 

204 yield urwid.Columns([(5, urwid.Text('')), self.edit_ssh_agent_key_help]) 

205 

206 yield urwid.Divider() 

207 yield self.template_header_columns 

208 

209 keys = self.variable_edits.keys() 

210 width = max(map(len, keys)) if keys else 5 

211 width = width if width > 4 else 5 

212 

213 yield urwid.Columns([ 

214 (width + 1, urwid.Text('Name:')), 

215 urwid.Text('Default:'), 

216 urwid.Text('Value:'), 

217 (10, urwid.Text('')), 

218 ], dividechars=1) 

219 for key in sorted(self.variable_edits.keys()): 

220 default_edit, value_edit = self.variable_edits[key] 

221 yield urwid.Columns([ 

222 (width + 1, urwid.Text(key)), 

223 FocusMap(default_edit, 'edit'), 

224 FocusMap(value_edit, 'edit'), 

225 (10, FocusMap( 

226 urwid.Button('Delete', on_press=self.delete_variable, user_data=key), 

227 'error', 

228 )), 

229 ], dividechars=1) 

230 

231 def show_variables(self, btn, user_data=None): 

232 """Open window showing all template context variables.""" 

233 overwrite = dict(self.clean_variable_values()) 

234 overwrite['server'] = self.edit_server.edit_text 

235 overwrite['is_proxy'] = self.checkbox_is_proxy.state 

236 overwrite['server_is_proxy'] = self.checkbox_server_is_proxy.state 

237 overwrite['cascade'] = self.checkbox_cascade.state 

238 overwrite['file'] = '<SourceFile object>' 

239 ContextWindow(self.app, overwrite=overwrite, parent=self).show() 

240 

241 def add_variable(self, btn, user_data=None): 

242 """Open window to input variable name and add it.""" 

243 def add(data): 

244 self.add_variable_edit(data['name']) 

245 self.refresh() 

246 VariableAddWindow(self.app, self.variable_edits.keys(), parent=self, 

247 on_save=add, defer_save=False).show() 

248 

249 def delete_variable(self, btn, user_data=None): 

250 """Remove the variable given as user_data.""" 

251 del self.variable_edits[user_data] 

252 self.refresh() 

253 

254 def on_edit_server_change(self, edit=None, value=None): 

255 """Called when changing `server`.""" 

256 self.server = value 

257 try: 

258 value = clean_server(value) 

259 except ValueError as e: 

260 value = 'ERROR: ' + str(e) 

261 else: 

262 if value is None: 

263 value = 'This is the master server.' 

264 else: 

265 if '@' not in value: 

266 value = '{}@{}'.format(pw_user(), value) 

267 value = '{}:.hods{}'.format(value, os.sep) 

268 self.edit_server_help.set_text(value) 

269 self.refresh() 

270 

271 def on_edit_ssh_agent_key_change(self, edit=None, value=None): 

272 """Called when changing `ssh_agent_key`.""" 

273 value = self.clean_key(value) or '' 

274 if not value: 

275 value = '.ssh/id_rsa' 

276 elif value.startswith(self.app.settings.home_path): 

277 value = value[len(self.app.settings.home_path) + 1:] 

278 

279 check_path = value 

280 if not os.path.isabs(check_path): 

281 check_path = os.path.join(self.app.settings.home_path, value) 

282 

283 if not os.path.exists(check_path): 

284 value = '(not found) {}'.format(value) 

285 self.edit_ssh_agent_key_help.set_text(value) 

286 

287 def clean_server_address(self, value): 

288 """Clean server address.""" 

289 try: 

290 value = clean_server(value) 

291 except ValueError as e: 

292 raise self.ValidationError(str(e)) 

293 return value 

294 

295 def clean_key(self, value=None): 

296 """Clean ssh agent key.""" 

297 if value is None: 

298 value = self.edit_ssh_agent_key.edit_text 

299 

300 value = value.strip() 

301 if not value: 

302 return 

303 

304 if os.path.isabs(value): 

305 return value 

306 

307 parts = value.split(os.sep) 

308 if parts[0] in ('~', '$HOME', '${HOME}'): 

309 parts.pop(0) 

310 return os.path.join(self.app.settings.home_path, *parts) 

311 

312 def clean_variable_defaults(self): 

313 """Collect and clean variables default values.""" 

314 for key, edits in self.variable_edits.items(): 

315 yield key, edits[0].edit_text 

316 

317 def clean_variable_values(self): 

318 """Collect and clean variables local values.""" 

319 for key, edits in self.variable_edits.items(): 

320 yield key, edits[1].edit_text 

321 

322 def on_close(self): 

323 """Disconnect signals on window close.""" 

324 urwid.disconnect_signal(self.edit_server, 'change', self.on_edit_server_change) 

325 urwid.disconnect_signal(self.edit_ssh_agent_key, 'change', self.on_edit_ssh_agent_key_change) 

326 super().on_close() 

327 

328 def clean(self): 

329 """Clean form data and return it.""" 

330 return { 

331 'server': self.clean_server_address(self.server), 

332 'is_proxy': self.checkbox_is_proxy.state, 

333 'ssh_agent_mode': self.ssh_agent_mode, 

334 'ssh_agent_key': self.clean_key(), 

335 'variables': dict(self.clean_variable_values()), 

336 'variables_defaults': dict(self.clean_variable_defaults()), 

337 } 

338 

339 def save(self, data): 

340 """Store variables default values in config.""" 

341 self.app.config.variables = data.pop('variables_defaults') 

342 os.makedirs(self.app.settings.dirname, exist_ok=True) 

343 return super().save(data)