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

1from datetime import UTC, date, datetime 

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] 

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

15from models import ( # pyright: ignore[reportMissingImports] 

16 Aircraft, 

17 Refuel, 

18 Role, 

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) 

28 

29refuels_bp = Blueprint("refuels", __name__) 

30 

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

32_UNITS = ["L", "gal"] 

33 

34 

35def _tenant_id() -> int: 

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

37 if not tu: 

38 abort(403) 

39 return int(tu.tenant_id) 

40 

41 

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

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

44 if ( 

45 not ac 

46 or ac.tenant_id != _tenant_id() 

47 or not user_can_access_aircraft(aircraft_id) 

48 ): 

49 abort(404) 

50 return ac 

51 

52 

53def _get_refuel_or_404(aircraft: Aircraft, refuel_id: int) -> Refuel: 

54 r = db.session.get(Refuel, refuel_id) 

55 if not r or r.aircraft_id != aircraft.id: 

56 abort(404) 

57 return r 

58 

59 

60def _parse_required_date(raw: str, label: str) -> tuple[date | None, str | None]: 

61 raw = raw.strip() 

62 if not raw: 

63 return None, str(_("%(label)s is required.", label=label)) 

64 try: 

65 d = date.fromisoformat(raw) 

66 except ValueError: 

67 return None, str(_("%(label)s must be a valid date (YYYY-MM-DD).", label=label)) 

68 if d > datetime.now(UTC).date(): 

69 return None, str(_("%(label)s cannot be in the future.", label=label)) 

70 return d, None 

71 

72 

73# ── Refuel list ─────────────────────────────────────────────────────────────── 

74 

75 

76@refuels_bp.route("/aircraft/<aircraft_ref:aircraft_id>/refuels") 

77@login_required 

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

79 ac = _get_aircraft_or_404(aircraft_id) 

80 refuels = ( 

81 Refuel.query.filter_by(aircraft_id=ac.id) 

82 .order_by(Refuel.date.desc(), Refuel.id.desc()) 

83 .all() 

84 ) 

85 return render_template("refuels/list.html", aircraft=ac, refuels=refuels) 

86 

87 

88# ── Add refuel ──────────────────────────────────────────────────────────────── 

89 

90 

91@refuels_bp.route( 

92 "/aircraft/<aircraft_ref:aircraft_id>/refuels/new", methods=["GET", "POST"] 

93) 

94@login_required 

95@require_role(*_CREW_ROLES) 

96def new_refuel(aircraft_id: int) -> ResponseReturnValue: 

97 ac = _get_aircraft_or_404(aircraft_id) 

98 if request.method == "POST": 

99 return _save_refuel(ac, None) 

100 return render_template( 

101 "refuels/refuel_form.html", 

102 aircraft=ac, 

103 refuel=None, 

104 units=_UNITS, 

105 today=date.today().isoformat(), 

106 ) 

107 

108 

109# ── Edit refuel ─────────────────────────────────────────────────────────────── 

110 

111 

112@refuels_bp.route( 

113 "/aircraft/<aircraft_ref:aircraft_id>/refuels/<int:refuel_id>/edit", 

114 methods=["GET", "POST"], 

115) 

116@login_required 

117@require_role(*_CREW_ROLES) 

118def edit_refuel(aircraft_id: int, refuel_id: int) -> ResponseReturnValue: 

119 ac = _get_aircraft_or_404(aircraft_id) 

120 r = _get_refuel_or_404(ac, refuel_id) 

121 if request.method == "POST": 

122 return _save_refuel(ac, r) 

123 return render_template( 

124 "refuels/refuel_form.html", 

125 aircraft=ac, 

126 refuel=r, 

127 units=_UNITS, 

128 today=date.today().isoformat(), 

129 ) 

130 

131 

132def _save_refuel(ac: Aircraft, r: Refuel | None) -> ResponseReturnValue: 

133 date_raw = request.form.get("date", "").strip() 

134 quantity_raw = request.form.get("quantity", "").strip() 

135 unit = request.form.get("unit", "").strip() 

136 note = request.form.get("note", "").strip() or None 

137 

138 errors = [] 

139 refuel_date, date_error = _parse_required_date(date_raw, str(_("Date"))) 

140 if date_error: 

141 errors.append(date_error) 

142 

143 quantity: float | None = None 

144 if not quantity_raw: 

145 errors.append(_("Quantity is required.")) 

146 else: 

147 try: 

148 quantity = float(quantity_raw) 

149 if quantity <= 0: 

150 raise ValueError 

151 except ValueError: 

152 errors.append(_("Quantity must be a positive number.")) 

153 

154 if unit not in _UNITS: 

155 unit = "L" 

156 

157 if errors: 

158 for msg in errors: 

159 flash(msg, "danger") 

160 return render_template( 

161 "refuels/refuel_form.html", 

162 aircraft=ac, 

163 refuel=r, 

164 units=_UNITS, 

165 today=date.today().isoformat(), 

166 ) 

167 

168 _refuel_is_new = r is None 

169 if r is None: 

170 r = Refuel(aircraft_id=ac.id, created_by_id=session.get("user_id")) 

171 db.session.add(r) 

172 

173 assert refuel_date is not None 

174 r.date = refuel_date 

175 r.quantity = quantity 

176 r.unit = unit 

177 r.note = note 

178 db.session.commit() 

179 

180 activity( 

181 "refuel.logged" if _refuel_is_new else "refuel.edited", 

182 refuel_id=r.id, 

183 aircraft_id=ac.id, 

184 quantity=str(r.quantity), 

185 unit=r.unit, 

186 ) 

187 flash(_("Refuel recorded."), "success") 

188 return redirect(url_for("refuels.list_refuels", aircraft_id=ac.id)) 

189 

190 

191# ── Delete refuel ───────────────────────────────────────────────────────────── 

192 

193 

194@refuels_bp.route( 

195 "/aircraft/<aircraft_ref:aircraft_id>/refuels/<int:refuel_id>/delete", 

196 methods=["POST"], 

197) 

198@login_required 

199@require_role(*_CREW_ROLES) 

200def delete_refuel(aircraft_id: int, refuel_id: int) -> ResponseReturnValue: 

201 ac = _get_aircraft_or_404(aircraft_id) 

202 r = _get_refuel_or_404(ac, refuel_id) 

203 db.session.delete(r) 

204 db.session.commit() 

205 flash(_("Refuel deleted."), "success") 

206 return redirect(url_for("refuels.list_refuels", aircraft_id=ac.id))