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