Coverage for app/airworthiness/routes.py: 100%
193 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 23:33 +0000
1import os
2from datetime import date
4from flask import ( # pyright: ignore[reportMissingImports]
5 Blueprint,
6 abort,
7 flash,
8 redirect,
9 render_template,
10 request,
11 session,
12 url_for,
13)
14from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
15from flask_babel import gettext as _, ngettext # pyright: ignore[reportMissingImports]
17from models import ( # pyright: ignore[reportMissingImports]
18 Aircraft,
19 AirworthinessDocument,
20 AirworthinessDocStatus,
21 AirworthinessDocType,
22 AirworthinessDocumentStatus,
23 Component,
24 EASASourceNode,
25 InstalledSTC,
26 Role,
27 TenantUser,
28 db,
29)
30from utils import login_required, require_role, user_can_access_aircraft # pyright: ignore[reportMissingImports]
32airworthiness_bp = Blueprint("airworthiness", __name__)
34_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
35_CREW_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT, Role.MAINTENANCE)
38def _tenant_id() -> int:
39 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
40 if not tu:
41 abort(403)
42 return int(tu.tenant_id)
45def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
46 ac = db.session.get(Aircraft, aircraft_id)
47 if (
48 not ac
49 or ac.tenant_id != _tenant_id()
50 or not user_can_access_aircraft(aircraft_id)
51 ):
52 abort(404)
53 return ac
56def _get_node_or_404(aircraft: Aircraft, node_id: int) -> EASASourceNode:
57 node = db.session.get(EASASourceNode, node_id)
58 if not node or node.component.aircraft_id != aircraft.id:
59 abort(404)
60 return node
63def _get_doc_or_404(aircraft: Aircraft, doc_id: int) -> AirworthinessDocument:
64 doc = db.session.get(AirworthinessDocument, doc_id)
65 if not doc:
66 abort(404)
67 # Document belongs to this aircraft if its component belongs to it
68 component = doc.component or (
69 doc.source_node.component if doc.source_node else None
70 )
71 if not component or component.aircraft_id != aircraft.id:
72 abort(404)
73 return doc
76def _get_stc_or_404(aircraft: Aircraft, stc_id: int) -> InstalledSTC:
77 stc = db.session.get(InstalledSTC, stc_id)
78 if not stc or stc.aircraft_id != aircraft.id:
79 abort(404)
80 return stc
83def _status_for(aircraft_id: int, doc_id: int) -> AirworthinessDocumentStatus | None:
84 return AirworthinessDocumentStatus.query.filter_by( # type: ignore[no-any-return]
85 aircraft_id=aircraft_id, document_id=doc_id
86 ).first()
89# ── Dashboard ─────────────────────────────────────────────────────────────────
92@airworthiness_bp.route("/aircraft/<int:aircraft_id>/airworthiness/")
93@login_required
94@require_role(*_CREW_ROLES)
95def dashboard(aircraft_id: int) -> ResponseReturnValue:
96 ac = _get_aircraft_or_404(aircraft_id)
98 # Gather all documents for this aircraft through its components
99 component_ids = [c.id for c in ac.components] # type: ignore[attr-defined]
101 # Documents via EASA source nodes
102 synced_docs = (
103 AirworthinessDocument.query.join(EASASourceNode)
104 .filter(EASASourceNode.component_id.in_(component_ids))
105 .all()
106 if component_ids
107 else []
108 )
109 # Manual documents attached directly to components
110 manual_docs = (
111 AirworthinessDocument.query.filter(
112 AirworthinessDocument.component_id.in_(component_ids),
113 AirworthinessDocument.source_node_id.is_(None),
114 ).all()
115 if component_ids
116 else []
117 )
118 all_docs = synced_docs + manual_docs
120 # Build status map {doc_id: AirworthinessDocumentStatus}
121 existing_statuses = {
122 s.document_id: s
123 for s in AirworthinessDocumentStatus.query.filter_by(
124 aircraft_id=aircraft_id
125 ).all()
126 }
128 # Attach status to each doc (or None if not yet created)
129 doc_rows = []
130 for doc in all_docs:
131 st = existing_statuses.get(doc.id)
132 component = doc.component or (
133 doc.source_node.component if doc.source_node else None
134 )
135 doc_rows.append({"doc": doc, "status": st, "component": component})
137 # Sort: pending first, then by doc_type, then reference
138 _STATUS_ORDER = {
139 AirworthinessDocStatus.PENDING_REVIEW: 0,
140 AirworthinessDocStatus.QUESTION: 1,
141 AirworthinessDocStatus.DEFERRED: 2,
142 AirworthinessDocStatus.COMPLIED: 3,
143 AirworthinessDocStatus.NOT_APPLICABLE: 4,
144 None: 0,
145 }
146 doc_rows.sort(
147 key=lambda r: (
148 _STATUS_ORDER.get(r["status"].status if r["status"] else None, 0),
149 r["doc"].doc_type,
150 r["doc"].reference,
151 )
152 )
154 # Summary counts
155 counts: dict[str, int] = {s: 0 for s in AirworthinessDocStatus.ALL}
156 counts["total"] = len(all_docs)
157 for row in doc_rows:
158 st_val = (
159 row["status"].status
160 if row["status"]
161 else AirworthinessDocStatus.PENDING_REVIEW
162 )
163 counts[st_val] = counts.get(st_val, 0) + 1
165 # Source nodes grouped by component
166 nodes_by_component: dict[int, list[EASASourceNode]] = {}
167 for comp in ac.components: # type: ignore[attr-defined]
168 if comp.easa_source_nodes:
169 nodes_by_component[comp.id] = comp.easa_source_nodes
171 is_production = os.environ.get("OPENHANGAR_ENV", "production") == "production"
172 return render_template(
173 "airworthiness/dashboard.html",
174 aircraft=ac,
175 doc_rows=doc_rows,
176 counts=counts,
177 nodes_by_component=nodes_by_component,
178 doc_types=AirworthinessDocType,
179 statuses=AirworthinessDocStatus,
180 installed_stcs=ac.installed_stcs,
181 is_production=is_production,
182 )
185# ── EASA sync (manual trigger) ────────────────────────────────────────────────
188@airworthiness_bp.route(
189 "/aircraft/<int:aircraft_id>/airworthiness/sync", methods=["POST"]
190)
191@login_required
192@require_role(*_OWNER_ROLES)
193def trigger_sync(aircraft_id: int) -> ResponseReturnValue:
194 if os.environ.get("OPENHANGAR_ENV", "production") != "production":
195 abort(403)
196 ac = _get_aircraft_or_404(aircraft_id)
197 from airworthiness_sync import sync_aircraft # pyright: ignore[reportMissingImports]
199 added, errors = sync_aircraft(ac)
200 if errors:
201 flash(
202 ngettext(
203 "Sync completed with errors on one node. Check logs.",
204 "Sync completed with errors on %(n)s nodes. Check logs.",
205 errors,
206 n=errors,
207 ),
208 "warning",
209 )
210 if added:
211 flash(
212 ngettext(
213 "One new document discovered.",
214 "%(n)s new documents discovered.",
215 added,
216 n=added,
217 ),
218 "success",
219 )
220 else:
221 flash(_("No new documents found."), "info")
222 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
225# ── EASA source nodes ─────────────────────────────────────────────────────────
228@airworthiness_bp.route(
229 "/aircraft/<int:aircraft_id>/airworthiness/nodes/new",
230 methods=["GET", "POST"],
231)
232@login_required
233@require_role(*_OWNER_ROLES)
234def add_node(aircraft_id: int) -> ResponseReturnValue:
235 ac = _get_aircraft_or_404(aircraft_id)
236 components = [c for c in ac.components if not c.removed_at] # type: ignore[attr-defined]
238 if request.method == "POST":
239 component_id = request.form.get("component_id", type=int)
240 comp = db.session.get(Component, component_id)
241 if not comp or comp.aircraft_id != aircraft_id:
242 abort(400)
244 node = EASASourceNode(
245 component_id=component_id,
246 tc_holder_node_id=request.form["tc_holder_node_id"].strip(),
247 tc_holder_name=request.form["tc_holder_name"].strip(),
248 type_node_id=request.form["type_node_id"].strip(),
249 type_name=request.form["type_name"].strip(),
250 model_node_id=request.form["model_node_id"].strip(),
251 model_name=request.form["model_name"].strip(),
252 )
253 db.session.add(node)
254 db.session.commit()
255 flash(_("EASA source node added."), "success")
256 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
258 preselect_component_id = request.args.get("component_id", type=int)
259 return render_template(
260 "airworthiness/node_form.html",
261 aircraft=ac,
262 components=components,
263 preselect_component_id=preselect_component_id,
264 )
267@airworthiness_bp.route(
268 "/aircraft/<int:aircraft_id>/airworthiness/nodes/<int:node_id>/delete",
269 methods=["POST"],
270)
271@login_required
272@require_role(*_OWNER_ROLES)
273def delete_node(aircraft_id: int, node_id: int) -> ResponseReturnValue:
274 ac = _get_aircraft_or_404(aircraft_id)
275 node = _get_node_or_404(ac, node_id)
276 db.session.delete(node)
277 db.session.commit()
278 flash(_("EASA source node removed."), "success")
279 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
282# ── Documents ─────────────────────────────────────────────────────────────────
285@airworthiness_bp.route(
286 "/aircraft/<int:aircraft_id>/airworthiness/documents/new",
287 methods=["GET", "POST"],
288)
289@login_required
290@require_role(*_OWNER_ROLES)
291def add_document(aircraft_id: int) -> ResponseReturnValue:
292 ac = _get_aircraft_or_404(aircraft_id)
293 components = [c for c in ac.components if not c.removed_at] # type: ignore[attr-defined]
295 if request.method == "POST":
296 component_id = request.form.get("component_id", type=int)
297 comp = db.session.get(Component, component_id)
298 if not comp or comp.aircraft_id != aircraft_id:
299 abort(400)
301 expiry_raw = request.form.get("expiry_date", "").strip()
302 expiry = date.fromisoformat(expiry_raw) if expiry_raw else None
304 doc = AirworthinessDocument(
305 doc_type=request.form["doc_type"].strip(),
306 reference=request.form["reference"].strip(),
307 title=request.form.get("title", "").strip() or None,
308 component_id=component_id,
309 doc_url=request.form.get("doc_url", "").strip() or None,
310 expiry_date=expiry,
311 )
312 db.session.add(doc)
313 db.session.flush()
315 status = AirworthinessDocumentStatus(
316 aircraft_id=aircraft_id,
317 document_id=doc.id,
318 status=AirworthinessDocStatus.PENDING_REVIEW,
319 )
320 db.session.add(status)
321 db.session.commit()
322 flash(_("Document added."), "success")
323 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
325 return render_template(
326 "airworthiness/document_form.html",
327 aircraft=ac,
328 components=components,
329 doc_types=AirworthinessDocType,
330 )
333@airworthiness_bp.route(
334 "/aircraft/<int:aircraft_id>/airworthiness/documents/<int:doc_id>/delete",
335 methods=["POST"],
336)
337@login_required
338@require_role(*_OWNER_ROLES)
339def delete_document(aircraft_id: int, doc_id: int) -> ResponseReturnValue:
340 ac = _get_aircraft_or_404(aircraft_id)
341 doc = _get_doc_or_404(ac, doc_id)
342 if not doc.is_manual:
343 flash(_("Only manually added documents can be deleted."), "danger")
344 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
345 db.session.delete(doc)
346 db.session.commit()
347 flash(_("Document deleted."), "success")
348 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
351# ── Status updates ────────────────────────────────────────────────────────────
354@airworthiness_bp.route(
355 "/aircraft/<int:aircraft_id>/airworthiness/documents/<int:doc_id>/status",
356 methods=["GET", "POST"],
357)
358@login_required
359@require_role(*_CREW_ROLES)
360def update_status(aircraft_id: int, doc_id: int) -> ResponseReturnValue:
361 ac = _get_aircraft_or_404(aircraft_id)
362 doc = _get_doc_or_404(ac, doc_id)
363 st = _status_for(aircraft_id, doc_id)
365 if request.method == "POST":
366 new_status = request.form.get("status", "").strip()
367 if new_status not in AirworthinessDocStatus.ALL:
368 abort(400)
370 compliance_raw = request.form.get("compliance_date", "").strip()
371 review_raw = request.form.get("next_review_date", "").strip()
373 if st is None:
374 st = AirworthinessDocumentStatus(
375 aircraft_id=aircraft_id, document_id=doc_id
376 )
377 db.session.add(st)
379 st.status = new_status
380 st.notes = request.form.get("notes", "").strip() or None
381 st.compliance_date = (
382 date.fromisoformat(compliance_raw) if compliance_raw else None
383 )
384 st.next_review_date = date.fromisoformat(review_raw) if review_raw else None
385 db.session.commit()
386 flash(_("Status updated."), "success")
387 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
389 return render_template(
390 "airworthiness/status_form.html",
391 aircraft=ac,
392 doc=doc,
393 current_status=st,
394 statuses=AirworthinessDocStatus,
395 )
398# ── Installed STCs ────────────────────────────────────────────────────────────
401@airworthiness_bp.route(
402 "/aircraft/<int:aircraft_id>/airworthiness/stcs/new",
403 methods=["GET", "POST"],
404)
405@login_required
406@require_role(*_OWNER_ROLES)
407def add_stc(aircraft_id: int) -> ResponseReturnValue:
408 ac = _get_aircraft_or_404(aircraft_id)
410 if request.method == "POST":
411 install_raw = request.form.get("installation_date", "").strip()
412 stc = InstalledSTC(
413 aircraft_id=aircraft_id,
414 stc_number=request.form["stc_number"].strip(),
415 title=request.form.get("title", "").strip() or None,
416 tc_holder=request.form.get("tc_holder", "").strip() or None,
417 installation_date=date.fromisoformat(install_raw) if install_raw else None,
418 notes=request.form.get("notes", "").strip() or None,
419 )
420 db.session.add(stc)
421 db.session.commit()
422 flash(_("Installed STC added."), "success")
423 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))
425 return render_template("airworthiness/stc_form.html", aircraft=ac)
428@airworthiness_bp.route(
429 "/aircraft/<int:aircraft_id>/airworthiness/stcs/<int:stc_id>/delete",
430 methods=["POST"],
431)
432@login_required
433@require_role(*_OWNER_ROLES)
434def delete_stc(aircraft_id: int, stc_id: int) -> ResponseReturnValue:
435 ac = _get_aircraft_or_404(aircraft_id)
436 stc = _get_stc_or_404(ac, stc_id)
437 db.session.delete(stc)
438 db.session.commit()
439 flash(_("Installed STC removed."), "success")
440 return redirect(url_for("airworthiness.dashboard", aircraft_id=aircraft_id))