Coverage for .tox/coverage/lib/python3.11/site-packages/wuttaweb/views/upgrades.py: 100%

1# -*- coding: utf-8; -*-

2################################################################################

4# wuttaweb -- Web App for Wutta Framework

7# This file is part of Wutta Framework.

9# Wutta Framework is free software: you can redistribute it and/or modify it

10# under the terms of the GNU General Public License as published by the Free

11# Software Foundation, either version 3 of the License, or (at your option) any

12# later version.

13#

14# Wutta Framework is distributed in the hope that it will be useful, but

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for

17# more details.

18#

19# You should have received a copy of the GNU General Public License along with

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.

21#

22################################################################################

23"""

24Upgrade Views

25"""

27import logging

28import os

29import shutil

30import subprocess

32from sqlalchemy import orm

34from wuttjamaican.db.model import Upgrade

35from wuttaweb.views import MasterView

36from wuttaweb.forms.schema import UserRef, WuttaEnum, FileDownload

37from wuttaweb.progress import get_progress_session

40log = logging.getLogger(__name__)

43class UpgradeView(MasterView): # pylint: disable=abstract-method

44 """

45 Master view for upgrades.

47 Default route prefix is ``upgrades``.

49 Notable URLs provided by this class:

51 * ``/upgrades/``

52 * ``/upgrades/new``

53 * ``/upgrades/XXX``

54 * ``/upgrades/XXX/edit``

55 * ``/upgrades/XXX/delete``

56 """

58 model_class = Upgrade

59 executable = True

60 execute_progress_template = "/upgrade.mako"

61 downloadable = True

62 configurable = True

64 grid_columns = [

65 "created",

66 "description",

67 "status",

68 "executed",

69 "executed_by",

70 ]

72 sort_defaults = ("created", "desc")

74 def configure_grid(self, grid): # pylint: disable=empty-docstring

75 """ """

76 g = grid

77 super().configure_grid(g)

78 model = self.app.model

79 enum = self.app.enum

81 # description

82 g.set_link("description")

84 # created_by

85 g.set_link("created_by")

86 Creator = orm.aliased(model.User) # pylint: disable=invalid-name

87 g.set_joiner(

88 "created_by",

89 lambda q: q.join(Creator, Creator.uuid == model.Upgrade.created_by_uuid),

90 )

91 g.set_filter("created_by", Creator.username, label="Created By Username")

93 # status

94 g.set_renderer("status", self.grid_render_enum, enum=enum.UpgradeStatus)

96 # executed_by

97 g.set_link("executed_by")

98 Executor = orm.aliased(model.User) # pylint: disable=invalid-name

99 g.set_joiner(

100 "executed_by",

101 lambda q: q.outerjoin(

102 Executor, Executor.uuid == model.Upgrade.executed_by_uuid

103 ),

104 )

105 g.set_filter("executed_by", Executor.username, label="Executed By Username")

106

107 def grid_row_class( # pylint: disable=empty-docstring,unused-argument

108 self, upgrade, data, i

109 ):

110 """ """

111 enum = self.app.enum

112 if upgrade.status == enum.UpgradeStatus.EXECUTING:

113 return "has-background-warning"

114 if upgrade.status == enum.UpgradeStatus.FAILURE:

115 return "has-background-warning"

116 return None

117

118 def configure_form(self, form): # pylint: disable=empty-docstring

119 """ """

120 f = form

121 super().configure_form(f)

122 enum = self.app.enum

123 upgrade = f.model_instance

124

125 # never show these

126 f.remove("created_by_uuid", "executing", "executed_by_uuid")

127

128 # sequence sanity

129 f.fields.set_sequence(

130 [

131 "description",

132 "notes",

133 "status",

134 "created",

135 "created_by",

136 "executed",

137 "executed_by",

138 ]

139 )

140

141 # created

142 if self.creating or self.editing:

143 f.remove("created")

144

145 # created_by

146 if self.creating or self.editing:

147 f.remove("created_by")

148 else:

149 f.set_node("created_by", UserRef(self.request))

150

151 # notes

152 f.set_widget("notes", "notes")

153

154 # status

155 if self.creating:

156 f.remove("status")

157 else:

158 f.set_node("status", WuttaEnum(self.request, enum.UpgradeStatus))

159

160 # executed

161 if self.creating or self.editing or not upgrade.executed:

162 f.remove("executed")

163

164 # executed_by

165 if self.creating or self.editing or not upgrade.executed:

166 f.remove("executed_by")

167 else:

168 f.set_node("executed_by", UserRef(self.request))

169

170 # exit_code

171 if self.creating or self.editing or not upgrade.executed:

172 f.remove("exit_code")

173

174 # stdout / stderr

175 if not (self.creating or self.editing) and upgrade.status in (

176 enum.UpgradeStatus.SUCCESS,

177 enum.UpgradeStatus.FAILURE,

178 ):

179

180 # stdout_file

181 f.append("stdout_file")

182 f.set_label("stdout_file", "STDOUT")

183 url = self.get_action_url(

184 "download", upgrade, _query={"filename": "stdout.log"}

185 )

186 f.set_node("stdout_file", FileDownload(self.request, url=url))

187 f.set_default(

188 "stdout_file", self.get_upgrade_filepath(upgrade, "stdout.log")

189 )

190

191 # stderr_file

192 f.append("stderr_file")

193 f.set_label("stderr_file", "STDERR")

194 url = self.get_action_url(

195 "download", upgrade, _query={"filename": "stderr.log"}

196 )

197 f.set_node("stderr_file", FileDownload(self.request, url=url))

198 f.set_default(

199 "stderr_file", self.get_upgrade_filepath(upgrade, "stderr.log")

200 )

201

202 def delete_instance(self, obj):

203 """

204 We override this method to delete any files associated with

205 the upgrade, in addition to deleting the upgrade proper.

206 """

207 upgrade = obj

208 path = self.get_upgrade_filepath(upgrade, create=False)

209 if os.path.exists(path):

210 shutil.rmtree(path)

211

212 super().delete_instance(upgrade)

213

214 def objectify(self, form): # pylint: disable=empty-docstring

215 """ """

216 upgrade = super().objectify(form)

217 enum = self.app.enum

218

219 # set user, status when creating

220 if self.creating:

221 upgrade.created_by = self.request.user

222 upgrade.status = enum.UpgradeStatus.PENDING

223

224 return upgrade

225

226 def download_path(self, obj, filename): # pylint: disable=empty-docstring

227 """ """

228 upgrade = obj

229 if filename:

230 return self.get_upgrade_filepath(upgrade, filename)

231 return None

232

233 def get_upgrade_filepath( # pylint: disable=empty-docstring

234 self, upgrade, filename=None, create=True

235 ):

236 """ """

237 uuid = str(upgrade.uuid)

238 path = self.app.get_appdir(

239 "data", "upgrades", uuid[:2], uuid[2:], create=create

240 )

241 if filename:

242 path = os.path.join(path, filename)

243 return path

244

245 def execute_instance(self, obj, user, progress=None):

246 """

247 This method runs the actual upgrade.

248

249 Default logic will get the script command from config, and run

250 it via shell in a subprocess.

251

252 The ``stdout`` and ``stderr`` streams are captured to separate

253 log files which are then available to download.

254

255 The upgrade itself is marked as "executed" with status of

256 either ``SUCCESS`` or ``FAILURE``.

257 """

258 upgrade = obj

259 enum = self.app.enum

260

261 # locate file paths

262 script = self.config.require(f"{self.app.appname}.upgrades.command")

263 stdout_path = self.get_upgrade_filepath(upgrade, "stdout.log")

264 stderr_path = self.get_upgrade_filepath(upgrade, "stderr.log")

265

266 # record the fact that execution has begun for this upgrade

267 # nb. this is done in separate session to ensure it sticks,

268 # but also update local object to reflect the change

269 with self.app.short_session(commit=True) as s:

270 alt = s.merge(upgrade)

271 alt.status = enum.UpgradeStatus.EXECUTING

272 upgrade.status = enum.UpgradeStatus.EXECUTING

273

274 # run the command

275 log.debug("running upgrade command: %s", script)

276 with open(stdout_path, "wb") as stdout:

277 with open(stderr_path, "wb") as stderr:

278 upgrade.exit_code = subprocess.call(

279 script, shell=True, text=True, stdout=stdout, stderr=stderr

280 )

281 logger = log.warning if upgrade.exit_code != 0 else log.debug

282 logger("upgrade command had exit code: %s", upgrade.exit_code)

283

284 # declare it complete

285 upgrade.executed = self.app.make_utc()

286 upgrade.executed_by = user

287 if upgrade.exit_code == 0:

288 upgrade.status = enum.UpgradeStatus.SUCCESS

289 else:

290 upgrade.status = enum.UpgradeStatus.FAILURE

291

292 def execute_progress(self): # pylint: disable=empty-docstring

293 """ """

294 route_prefix = self.get_route_prefix()

295 upgrade = self.get_instance()

296 session = get_progress_session(self.request, f"{route_prefix}.execute")

297

298 # session has 'complete' flag set when operation is over

299 if session.get("complete"):

300

301 # set a flash msg for user if one is defined. this is the

302 # time to do it since user is about to get redirected.

303 msg = session.get("success_msg")

304 if msg:

305 self.request.session.flash(msg)

306

307 elif session.get("error"): # uh-oh

308

309 # set an error flash msg for user. this is the time to do it

310 # since user is about to get redirected.

311 msg = session.get("error_msg", "An unspecified error occurred.")

312 self.request.session.flash(msg, "error")

313

314 # our return value will include all from progress session

315 data = dict(session)

316

317 # add whatever might be new from upgrade process STDOUT

318 path = self.get_upgrade_filepath(upgrade, filename="stdout.log")

319 offset = session.get("stdout.offset", 0)

320 if os.path.exists(path):

321 size = os.path.getsize(path) - offset

322 if size > 0:

323 # with open(path, 'rb') as f:

324 with open(path, "rt", encoding="utf_8") as f:

325 f.seek(offset)

326 chunk = f.read(size)

327 # data['stdout'] = chunk.decode('utf8').replace('\n', '<br />')

328 data["stdout"] = chunk.replace("\n", "<br />")

329 session["stdout.offset"] = offset + size

330 session.save()

331

332 return data

333

334 def configure_get_simple_settings(self): # pylint: disable=empty-docstring

335 """ """

336

337 script = self.config.get(f"{self.app.appname}.upgrades.command")

338 if not script:

339 pass

340

341 return [

342 # basics

343 {"name": f"{self.app.appname}.upgrades.command", "default": script},

344 ]

345

346 @classmethod

347 def defaults(cls, config): # pylint: disable=empty-docstring

348 """ """

349

350 # nb. Upgrade may come from custom model

351 wutta_config = config.registry.settings["wutta_config"]

352 app = wutta_config.get_app()

353 cls.model_class = app.model.Upgrade

354

355 cls._defaults(config)

356 cls._upgrade_defaults(config)

357

358 @classmethod

359 def _upgrade_defaults(cls, config):

360 route_prefix = cls.get_route_prefix()

361 permission_prefix = cls.get_permission_prefix()

362 instance_url_prefix = cls.get_instance_url_prefix()

363

364 # execution progress

365 config.add_route(

366 f"{route_prefix}.execute_progress",

367 f"{instance_url_prefix}/execute/progress",

368 )

369 config.add_view(

370 cls,

371 attr="execute_progress",

372 route_name=f"{route_prefix}.execute_progress",

373 permission=f"{permission_prefix}.execute",

374 renderer="json",

375 )

376

377

378def defaults(config, **kwargs): # pylint: disable=missing-function-docstring

379 base = globals()

380

381 UpgradeView = kwargs.get( # pylint: disable=invalid-name,redefined-outer-name

382 "UpgradeView", base["UpgradeView"]

383 )

384 UpgradeView.defaults(config)

385

386

387def includeme(config): # pylint: disable=missing-function-docstring

388 defaults(config)

Coverage for .tox / coverage / lib / python3.11 / site-packages / wuttaweb / views / upgrades.py: 100%

168 statements