Coverage for app/snags/routes.py: 100%

1from datetime import datetime, timezone 

2 

3from flask import ( # pyright: ignore[reportMissingImports] 

4 Blueprint, 

5 abort, 

6 flash, 

7 redirect, 

8 render_template, 

9 request, 

10 session, 

11 url_for, 

12) 

13from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports] 

14 

15from flask_babel import gettext as _ # pyright: ignore[reportMissingImports] 

16 

17from models import Aircraft, Role, Snag, TenantUser, db # pyright: ignore[reportMissingImports] 

18from utils import activity, login_required, require_role, user_can_access_aircraft # pyright: ignore[reportMissingImports] 

19 

20snags_bp = Blueprint("snags", __name__) 

21 

22_CREW_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT, Role.MAINTENANCE) 

23 

24 

25def _tenant_id() -> int: 

26 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first() 

27 if not tu: 

28 abort(403) 

29 return int(tu.tenant_id) 

30 

31 

32def _get_aircraft_or_404(aircraft_id: int) -> Aircraft: 

33 ac = db.session.get(Aircraft, aircraft_id) 

34 if ( 

35 not ac 

36 or ac.tenant_id != _tenant_id() 

37 or not user_can_access_aircraft(aircraft_id) 

38 ): 

39 abort(404) 

40 return ac 

41 

42 

43def _get_snag_or_404(aircraft: Aircraft, snag_id: int) -> Snag: 

44 s = db.session.get(Snag, snag_id) 

45 if not s or s.aircraft_id != aircraft.id: 

46 abort(404) 

47 return s 

48 

49 

50# ── Snag list ───────────────────────────────────────────────────────────────── 

51 

52 

53@snags_bp.route("/aircraft/<int:aircraft_id>/snags") 

54@login_required 

55def list_snags(aircraft_id: int) -> ResponseReturnValue: 

56 ac = _get_aircraft_or_404(aircraft_id) 

57 open_snags = ( 

58 Snag.query.filter_by(aircraft_id=ac.id, resolved_at=None) 

59 .order_by(Snag.is_grounding.desc(), Snag.reported_at.desc()) 

60 .all() 

61 ) 

62 closed_snags = ( 

63 Snag.query.filter(Snag.aircraft_id == ac.id, Snag.resolved_at.isnot(None)) 

64 .order_by(Snag.resolved_at.desc()) 

65 .all() 

66 ) 

67 return render_template( 

68 "snags/list.html", aircraft=ac, open_snags=open_snags, closed_snags=closed_snags 

69 ) 

70 

71 

72# ── Add snag ────────────────────────────────────────────────────────────────── 

73 

74 

75@snags_bp.route("/aircraft/<int:aircraft_id>/snags/new", methods=["GET", "POST"]) 

76@login_required 

77@require_role(*_CREW_ROLES) 

78def new_snag(aircraft_id: int) -> ResponseReturnValue: 

79 ac = _get_aircraft_or_404(aircraft_id) 

80 if request.method == "POST": 

81 return _save_snag(ac, None) 

82 return render_template("snags/snag_form.html", aircraft=ac, snag=None) 

83 

84 

85# ── Edit snag ───────────────────────────────────────────────────────────────── 

86 

87 

88@snags_bp.route( 

89 "/aircraft/<int:aircraft_id>/snags/<int:snag_id>/edit", methods=["GET", "POST"] 

90) 

91@login_required 

92@require_role(*_CREW_ROLES) 

93def edit_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue: 

94 ac = _get_aircraft_or_404(aircraft_id) 

95 s = _get_snag_or_404(ac, snag_id) 

96 if not s.is_open: 

97 flash(_("Closed snags cannot be edited."), "danger") 

98 return redirect(url_for("snags.list_snags", aircraft_id=ac.id)) 

99 if request.method == "POST": 

100 return _save_snag(ac, s) 

101 return render_template("snags/snag_form.html", aircraft=ac, snag=s) 

102 

103 

104def _save_snag(ac: Aircraft, s: Snag | None) -> ResponseReturnValue: 

105 title = request.form.get("title", "").strip() 

106 description = request.form.get("description", "").strip() or None 

107 reporter = request.form.get("reporter", "").strip() or None 

108 is_grounding = bool(request.form.get("is_grounding")) 

109 

110 errors = [] 

111 if not title: 

112 errors.append(_("Title is required.")) 

113 

114 if errors: 

115 for msg in errors: 

116 flash(msg, "danger") 

117 return render_template("snags/snag_form.html", aircraft=ac, snag=s) 

118 

119 _snag_is_new = s is None 

120 if s is None: 

121 s = Snag(aircraft_id=ac.id) 

122 db.session.add(s) 

123 

124 s.title = title 

125 s.description = description 

126 s.reporter = reporter 

127 s.is_grounding = is_grounding 

128 db.session.commit() 

129 

130 if _snag_is_new: 

131 activity( 

132 "snag.opened", 

133 snag_id=s.id, 

134 aircraft_id=ac.id, 

135 title=title, 

136 is_grounding=is_grounding, 

137 ) 

138 try: 

139 from models import NotificationType # pyright: ignore[reportMissingImports] 

140 from services.notification_service import dispatch # pyright: ignore[reportMissingImports] 

141 

142 tid = _tenant_id() 

143 notif_type = ( 

144 NotificationType.GROUNDING_SNAG_OPENED 

145 if is_grounding 

146 else NotificationType.SNAG_REPORTED 

147 ) 

148 dispatch( 

149 notif_type, 

150 tid, 

151 { 

152 "subject": f"{'Grounding snag' if is_grounding else 'Snag'} reported: {title} — {ac.registration}", 

153 "notification_title": f"{'Grounding snag' if is_grounding else 'Snag'} reported: {title}", 

154 "notification_message": f"A {'grounding ' if is_grounding else ''}snag was reported on {ac.registration}.", 

155 "details": [ 

156 ("Aircraft", ac.registration), 

157 ("Title", title), 

158 ("Reporter", s.reporter or "—"), 

159 ], 

160 "is_grounding": is_grounding, 

161 }, 

162 ) 

163 except Exception: 

164 import logging as _log 

165 

166 _log.getLogger(__name__).exception("Failed to dispatch snag notification") 

167 

168 flash(_("Snag '%(title)s' saved.", title=s.title), "success") 

169 return redirect(url_for("snags.list_snags", aircraft_id=ac.id)) 

170 

171 

172# ── Resolve snag ────────────────────────────────────────────────────────────── 

173 

174 

175@snags_bp.route( 

176 "/aircraft/<int:aircraft_id>/snags/<int:snag_id>/resolve", methods=["GET", "POST"] 

177) 

178@login_required 

179@require_role(*_CREW_ROLES) 

180def resolve_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue: 

181 ac = _get_aircraft_or_404(aircraft_id) 

182 s = _get_snag_or_404(ac, snag_id) 

183 if not s.is_open: 

184 flash(_("Snag is already closed."), "danger") 

185 return redirect(url_for("snags.list_snags", aircraft_id=ac.id)) 

186 

187 if request.method == "POST": 

188 note = request.form.get("resolution_note", "").strip() 

189 if not note: 

190 flash(_("A resolution note is required."), "danger") 

191 return render_template("snags/resolve_form.html", aircraft=ac, snag=s) 

192 s.resolved_at = datetime.now(timezone.utc) 

193 s.resolution_note = note 

194 db.session.commit() 

195 activity( 

196 "snag.resolved", snag_id=snag_id, aircraft_id=aircraft_id, title=s.title 

197 ) 

198 flash(_("Snag '%(title)s' closed.", title=s.title), "success") 

199 return redirect(url_for("snags.list_snags", aircraft_id=ac.id)) 

200 

201 return render_template("snags/resolve_form.html", aircraft=ac, snag=s) 

202 

203 

204# ── Delete snag ─────────────────────────────────────────────────────────────── 

205 

206 

207@snags_bp.route( 

208 "/aircraft/<int:aircraft_id>/snags/<int:snag_id>/delete", methods=["POST"] 

209) 

210@login_required 

211@require_role(*_CREW_ROLES) 

212def delete_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue: 

213 ac = _get_aircraft_or_404(aircraft_id) 

214 s = _get_snag_or_404(ac, snag_id) 

215 title = s.title 

216 db.session.delete(s) 

217 db.session.commit() 

218 flash(_("Snag '%(title)s' deleted.", title=title), "success") 

219 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))