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

1from datetime import date as _date 

2from datetime import timedelta 

3from typing import Any 

4 

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

6 Blueprint, 

7 abort, 

8 flash, 

9 redirect, 

10 render_template, 

11 request, 

12 session, 

13 url_for, 

14) 

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

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

17from models import ( 

18 Aircraft, 

19 Expense, 

20 ExpenseType, 

21 Flight, 

22 Role, 

23 TenantUser, 

24 db, 

25) # pyright: ignore[reportMissingImports] 

26from utils import ( # pyright: ignore[reportMissingImports] 

27 login_required, 

28 require_role, 

29 user_can_access_aircraft, 

30) 

31 

32from expenses.cost_dashboard import ( # pyright: ignore[reportMissingImports] 

33 DEFAULT_PERIOD_MONTHS, 

34 PERIOD_OPTIONS, 

35 compute_cost_dashboard, 

36) 

37from expenses.form_parsing import ( 

38 parse_expense_fields, # pyright: ignore[reportMissingImports] 

39) 

40 

41expenses_bp = Blueprint("expenses", __name__) 

42 

43_OWNER_ROLES = (Role.ADMIN, Role.OWNER) 

44 

45_CURRENCIES = ["EUR", "USD", "GBP", "CHF"] 

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

47_DEFAULT_PERIOD = 12 # months 

48 

49 

50def _tenant_id() -> int: 

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

52 if not tu: 

53 abort(403) 

54 return int(tu.tenant_id) 

55 

56 

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

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

59 if ( 

60 not ac 

61 or ac.tenant_id != _tenant_id() 

62 or not user_can_access_aircraft(aircraft_id) 

63 ): 

64 abort(404) 

65 return ac 

66 

67 

68def _get_expense_or_404(aircraft: Aircraft, expense_id: int) -> Expense: 

69 exp = db.session.get(Expense, expense_id) 

70 if not exp or exp.aircraft_id != aircraft.id: 

71 abort(404) 

72 return exp 

73 

74 

75def _fuel_expense_prefill(aircraft: Aircraft) -> dict[str, Any] | None: 

76 """Build "add expense" form defaults from a "Log as expense" link off a 

77 logged refuel — either flight-linked (`?flight_entry_id=...`, from 

78 aircraft/flight_detail.html) or standalone (`?date=...`, from a Refuel 

79 record with no flight, see refuels/list.html). A flight's before/after 

80 refuel quantities, and a standalone Refuel, are all independent 

81 purchases for cost-tracking purposes — only the date and aircraft 

82 matter, not which record a given amount happens to be logged against.""" 

83 flight_entry_id_raw = request.args.get("flight_entry_id", "").strip() 

84 date_raw = request.args.get("date", "").strip() 

85 

86 flight_id: int | None = None 

87 prefill_date: str | None = None 

88 

89 if flight_entry_id_raw.isdigit(): 

90 flight = db.session.get(Flight, int(flight_entry_id_raw)) 

91 if flight is not None and flight.aircraft_id == aircraft.id: 

92 flight_id = flight.id 

93 prefill_date = flight.date.isoformat() 

94 elif date_raw: 

95 try: 

96 _date.fromisoformat(date_raw) 

97 except ValueError: 

98 pass 

99 else: 

100 prefill_date = date_raw 

101 

102 if prefill_date is None: 

103 return None 

104 

105 unit = request.args.get("unit", "").strip() 

106 return { 

107 "flight_entry_id": flight_id, 

108 "date": prefill_date, 

109 "quantity": request.args.get("quantity", "").strip() or None, 

110 "unit": unit if unit in _UNITS else "L", 

111 } 

112 

113 

114def _compute_stats( 

115 expenses: list[Any], aircraft_id: int, period_months: int 

116) -> tuple[float, float | None, str]: 

117 """Return (total_cost, cost_per_hour, period_label) for the filtered expense list.""" 

118 total_cost = sum(float(e.amount) for e in expenses) 

119 

120 if period_months > 0: 

121 cutoff = _date.today() - timedelta(days=period_months * 30) 

122 flights = Flight.query.filter( 

123 Flight.aircraft_id == aircraft_id, 

124 Flight.date >= cutoff, 

125 ).all() 

126 period_label = f"last {period_months} months" 

127 else: 

128 flights = Flight.query.filter_by(aircraft_id=aircraft_id).all() 

129 period_label = "all time" 

130 

131 total_hours = sum( 

132 float(f.flight_time) 

133 if f.flight_time is not None 

134 else float(f.flight_time_counter_end) - float(f.flight_time_counter_start) 

135 for f in flights 

136 if f.flight_time is not None 

137 or ( 

138 f.flight_time_counter_end is not None 

139 and f.flight_time_counter_start is not None 

140 ) 

141 ) 

142 cost_per_hour = round(total_cost / total_hours, 2) if total_hours > 0 else None 

143 return total_cost, cost_per_hour, period_label 

144 

145 

146# ── Expense list ────────────────────────────────────────────────────────────── 

147 

148 

149@expenses_bp.route("/aircraft/<aircraft_ref:aircraft_id>/expenses") 

150@login_required 

151def list_expenses(aircraft_id: int) -> ResponseReturnValue: 

152 ac = _get_aircraft_or_404(aircraft_id) 

153 

154 type_filter = request.args.get("type", "") 

155 try: 

156 period_months = int(request.args.get("period", _DEFAULT_PERIOD)) 

157 except ValueError: 

158 period_months = _DEFAULT_PERIOD 

159 

160 query = Expense.query.filter_by(aircraft_id=ac.id) 

161 

162 if type_filter and type_filter in ExpenseType.ALL: 

163 query = query.filter_by(expense_type=type_filter) 

164 

165 if period_months > 0: 

166 cutoff = _date.today() - timedelta(days=period_months * 30) 

167 query = query.filter(Expense.date >= cutoff) 

168 

169 expenses = query.order_by(Expense.date.desc(), Expense.id.desc()).all() 

170 total_cost, cost_per_hour, period_label = _compute_stats( 

171 expenses, ac.id, period_months 

172 ) 

173 

174 return render_template( 

175 "expenses/list.html", 

176 aircraft=ac, 

177 expenses=expenses, 

178 type_filter=type_filter, 

179 period_months=period_months, 

180 total_cost=total_cost, 

181 cost_per_hour=cost_per_hour, 

182 period_label=period_label, 

183 expense_type_labels=ExpenseType.LABELS, 

184 currencies=_CURRENCIES, 

185 ) 

186 

187 

188# ── Cost dashboard (Phase 36) ────────────────────────────────────────────────── 

189 

190 

191@expenses_bp.route("/aircraft/<aircraft_ref:aircraft_id>/costs") 

192@login_required 

193def cost_dashboard(aircraft_id: int) -> ResponseReturnValue: 

194 ac = _get_aircraft_or_404(aircraft_id) 

195 

196 try: 

197 period_months = int(request.args.get("period", DEFAULT_PERIOD_MONTHS)) 

198 except ValueError: 

199 period_months = DEFAULT_PERIOD_MONTHS 

200 if period_months not in PERIOD_OPTIONS: 

201 period_months = DEFAULT_PERIOD_MONTHS 

202 

203 dashboard = compute_cost_dashboard(ac, period_months) 

204 

205 return render_template( 

206 "expenses/cost_dashboard.html", 

207 aircraft=ac, 

208 period_months=period_months, 

209 period_options=PERIOD_OPTIONS, 

210 **dashboard, 

211 ) 

212 

213 

214# ── Add expense ─────────────────────────────────────────────────────────────── 

215 

216 

217@expenses_bp.route( 

218 "/aircraft/<aircraft_ref:aircraft_id>/expenses/add", methods=["GET", "POST"] 

219) 

220@login_required 

221@require_role(*_OWNER_ROLES) 

222def add_expense(aircraft_id: int) -> ResponseReturnValue: 

223 ac = _get_aircraft_or_404(aircraft_id) 

224 

225 if request.method == "POST": 

226 err = _validate_and_save(ac, expense=None) 

227 if err is None: 

228 flash(_("Expense recorded."), "success") 

229 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id)) 

230 flash(err, "danger") 

231 

232 return render_template( 

233 "expenses/expense_form.html", 

234 aircraft=ac, 

235 expense=None, 

236 expense_types=ExpenseType.LABELS, 

237 currencies=_CURRENCIES, 

238 units=_UNITS, 

239 today=_date.today().isoformat(), 

240 prefill=_fuel_expense_prefill(ac) if request.method == "GET" else None, 

241 ) 

242 

243 

244# ── Edit expense ────────────────────────────────────────────────────────────── 

245 

246 

247@expenses_bp.route( 

248 "/aircraft/<aircraft_ref:aircraft_id>/expenses/<int:expense_id>/edit", 

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

250) 

251@login_required 

252@require_role(*_OWNER_ROLES) 

253def edit_expense(aircraft_id: int, expense_id: int) -> ResponseReturnValue: 

254 ac = _get_aircraft_or_404(aircraft_id) 

255 exp = _get_expense_or_404(ac, expense_id) 

256 

257 if request.method == "POST": 

258 err = _validate_and_save(ac, expense=exp) 

259 if err is None: 

260 flash(_("Expense updated."), "success") 

261 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id)) 

262 flash(err, "danger") 

263 

264 return render_template( 

265 "expenses/expense_form.html", 

266 aircraft=ac, 

267 expense=exp, 

268 expense_types=ExpenseType.LABELS, 

269 currencies=_CURRENCIES, 

270 units=_UNITS, 

271 today=_date.today().isoformat(), 

272 prefill=None, 

273 ) 

274 

275 

276# ── Delete expense ──────────────────────────────────────────────────────────── 

277 

278 

279@expenses_bp.route( 

280 "/aircraft/<aircraft_ref:aircraft_id>/expenses/<int:expense_id>/delete", 

281 methods=["POST"], 

282) 

283@login_required 

284@require_role(*_OWNER_ROLES) 

285def delete_expense(aircraft_id: int, expense_id: int) -> ResponseReturnValue: 

286 ac = _get_aircraft_or_404(aircraft_id) 

287 exp = _get_expense_or_404(ac, expense_id) 

288 db.session.delete(exp) 

289 db.session.commit() 

290 flash(_("Expense deleted."), "success") 

291 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id)) 

292 

293 

294# ── Shared save helper ──────────────────────────────────────────────────────── 

295 

296 

297def _validate_and_save(aircraft: Aircraft, expense: Expense | None) -> str | None: 

298 """Validate POST data, persist, return error string or None on success. 

299 

300 Field validation (excluding the receipt file) is delegated to 

301 parse_expense_fields(); note this checks the receipt file extension 

302 *after* all other fields now rather than interleaved between coverage 

303 and recurrence — a harmless reordering of which single error message 

304 wins when multiple fields are simultaneously invalid, not a behaviour 

305 change to what's ultimately accepted or persisted. 

306 """ 

307 values, error = parse_expense_fields(request.form) 

308 if error: 

309 return error 

310 

311 receipt_file = request.files.get("receipt") 

312 if receipt_file is not None and not receipt_file.filename: 

313 receipt_file = None 

314 if receipt_file is not None: 

315 import os as _os 

316 

317 from documents.routes import ( 

318 _ALLOWED_EXTS, # pyright: ignore[reportMissingImports] 

319 ) 

320 

321 ext = _os.path.splitext(receipt_file.filename or "")[1].lower() 

322 if ext not in _ALLOWED_EXTS: 

323 return str(_("This file type is not allowed for receipts.")) 

324 

325 date_val = values["date"] 

326 description = values["description"] 

327 quantity = values["quantity"] 

328 recurrence = values["recurrence"] 

329 

330 if expense is None: 

331 expense = Expense(aircraft_id=aircraft.id, created_by_id=session.get("user_id")) 

332 flight_entry_id_raw = request.form.get("flight_entry_id", "").strip() 

333 if flight_entry_id_raw.isdigit(): 

334 flight = db.session.get(Flight, int(flight_entry_id_raw)) 

335 if flight is not None and flight.aircraft_id == aircraft.id: 

336 expense.flight_entry_id = flight.id 

337 db.session.add(expense) 

338 

339 expense.date = date_val 

340 expense.expense_type = values["expense_type"] 

341 expense.expense_category = values["expense_category"] 

342 expense.description = description 

343 expense.amount = values["amount"] 

344 expense.currency = values["currency"] 

345 expense.quantity = quantity 

346 expense.unit = values["unit"] if quantity else None 

347 expense.coverage_start = values["coverage_start"] 

348 expense.coverage_end = values["coverage_end"] 

349 expense.recurrence = recurrence 

350 expense.recurrence_end = values["recurrence_end"] 

351 if recurrence is None: 

352 expense.recurrence_last_date = None 

353 

354 if receipt_file is not None: 

355 from documents.routes import ( # pyright: ignore[reportMissingImports] 

356 _delete_file, 

357 _save_upload_canonical, 

358 ) 

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

360 DocCategory, 

361 Document, 

362 Tenant, 

363 ) 

364 from werkzeug.utils import ( 

365 secure_filename, # pyright: ignore[reportMissingImports] 

366 ) 

367 

368 db.session.flush() 

369 title = description or _("Expense receipt %(date)s", date=date_val.isoformat()) 

370 tenant = db.session.get(Tenant, aircraft.tenant_id) 

371 if tenant is None: # pragma: no cover — the aircraft FK guarantees it exists 

372 abort(404) 

373 relpath, mime, size = _save_upload_canonical( 

374 receipt_file, tenant, aircraft, DocCategory.INVOICE, title 

375 ) 

376 # A new upload replaces any previous receipt on this expense. 

377 for old in Document.query.filter_by(expense_id=expense.id).all(): 

378 _delete_file(old.filename) 

379 db.session.delete(old) 

380 db.session.add( 

381 Document( 

382 aircraft_id=aircraft.id, 

383 expense_id=expense.id, 

384 filename=relpath, 

385 original_filename=secure_filename(receipt_file.filename or "receipt"), 

386 mime_type=mime, 

387 size_bytes=size, 

388 title=title, 

389 category=DocCategory.INVOICE, 

390 ) 

391 ) 

392 

393 db.session.commit() 

394 return None