Coverage for app/snags/routes.py: 100%
198 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1from datetime import UTC, date, datetime, time
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]
14from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
15from models import ( # pyright: ignore[reportMissingImports]
16 Aircraft,
17 Role,
18 Snag,
19 TenantUser,
20 db,
21)
22from utils import ( # pyright: ignore[reportMissingImports]
23 activity,
24 login_required,
25 require_role,
26 user_can_access_aircraft,
27)
29snags_bp = Blueprint("snags", __name__)
31_CREW_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT, Role.MAINTENANCE)
34def _tenant_id() -> int:
35 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
36 if not tu:
37 abort(403)
38 return int(tu.tenant_id)
41def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
42 ac = db.session.get(Aircraft, aircraft_id)
43 if (
44 not ac
45 or ac.tenant_id != _tenant_id()
46 or not user_can_access_aircraft(aircraft_id)
47 ):
48 abort(404)
49 return ac
52def _get_snag_or_404(aircraft: Aircraft, snag_id: int) -> Snag:
53 s = db.session.get(Snag, snag_id)
54 if not s or s.aircraft_id != aircraft.id:
55 abort(404)
56 return s
59def _parse_required_date(raw: str, label: str) -> tuple[date | None, str | None]:
60 """Parse a required ``YYYY-MM-DD`` form field. Returns ``(value, None)``
61 on success or ``(None, error_message)`` on the first failure — mirrors
62 the ``(values, error)`` idiom used by other blueprints' form parsers."""
63 raw = raw.strip()
64 if not raw:
65 return None, str(_("%(label)s is required.", label=label))
66 try:
67 d = date.fromisoformat(raw)
68 except ValueError:
69 return None, str(_("%(label)s must be a valid date (YYYY-MM-DD).", label=label))
70 if d > datetime.now(UTC).date():
71 return None, str(_("%(label)s cannot be in the future.", label=label))
72 return d, None
75# ── Snag list ─────────────────────────────────────────────────────────────────
78@snags_bp.route("/aircraft/<aircraft_ref:aircraft_id>/snags")
79@login_required
80def list_snags(aircraft_id: int) -> ResponseReturnValue:
81 ac = _get_aircraft_or_404(aircraft_id)
82 open_snags = (
83 Snag.query.filter_by(aircraft_id=ac.id, resolved_at=None)
84 .order_by(Snag.is_grounding.desc(), Snag.reported_at.desc())
85 .all()
86 )
87 closed_snags = (
88 Snag.query.filter(Snag.aircraft_id == ac.id, Snag.resolved_at.isnot(None))
89 .order_by(Snag.resolved_at.desc())
90 .all()
91 )
92 return render_template(
93 "snags/list.html", aircraft=ac, open_snags=open_snags, closed_snags=closed_snags
94 )
97# ── Add snag ──────────────────────────────────────────────────────────────────
100@snags_bp.route(
101 "/aircraft/<aircraft_ref:aircraft_id>/snags/new", methods=["GET", "POST"]
102)
103@login_required
104@require_role(*_CREW_ROLES)
105def new_snag(aircraft_id: int) -> ResponseReturnValue:
106 ac = _get_aircraft_or_404(aircraft_id)
107 if request.method == "POST":
108 return _save_snag(ac, None)
109 return render_template(
110 "snags/snag_form.html",
111 aircraft=ac,
112 snag=None,
113 today=date.today().isoformat(),
114 )
117# ── Edit snag ─────────────────────────────────────────────────────────────────
120@snags_bp.route(
121 "/aircraft/<aircraft_ref:aircraft_id>/snags/<int:snag_id>/edit",
122 methods=["GET", "POST"],
123)
124@login_required
125@require_role(*_CREW_ROLES)
126def edit_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue:
127 ac = _get_aircraft_or_404(aircraft_id)
128 s = _get_snag_or_404(ac, snag_id)
129 if request.method == "POST":
130 return _save_snag(ac, s)
131 return render_template(
132 "snags/snag_form.html", aircraft=ac, snag=s, today=date.today().isoformat()
133 )
136def _save_snag(ac: Aircraft, s: Snag | None) -> ResponseReturnValue:
137 title = request.form.get("title", "").strip()
138 description = request.form.get("description", "").strip() or None
139 reporter = request.form.get("reporter", "").strip() or None
140 is_grounding = bool(request.form.get("is_grounding"))
141 reported_at_raw = request.form.get("reported_at", "").strip()
143 errors = []
144 if not title:
145 errors.append(_("Title is required."))
147 reported_date, date_error = _parse_required_date(
148 reported_at_raw, str(_("Date signalled"))
149 )
150 if date_error:
151 errors.append(date_error)
153 # Editing an already-closed snag also exposes the resolution fields, so a
154 # mistake made when it was originally closed (wrong date, typo in the
155 # note) can be fixed without leaving the snag stuck unresolved.
156 editing_closed = s is not None and not s.is_open
157 resolved_date = None
158 resolution_note = None
159 if editing_closed:
160 resolved_at_raw = request.form.get("resolved_at", "").strip()
161 resolution_note = request.form.get("resolution_note", "").strip()
162 resolved_date, resolved_error = _parse_required_date(
163 resolved_at_raw, str(_("Resolution date"))
164 )
165 if resolved_error:
166 errors.append(resolved_error)
167 if not resolution_note:
168 errors.append(_("A resolution note is required."))
169 if reported_date and resolved_date and resolved_date < reported_date:
170 errors.append(_("Resolution date cannot be before the date signalled."))
172 if errors:
173 for msg in errors:
174 flash(msg, "danger")
175 return render_template(
176 "snags/snag_form.html", aircraft=ac, snag=s, today=date.today().isoformat()
177 )
179 _snag_is_new = s is None
180 old_values = None
181 if s is not None and not s.is_open:
182 old_values = {
183 "title": s.title,
184 "description": s.description,
185 "reporter": s.reporter,
186 "is_grounding": s.is_grounding,
187 "reported_at": s.reported_at.isoformat(),
188 "resolved_at": s.resolved_at.isoformat() if s.resolved_at else None,
189 "resolution_note": s.resolution_note,
190 }
192 if s is None:
193 s = Snag(aircraft_id=ac.id)
194 db.session.add(s)
196 s.title = title
197 s.description = description
198 s.reporter = reporter
199 s.is_grounding = is_grounding
200 assert reported_date is not None
201 s.reported_at = datetime.combine(reported_date, time.min, tzinfo=UTC)
202 if editing_closed:
203 assert resolved_date is not None
204 s.resolved_at = datetime.combine(resolved_date, time.min, tzinfo=UTC)
205 s.resolution_note = resolution_note
206 db.session.commit()
208 if old_values is not None:
209 activity(
210 "snag.edited_closed",
211 snag_id=s.id,
212 aircraft_id=ac.id,
213 old=old_values,
214 new={
215 "title": s.title,
216 "description": s.description,
217 "reporter": s.reporter,
218 "is_grounding": s.is_grounding,
219 "reported_at": s.reported_at.isoformat(),
220 "resolved_at": s.resolved_at.isoformat() if s.resolved_at else None,
221 "resolution_note": s.resolution_note,
222 },
223 )
225 if _snag_is_new:
226 activity(
227 "snag.opened",
228 snag_id=s.id,
229 aircraft_id=ac.id,
230 title=title,
231 is_grounding=is_grounding,
232 )
233 try:
234 from flask_babel import (
235 lazy_gettext as _l, # pyright: ignore[reportMissingImports]
236 )
237 from models import NotificationType # pyright: ignore[reportMissingImports]
238 from services.notification_service import (
239 dispatch, # pyright: ignore[reportMissingImports]
240 )
242 tid = _tenant_id()
243 notif_type = (
244 NotificationType.GROUNDING_SNAG_OPENED
245 if is_grounding
246 else NotificationType.SNAG_REPORTED
247 )
248 if is_grounding:
249 subject_key = _l("Grounding snag reported: %(title)s — %(reg)s")
250 title_key = _l("Grounding snag reported: %(title)s")
251 message_key = _l("A grounding snag was reported on %(reg)s.")
252 else:
253 subject_key = _l("Snag reported: %(title)s — %(reg)s")
254 title_key = _l("Snag reported: %(title)s")
255 message_key = _l("A snag was reported on %(reg)s.")
256 dispatch(
257 notif_type,
258 tid,
259 {
260 "subject_key": subject_key,
261 "subject_args": {"title": title, "reg": ac.registration},
262 "notification_title_key": title_key,
263 "notification_title_args": {"title": title},
264 "notification_message_key": message_key,
265 "notification_message_args": {"reg": ac.registration},
266 "details": [
267 (_l("Aircraft"), ac.registration),
268 (_l("Title"), title),
269 (_l("Reporter"), s.reporter or "—"),
270 ],
271 "is_grounding": is_grounding,
272 },
273 )
275 if is_grounding:
276 from models import ( # pyright: ignore[reportMissingImports]
277 Reservation,
278 ReservationStatus,
279 )
281 now = datetime.now(UTC)
282 affected = (
283 Reservation.query.filter(
284 Reservation.aircraft_id == ac.id,
285 Reservation.status == ReservationStatus.CONFIRMED,
286 Reservation.end_dt >= now,
287 Reservation.pilot_user_id.isnot(None),
288 )
289 .order_by(Reservation.start_dt)
290 .all()
291 )
292 pilot_ids = sorted({r.pilot_user_id for r in affected})
293 if pilot_ids:
294 dates = ", ".join(
295 r.start_dt.strftime("%Y-%m-%d %H:%M") for r in affected
296 )
297 dispatch(
298 NotificationType.RESERVATION_AIRCRAFT_GROUNDED,
299 tid,
300 {
301 "subject_key": _l(
302 "Aircraft grounded — check your reservation: %(reg)s"
303 ),
304 "subject_args": {"reg": ac.registration},
305 "notification_title_key": _l("Aircraft grounded: %(reg)s"),
306 "notification_title_args": {"reg": ac.registration},
307 "notification_message_key": _l(
308 "A grounding snag (%(title)s) was reported on "
309 "%(reg)s, which you hold a confirmed "
310 "reservation for."
311 ),
312 "notification_message_args": {
313 "title": title,
314 "reg": ac.registration,
315 },
316 "details": [
317 (_l("Aircraft"), ac.registration),
318 (_l("Snag"), title),
319 (_l("Affected reservation dates"), dates),
320 ],
321 },
322 target_user_ids=pilot_ids,
323 )
324 except Exception:
325 import logging as _log
327 _log.getLogger(__name__).exception("Failed to dispatch snag notification")
329 flash(_("Snag '%(title)s' saved.", title=s.title), "success")
330 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
333# ── Resolve snag ──────────────────────────────────────────────────────────────
336@snags_bp.route(
337 "/aircraft/<aircraft_ref:aircraft_id>/snags/<int:snag_id>/resolve",
338 methods=["GET", "POST"],
339)
340@login_required
341@require_role(*_CREW_ROLES)
342def resolve_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue:
343 ac = _get_aircraft_or_404(aircraft_id)
344 s = _get_snag_or_404(ac, snag_id)
345 if not s.is_open:
346 flash(_("Snag is already closed."), "danger")
347 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
349 if request.method == "POST":
350 note = request.form.get("resolution_note", "").strip()
351 resolved_at_raw = request.form.get("resolved_at", "").strip()
353 errors = []
354 if not note:
355 errors.append(_("A resolution note is required."))
356 resolved_date, date_error = _parse_required_date(
357 resolved_at_raw, str(_("Resolution date"))
358 )
359 if date_error:
360 errors.append(date_error)
361 if resolved_date and resolved_date < s.reported_at.date():
362 errors.append(_("Resolution date cannot be before the date signalled."))
364 if errors:
365 for msg in errors:
366 flash(msg, "danger")
367 return render_template(
368 "snags/resolve_form.html",
369 aircraft=ac,
370 snag=s,
371 today=date.today().isoformat(),
372 )
374 assert resolved_date is not None
375 s.resolved_at = datetime.combine(resolved_date, time.min, tzinfo=UTC)
376 s.resolution_note = note
377 db.session.commit()
378 activity(
379 "snag.resolved",
380 snag_id=snag_id,
381 aircraft_id=aircraft_id,
382 title=s.title,
383 resolved_at=s.resolved_at.isoformat(),
384 )
385 flash(_("Snag '%(title)s' closed.", title=s.title), "success")
386 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
388 return render_template(
389 "snags/resolve_form.html", aircraft=ac, snag=s, today=date.today().isoformat()
390 )
393# ── Reopen snag ───────────────────────────────────────────────────────────────
396@snags_bp.route(
397 "/aircraft/<aircraft_ref:aircraft_id>/snags/<int:snag_id>/reopen",
398 methods=["POST"],
399)
400@login_required
401@require_role(*_CREW_ROLES)
402def reopen_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue:
403 ac = _get_aircraft_or_404(aircraft_id)
404 s = _get_snag_or_404(ac, snag_id)
405 if s.is_open:
406 flash(_("Snag is already open."), "danger")
407 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
409 previous_resolved_at = s.resolved_at
410 s.resolved_at = None
411 s.resolution_note = None
412 db.session.commit()
413 activity(
414 "snag.reopened",
415 snag_id=s.id,
416 aircraft_id=ac.id,
417 title=s.title,
418 previous_resolved_at=(
419 previous_resolved_at.isoformat() if previous_resolved_at else None
420 ),
421 )
422 flash(_("Snag '%(title)s' reopened.", title=s.title), "warning")
423 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
426# ── Delete snag ───────────────────────────────────────────────────────────────
429@snags_bp.route(
430 "/aircraft/<aircraft_ref:aircraft_id>/snags/<int:snag_id>/delete", methods=["POST"]
431)
432@login_required
433@require_role(*_CREW_ROLES)
434def delete_snag(aircraft_id: int, snag_id: int) -> ResponseReturnValue:
435 ac = _get_aircraft_or_404(aircraft_id)
436 s = _get_snag_or_404(ac, snag_id)
437 if not s.is_open:
438 flash(_("Closed snags are archived and cannot be deleted."), "danger")
439 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))
440 title = s.title
441 db.session.delete(s)
442 db.session.commit()
443 flash(_("Snag '%(title)s' deleted.", title=title), "success")
444 return redirect(url_for("snags.list_snags", aircraft_id=ac.id))