Coverage for .tox/coverage/lib/python3.13/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 json

28import logging

29import os

30import re

32import sqlalchemy as sa

33from sqlalchemy import orm

34from deform.widget import SelectWidget

36from wuttjamaican.db.model import Upgrade

37from wuttaweb.views import MasterView

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

39from wuttaweb.forms.widgets import DiffWidget

40from wuttaweb.progress import get_progress_session

42log = logging.getLogger(__name__)

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

46 """

47 Master view for upgrades.

49 Default route prefix is ``upgrades``.

51 Notable URLs provided by this class:

53 * ``/upgrades/``

54 * ``/upgrades/new``

55 * ``/upgrades/XXX``

56 * ``/upgrades/XXX/edit``

57 * ``/upgrades/XXX/delete``

58 """

60 model_class = Upgrade

61 executable = True

62 execute_progress_template = "/upgrade.mako"

63 downloadable = True

64 configurable = True

66 labels = {

67 "target_key": "Target",

68 }

70 grid_columns = [

71 "created",

72 "target_key",

73 "description",

74 "status",

75 "executed",

76 "executed_by",

77 ]

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

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

82 """ """

83 g = grid

84 super().configure_grid(g)

85 model = self.app.model

86 enum = self.app.enum

87 handler = self.app.get_upgrade_handler()

89 # target_key

90 targets = handler.get_all_targets()

91 if len(targets) == 1:

92 g.remove("target_key")

93 else:

94 g.set_renderer(

95 "target_key", self.render_target_key_for_grid, targets=targets

96 )

98 # description

99 g.set_link("description")

100

101 # created_by

102 g.set_link("created_by")

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

104 g.set_joiner(

105 "created_by",

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

107 )

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

109

110 # status

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

112

113 # executed_by

114 g.set_link("executed_by")

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

116 g.set_joiner(

117 "executed_by",

118 lambda q: q.outerjoin(

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

120 ),

121 )

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

123

124 def render_target_key_for_grid(

125 self, upgrade, field, value, targets=None

126 ): # pylint: disable=empty-docstring,unused-argument

127 """ """

128 key = value or "default"

129 target = targets.get(key)

130 return target.get("label", key) if target else key

131

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

133 self, upgrade, data, i

134 ):

135 """ """

136 enum = self.app.enum

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

138 return "has-background-warning"

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

140 return "has-background-warning"

141 return None

142

143 def configure_form(

144 self, form

145 ): # pylint: disable=empty-docstring,too-many-branches,too-many-statements

146 """ """

147 f = form

148 super().configure_form(f)

149 enum = self.app.enum

150 upgrade = f.model_instance

151 handler = self.app.get_upgrade_handler()

152

153 # never show these

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

155

156 # sequence sanity

157 f.fields.set_sequence(

158 [

159 "target_key",

160 "description",

161 "notes",

162 "status",

163 "created",

164 "created_by",

165 "executed",

166 "executed_by",

167 ]

168 )

169

170 # target_key

171 if self.creating:

172 targets = handler.get_all_targets()

173 if len(targets) == 1:

174 f.remove("target_key")

175 else:

176 values = [(p["key"], p["label"]) for p in targets.values()]

177 f.set_widget("target_key", SelectWidget(values=values))

178 f.set_required("target_key")

179 else:

180 target = handler.get_target(upgrade.target_key)

181 f.fields.insert_after("target_key", "_target_key")

182 f.remove("target_key")

183 f.set_readonly("_target_key")

184 f.set_label("_target_key", "Target")

185 f.set_default("_target_key", target["label"])

186

187 # created

188 if self.creating or self.editing:

189 f.remove("created")

190

191 # created_by

192 if self.creating or self.editing:

193 f.remove("created_by")

194 else:

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

196

197 # notes

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

199

200 # status

201 if self.creating:

202 f.remove("status")

203 else:

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

205

206 # executed

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

208 f.remove("executed")

209

210 # executed_by

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

212 f.remove("executed_by")

213 else:

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

215

216 # exit_code

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

218 f.remove("exit_code")

219

220 # stdout / stderr (and package_diff)

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

222 enum.UpgradeStatus.SUCCESS,

223 enum.UpgradeStatus.FAILURE,

224 ):

225

226 # stdout_file

227 f.append("stdout_file")

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

229 url = self.get_action_url(

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

231 )

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

233 f.set_default(

234 "stdout_file", handler.get_upgrade_filepath(upgrade, "stdout.log")

235 )

236

237 # stderr_file

238 f.append("stderr_file")

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

240 url = self.get_action_url(

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

242 )

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

244 f.set_default(

245 "stderr_file", handler.get_upgrade_filepath(upgrade, "stderr.log")

246 )

247

248 # package_diff

249 if self.viewing and upgrade.target_key in ("default", None):

250 f.append("package_diff")

251 try:

252 before = self.read_pip_freeze_output(upgrade, "before")

253 after = self.read_pip_freeze_output(upgrade, "after")

254 except Exception: # pylint: disable=broad-exception-caught

255 f.set_default(

256 "package_diff", "(failed to read package data for this upgrade)"

257 )

258 else:

259 f.set_widget(

260 "package_diff",

261 DiffWidget(

262 self.request,

263 before,

264 after,

265 columns=["package", "old version", "new version"],

266 show_filter=True,

267 filtered_by_default=True,

268 repr_values=False,

269 ),

270 )

271

272 def read_pip_freeze_output(

273 self, upgrade, occurrence

274 ): # pylint: disable=empty-docstring

275 """ """

276 handler = self.app.get_upgrade_handler()

277 path = handler.get_upgrade_filepath(upgrade, f"requirements.{occurrence}.txt")

278 packages = {}

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

280 for line in f:

281 if line := line.strip():

282 if line.startswith("#"):

283 continue

284 if req := self.parse_requirement(line):

285 packages[req["name"]] = req["version"]

286 else:

287 log.warning("could not parse req from line: %s", line)

288 return packages

289

290 def parse_requirement(self, line): # pylint: disable=empty-docstring

291 """ """

292 if match := re.match(r"^.*@(.*)#egg=(.*)$", line):

293 return {"name": match.group(2), "version": match.group(1)}

294

295 if match := re.match(r"^(.*)==(.*)$", line):

296 return {"name": match.group(1), "version": match.group(2)}

297

298 return None

299

300 def delete_instance(self, obj):

301 """

302 We override this method to delete any files associated with

303 the upgrade, in addition to deleting the upgrade record. It

304 calls

305 :meth:`~wuttjamaican.upgrades.UpgradeHandler.delete_files()`

306 to accomplish this.

307 """

308 upgrade = obj

309 handler = self.app.get_upgrade_handler()

310

311 # delete output files

312 handler.delete_files(upgrade)

313

314 # continue as normal

315 super().delete_instance(upgrade)

316

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

318 """ """

319 upgrade = super().objectify(form)

320 enum = self.app.enum

321

322 # set user, status when creating

323 if self.creating:

324 upgrade.created_by = self.request.user

325 upgrade.status = enum.UpgradeStatus.PENDING

326

327 return upgrade

328

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

330 """ """

331 upgrade = obj

332 if filename:

333 handler = self.app.get_upgrade_handler()

334 return handler.get_upgrade_filepath(upgrade, filename)

335 return None

336

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

338 """

339 This will invoke

340 :meth:`~wuttjamaican.upgrades.UpgradeHandler.do_execute()` on

341 the :term:`upgrade handler`.

342

343 It returns a different "success" message based on whether or

344 not the execute call failed. (Even if the upgrade *command*

345 fails, the overall execution did technically complete okay.)

346 """

347 upgrade = obj

348 handler = self.app.get_upgrade_handler()

349

350 if not handler.do_execute(upgrade, user):

351 # TODO: would be nice to give this msg a "warning" background

352 return "Upgrade was executed, but it failed"

353

354 return "Upgrade was executed okay"

355

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

357 """ """

358 route_prefix = self.get_route_prefix()

359 handler = self.app.get_upgrade_handler()

360 upgrade = self.get_instance()

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

362

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

364 if session.get("complete"):

365

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

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

368 msg = session.get("success_msg")

369 if msg:

370 self.request.session.flash(msg)

371

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

373

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

375 # since user is about to get redirected.

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

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

378

379 # our return value will include all from progress session

380 data = dict(session)

381

382 # add whatever might be new from upgrade process STDOUT

383 path = handler.get_upgrade_filepath(upgrade, filename="stdout.log")

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

385 if os.path.exists(path):

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

387 if size > 0:

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

389 f.seek(offset)

390 chunk = f.read(size)

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

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

393 session.save()

394

395 return data

396

397 def configure_get_context(

398 self, simple_settings=None

399 ): # pylint: disable=empty-docstring

400 """ """

401 context = super().configure_get_context(simple_settings=simple_settings)

402 handler = self.app.get_upgrade_handler()

403

404 # targets

405 targets = handler.get_all_targets()

406 context["upgrade_targets"] = list(targets.values())

407

408 return context

409

410 def configure_gather_settings(

411 self, data, simple_settings=None

412 ): # pylint: disable=empty-docstring

413 """ """

414 settings = super().configure_gather_settings(

415 data, simple_settings=simple_settings

416 )

417

418 keys = []

419 for target in json.loads(data["upgrade_targets"]):

420 key = target["key"]

421 keys.append(key)

422

423 settings.append(

424 {

425 "name": f"{self.config.appname}.upgrades.{key}.command",

426 "value": target["command"],

427 }

428 )

429

430 if key != "default":

431 if label := target.get("label"):

432 settings.append(

433 {

434 "name": f"{self.config.appname}.upgrades.{key}.label",

435 "value": label,

436 }

437 )

438

439 settings.append(

440 {

441 "name": f"{self.config.appname}.upgrades.keys",

442 "value": ", ".join(keys),

443 }

444 )

445

446 return settings

447

448 def configure_remove_settings(

449 self, simple_settings=None, session=None

450 ): # pylint: disable=empty-docstring

451 """ """

452 super().configure_remove_settings(

453 simple_settings=simple_settings, session=session

454 )

455 model = self.app.model

456 session = session or self.Session()

457

458 to_delete = (

459 session.query(model.Setting)

460 .filter(

461 sa.or_(

462 model.Setting.name == f"{self.config.appname}.upgrades.keys",

463 model.Setting.name

464 == f"{self.config.appname}.upgrades.command", # deprecated

465 model.Setting.name.like(f"{self.config.appname}.upgrades.%.label"),

466 model.Setting.name.like(

467 f"{self.config.appname}.upgrades.%.command"

468 ),

469 )

470 )

471 .all()

472 )

473 for setting in to_delete:

474 self.app.delete_setting(session, setting.name)

475

476 @classmethod

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

478 """ """

479

480 # nb. Upgrade may come from custom model

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

482 app = wutta_config.get_app()

483 cls.model_class = app.model.Upgrade

484

485 cls._defaults(config)

486 cls._upgrade_defaults(config)

487

488 @classmethod

489 def _upgrade_defaults(cls, config):

490 route_prefix = cls.get_route_prefix()

491 permission_prefix = cls.get_permission_prefix()

492 instance_url_prefix = cls.get_instance_url_prefix()

493

494 # execution progress

495 config.add_route(

496 f"{route_prefix}.execute_progress",

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

498 )

499 config.add_view(

500 cls,

501 attr="execute_progress",

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

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

504 renderer="json",

505 )

506

507

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

509 base = globals()

510

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

512 "UpgradeView", base["UpgradeView"]

513 )

514 UpgradeView.defaults(config)

515

516

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

518 defaults(config)