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

1"""

2Reservations blueprint — aircraft booking calendar, create/edit/cancel,

3owner approval workflow, and per-aircraft booking settings.

4"""

6import calendar

7from datetime import datetime, timedelta, timezone

8from urllib.parse import urlparse

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

11 Blueprint,

12 abort,

13 flash,

14 redirect,

15 render_template,

16 request,

17 session,

18 url_for,

19)

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

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

23 Aircraft,

24 AircraftBookingSettings,

25 Reservation,

26 ReservationStatus,

27 Role,

28 TenantUser,

29 db,

30)

31from utils import login_required, require_role, user_can_access_aircraft # pyright: ignore[reportMissingImports]

33reservations_bp = Blueprint("reservations", __name__)

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

36_BOOKING_ROLES = (Role.ADMIN, Role.OWNER, Role.PILOT)

39def _safe_next(next_url: str, fallback: str) -> str:

40 """Return next_url only when it is a safe relative path, otherwise fallback."""

41 next_url = next_url.replace("\\", "")

42 parsed = urlparse(next_url)

43 if (

44 next_url

45 and not parsed.scheme

46 and not parsed.netloc

47 and next_url.startswith("/")

48 ):

49 return next_url

50 return fallback

53# ── Helpers ───────────────────────────────────────────────────────────────────

56def _tenant_id() -> int:

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

58 if not tu:

59 abort(403) # pragma: no cover

60 return tu.tenant_id

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

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

65 if (

66 not ac

67 or ac.tenant_id != _tenant_id()

68 or not user_can_access_aircraft(aircraft_id)

69 ):

70 abort(404)

71 return ac

74def _get_reservation_or_404(ac: Aircraft, res_id: int) -> Reservation:

75 r = db.session.get(Reservation, res_id)

76 if not r or r.aircraft_id != ac.id:

77 abort(404)

78 return r

81def _has_conflict(

82 aircraft_id: int,

83 start_dt: datetime,

84 end_dt: datetime,

85 exclude_id: int | None = None,

86) -> bool:

87 """Return True if any confirmed reservation overlaps [start_dt, end_dt)."""

88 q = Reservation.query.filter(

89 Reservation.aircraft_id == aircraft_id,

90 Reservation.status == ReservationStatus.CONFIRMED,

91 Reservation.start_dt < end_dt,

92 Reservation.end_dt > start_dt,

93 )

94 if exclude_id is not None:

95 q = q.filter(Reservation.id != exclude_id)

96 return q.first() is not None

99def _parse_datetime(s: str) -> datetime | None:

100 """Parse 'YYYY-MM-DDTHH:MM' (HTML datetime-local) → UTC-aware datetime."""

101 try:

102 return datetime.fromisoformat(s).replace(tzinfo=timezone.utc)

103 except (ValueError, AttributeError):

104 return None

105

106

107def _compute_cost(

108 duration_hours: float, settings: AircraftBookingSettings | None

109) -> tuple[float | None, float | None]:

110 """Return (hourly_rate, estimated_cost) or (None, None) if no rate configured."""

111 if not settings or settings.hourly_rate is None:

112 return None, None

113 rate = float(settings.hourly_rate)

114 return rate, round(rate * duration_hours, 2)

115

116

117def _build_calendar_grid(year: int, month: int):

118 """Return a list of weeks; each week is a list of date objects (Mon–Sun).

119 Days outside the month are included to complete the grid."""

120 cal = calendar.Calendar(firstweekday=0) # Monday first

121 return cal.monthdatescalendar(year, month)

122

123

124# ── Fleet reservations overview (admin/owner) ─────────────────────────────────

125

126

127@reservations_bp.route("/reservations/fleet/")

128@login_required

129@require_role(*_OWNER_ROLES)

130def fleet_reservations():

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

132 if not tu:

133 abort(403) # pragma: no cover

134

135 role = tu.role

136 from utils import accessible_aircraft # pyright: ignore[reportMissingImports]

137

138 aircraft_qs = accessible_aircraft(tu.tenant_id)

139 if role == Role.OWNER:

140 # Owners only see planes they explicitly have access to

141 from models import UserAircraftAccess, UserAllAircraftAccess # pyright: ignore[reportMissingImports]

142

143 all_access = UserAllAircraftAccess.query.filter_by(user_id=tu.user_id).first()

144 if not all_access:

145 owned_ids = [

146 r.aircraft_id

147 for r in UserAircraftAccess.query.filter_by(user_id=tu.user_id).all()

148 ]

149 aircraft_qs = aircraft_qs.filter(Aircraft.id.in_(owned_ids))

150

151 aircraft_list = aircraft_qs.order_by(Aircraft.registration).all()

152 aircraft_ids = [a.id for a in aircraft_list]

153

154 now = datetime.now(timezone.utc)

155 expired_cutoff = now - timedelta(days=60)

156

157 reservations = (

158 (

159 Reservation.query.filter(

160 Reservation.aircraft_id.in_(aircraft_ids),

161 # Exclude expired-pending older than 60 days — they're just noise

162 db.or_(

163 Reservation.status != ReservationStatus.PENDING,

164 Reservation.start_dt >= expired_cutoff,

165 ),

166 )

167 .order_by(Reservation.start_dt)

168 .all()

169 )

170 if aircraft_ids

171 else []

172 )

173

174 # SQLite returns naive datetimes even for DateTime(timezone=True) columns;

175 # PostgreSQL returns timezone-aware. Normalize `now` to match so that

176 # Python comparisons and Jinja2 template filters stay compatible with both.

177 if reservations and reservations[0].start_dt.tzinfo is None:

178 now = now.replace(tzinfo=None)

179

180 # Detect overlapping confirmed reservations per aircraft

181 overlapping_ids: set[int] = set()

182 from itertools import combinations

183

184 confirmed = [r for r in reservations if r.status == ReservationStatus.CONFIRMED]

185 by_aircraft: dict[int, list] = {}

186 for r in confirmed:

187 by_aircraft.setdefault(r.aircraft_id, []).append(r)

188 for group in by_aircraft.values():

189 for r1, r2 in combinations(group, 2):

190 if r1.start_dt < r2.end_dt and r1.end_dt > r2.start_dt:

191 overlapping_ids.add(r1.id)

192 overlapping_ids.add(r2.id)

193

194 # Find past confirmed reservations with no flight logged on that aircraft/date

195 from models import FlightEntry # pyright: ignore[reportMissingImports]

196

197 missing_flight_ids: set[int] = set()

198 for r in reservations:

199 if r.status == ReservationStatus.CONFIRMED and r.end_dt <= now:

200 start_date = r.start_dt.date()

201 end_date = r.end_dt.date()

202 has_flight = (

203 FlightEntry.query.filter(

204 FlightEntry.aircraft_id == r.aircraft_id,

205 FlightEntry.date >= start_date,

206 FlightEntry.date <= end_date,

207 ).first()

208 is not None

209 )

210 if not has_flight:

211 missing_flight_ids.add(r.id)

212

213 aircraft_map = {a.id: a for a in aircraft_list}

214

215 return render_template(

216 "reservations/fleet.html",

217 reservations=reservations,

218 aircraft_map=aircraft_map,

219 overlapping_ids=overlapping_ids,

220 missing_flight_ids=missing_flight_ids,

221 now=now,

222 ReservationStatus=ReservationStatus,

223 )

224

225

226# ── Calendar view ─────────────────────────────────────────────────────────────

227

228

229@reservations_bp.route("/aircraft/<int:aircraft_id>/reservations/")

230@login_required

231def calendar_view(aircraft_id: int):

232 ac = _get_aircraft_or_404(aircraft_id)

233 today = datetime.now(timezone.utc).date()

234

235 try:

236 year = int(request.args.get("year", today.year))

237 month = int(request.args.get("month", today.month))

238 except ValueError:

239 year, month = today.year, today.month

240

241 # Clamp to valid range

242 if month < 1:

243 year -= 1

244 month = 12

245 if month > 12:

246 year += 1

247 month = 1

248

249 # Month boundaries in UTC

250 month_start = datetime(year, month, 1, tzinfo=timezone.utc)

251 last_day = calendar.monthrange(year, month)[1]

252 month_end = datetime(year, month, last_day, 23, 59, 59, tzinfo=timezone.utc)

253

254 reservations = (

255 Reservation.query.filter(

256 Reservation.aircraft_id == ac.id,

257 Reservation.start_dt <= month_end,

258 Reservation.end_dt >= month_start,

259 )

260 .order_by(Reservation.start_dt)

261 .all()

262 )

263

264 # Build a dict day → list of reservations for fast template lookup

265 from collections import defaultdict

266

267 day_reservations: dict = defaultdict(list)

268 for r in reservations:

269 # A reservation may span multiple days — add it to each day it touches

270 cur = r.start_dt.date()

271 end = r.end_dt.date()

272 while cur <= end:

273 day_reservations[cur].append(r)

274 cur += timedelta(days=1)

275

276 # Prev / next month navigation

277 prev_month = month - 1 or 12

278 prev_year = year - 1 if month == 1 else year

279 next_month = month % 12 + 1

280 next_year = year + 1 if month == 12 else year

281

282 weeks = _build_calendar_grid(year, month)

283

284 return render_template(

285 "reservations/calendar.html",

286 aircraft=ac,

287 weeks=weeks,

288 day_reservations=day_reservations,

289 year=year,

290 month=month,

291 month_name=datetime(year, month, 1).strftime("%B %Y"),

292 today=today,

293 prev_year=prev_year,

294 prev_month=prev_month,

295 next_year=next_year,

296 next_month=next_month,

297 ReservationStatus=ReservationStatus,

298 )

299

300

301# ── Create reservation ────────────────────────────────────────────────────────

302

303

304@reservations_bp.route(

305 "/aircraft/<int:aircraft_id>/reservations/new", methods=["GET", "POST"]

306)

307@login_required

308@require_role(*_BOOKING_ROLES)

309def new_reservation(aircraft_id: int):

310 ac = _get_aircraft_or_404(aircraft_id)

311 settings = ac.booking_settings

312 if request.method == "POST":

313 return _save_reservation(ac, None, settings)

314 # Pre-fill start from query string (clicked day on calendar)

315 prefill_start = request.args.get("date", "")

316 return render_template(

317 "reservations/form.html",

318 aircraft=ac,

319 reservation=None,

320 settings=settings,

321 prefill_start=prefill_start,

322 )

323

324

325# ── Edit reservation ──────────────────────────────────────────────────────────

326

327

328@reservations_bp.route(

329 "/aircraft/<int:aircraft_id>/reservations/<int:res_id>/edit",

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

331)

332@login_required

333def edit_reservation(aircraft_id: int, res_id: int):

334 ac = _get_aircraft_or_404(aircraft_id)

335 r = _get_reservation_or_404(ac, res_id)

336

337 # Pilots may only edit their own pending reservations

338 role = TenantUser.query.filter_by(user_id=session["user_id"]).first()

339 user_role = role.role if role else None

340 is_owner_role = user_role in _OWNER_ROLES

341 if not is_owner_role:

342 if (

343 r.pilot_user_id != session["user_id"]

344 or r.status != ReservationStatus.PENDING

345 ):

346 abort(403)

347

348 settings = ac.booking_settings

349 if request.method == "POST":

350 return _save_reservation(ac, r, settings)

351 return render_template(

352 "reservations/form.html",

353 aircraft=ac,

354 reservation=r,

355 settings=settings,

356 prefill_start="",

357 )

358

359

360# ── Cancel reservation ────────────────────────────────────────────────────────

361

362

363@reservations_bp.route(

364 "/aircraft/<int:aircraft_id>/reservations/<int:res_id>/cancel", methods=["POST"]

365)

366@login_required

367def cancel_reservation(aircraft_id: int, res_id: int):

368 ac = _get_aircraft_or_404(aircraft_id)

369 r = _get_reservation_or_404(ac, res_id)

370

371 role = TenantUser.query.filter_by(user_id=session["user_id"]).first()

372 user_role = role.role if role else None

373 is_owner_role = user_role in _OWNER_ROLES

374 if not is_owner_role and r.pilot_user_id != session["user_id"]:

375 abort(403)

376

377 if r.status == ReservationStatus.CANCELLED:

378 flash(_("Reservation is already cancelled."), "warning")

379 return redirect(url_for("reservations.calendar_view", aircraft_id=ac.id))

380

381 r.status = ReservationStatus.CANCELLED

382 db.session.commit()

383 if r.pilot_user_id:

384 try:

385 from models import NotificationType # pyright: ignore[reportMissingImports]

386 from services.notification_service import dispatch # pyright: ignore[reportMissingImports]

387

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

389 if tu:

390 dispatch(

391 NotificationType.RESERVATION_CANCELLED,

392 tu.tenant_id,

393 {

394 "subject": f"Reservation cancelled — {ac.registration}",

395 "notification_title": f"Reservation cancelled: {ac.registration}",

396 "notification_message": f"Your reservation for {ac.registration} from {r.start_dt.strftime('%Y-%m-%d %H:%M')} to {r.end_dt.strftime('%Y-%m-%d %H:%M')} UTC has been cancelled.",

397 "details": [

398 ("Aircraft", ac.registration),

399 ("Start", r.start_dt.strftime("%Y-%m-%d %H:%M UTC")),

400 ("End", r.end_dt.strftime("%Y-%m-%d %H:%M UTC")),

401 ],

402 },

403 target_user_ids=[r.pilot_user_id],

404 )

405 except Exception:

406 import logging as _log

407

408 _log.getLogger(__name__).exception(

409 "Failed to dispatch reservation cancelled notification"

410 )

411 flash(_("Reservation cancelled."), "success")

412 return redirect(url_for("reservations.calendar_view", aircraft_id=ac.id))

413

414

415# ── Confirm / decline (owner only) ───────────────────────────────────────────

416

417

418@reservations_bp.route(

419 "/aircraft/<int:aircraft_id>/reservations/<int:res_id>/confirm", methods=["POST"]

420)

421@login_required

422@require_role(*_OWNER_ROLES)

423def confirm_reservation(aircraft_id: int, res_id: int):

424 ac = _get_aircraft_or_404(aircraft_id)

425 r = _get_reservation_or_404(ac, res_id)

426 _next = request.form.get("next", "")

427 _fallback = url_for("reservations.calendar_view", aircraft_id=ac.id)

428 _dest = _safe_next(_next, _fallback)

429

430 if r.status != ReservationStatus.PENDING:

431 flash(_("Only pending reservations can be confirmed."), "warning")

432 return redirect(_dest)

433

434 if _has_conflict(ac.id, r.start_dt, r.end_dt, exclude_id=r.id):

435 flash(_("Cannot confirm: overlapping confirmed reservation exists."), "danger")

436 return redirect(_dest)

437

438 r.status = ReservationStatus.CONFIRMED

439 db.session.commit()

440 if r.pilot_user_id:

441 try:

442 from models import NotificationType # pyright: ignore[reportMissingImports]

443 from services.notification_service import dispatch # pyright: ignore[reportMissingImports]

444

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

446 if tu:

447 dispatch(

448 NotificationType.RESERVATION_CONFIRMED,

449 tu.tenant_id,

450 {

451 "subject": f"Reservation confirmed — {ac.registration}",

452 "notification_title": f"Reservation confirmed: {ac.registration}",

453 "notification_message": f"Your reservation for {ac.registration} from {r.start_dt.strftime('%Y-%m-%d %H:%M')} to {r.end_dt.strftime('%Y-%m-%d %H:%M')} UTC has been confirmed.",

454 "details": [

455 ("Aircraft", ac.registration),

456 ("Start", r.start_dt.strftime("%Y-%m-%d %H:%M UTC")),

457 ("End", r.end_dt.strftime("%Y-%m-%d %H:%M UTC")),

458 ],

459 },

460 target_user_ids=[r.pilot_user_id],

461 )

462 except Exception:

463 import logging as _log

464

465 _log.getLogger(__name__).exception(

466 "Failed to dispatch reservation confirmed notification"

467 )

468 flash(_("Reservation confirmed."), "success")

469 return redirect(_dest)

470

471

472@reservations_bp.route(

473 "/aircraft/<int:aircraft_id>/reservations/<int:res_id>/decline", methods=["POST"]

474)

475@login_required

476@require_role(*_OWNER_ROLES)

477def decline_reservation(aircraft_id: int, res_id: int):

478 ac = _get_aircraft_or_404(aircraft_id)

479 r = _get_reservation_or_404(ac, res_id)

480 _next = request.form.get("next", "")

481 _fallback = url_for("reservations.calendar_view", aircraft_id=ac.id)

482 _dest = _safe_next(_next, _fallback)

483

484 if r.status != ReservationStatus.PENDING:

485 flash(_("Only pending reservations can be declined."), "warning")

486 return redirect(_dest)

487

488 r.status = ReservationStatus.CANCELLED

489 db.session.commit()

490 flash(_("Reservation declined."), "success")

491 return redirect(_dest)

492

493

494# ── Booking settings (owner only) ─────────────────────────────────────────────

495

496

497@reservations_bp.route(

498 "/aircraft/<int:aircraft_id>/reservations/settings", methods=["GET", "POST"]

499)

500@login_required

501@require_role(*_OWNER_ROLES)

502def booking_settings(aircraft_id: int):

503 ac = _get_aircraft_or_404(aircraft_id)

504 settings = ac.booking_settings

505

506 if request.method == "POST":

507 return _save_booking_settings(ac, settings)

508

509 return render_template("reservations/settings.html", aircraft=ac, settings=settings)

510

511

512def _save_booking_settings(ac: Aircraft, settings: AircraftBookingSettings | None):

513 def _float_or_none(key: str) -> float | None:

514 val = request.form.get(key, "").strip()

515 try:

516 return float(val) if val else None

517 except ValueError:

518 return None

519

520 min_h = _float_or_none("min_booking_hours")

521 max_h = _float_or_none("max_booking_hours")

522 rate = _float_or_none("hourly_rate")

523

524 errors = []

525 if min_h is not None and min_h <= 0:

526 errors.append(_("Minimum booking duration must be positive."))

527 if max_h is not None and max_h <= 0:

528 errors.append(_("Maximum booking duration must be positive."))

529 if min_h is not None and max_h is not None and min_h > max_h:

530 errors.append(_("Minimum duration cannot exceed maximum duration."))

531 if rate is not None and rate < 0:

532 errors.append(_("Hourly rate cannot be negative."))

533

534 if errors:

535 for msg in errors:

536 flash(msg, "danger")

537 return render_template(

538 "reservations/settings.html", aircraft=ac, settings=settings

539 )

540

541 if settings is None:

542 settings = AircraftBookingSettings(aircraft_id=ac.id)

543 db.session.add(settings)

544

545 settings.min_booking_hours = min_h

546 settings.max_booking_hours = max_h

547 settings.hourly_rate = rate

548 db.session.commit()

549 flash(_("Booking settings saved."), "success")

550 return redirect(url_for("reservations.calendar_view", aircraft_id=ac.id))

551

552

553# ── Shared save logic ─────────────────────────────────────────────────────────

554

555

556def _save_reservation(

557 ac: Aircraft, r: Reservation | None, settings: AircraftBookingSettings | None

558):

559 start_raw = request.form.get("start_dt", "").strip()

560 end_raw = request.form.get("end_dt", "").strip()

561 notes = request.form.get("notes", "").strip() or None

562

563 start_dt = _parse_datetime(start_raw)

564 end_dt = _parse_datetime(end_raw)

565

566 errors = []

567 if not start_dt:

568 errors.append(_("Start date/time is required."))

569 if not end_dt:

570 errors.append(_("End date/time is required."))

571 if start_dt and end_dt:

572 if end_dt <= start_dt:

573 errors.append(_("End must be after start."))

574 else:

575 duration = (end_dt - start_dt).total_seconds() / 3600

576 if settings:

577 if settings.min_booking_hours and duration < float(

578 settings.min_booking_hours

579 ):

580 errors.append(

581 _(

582 "Minimum booking duration is %(h)s h.",

583 h=settings.min_booking_hours,

584 )

585 )

586 if settings.max_booking_hours and duration > float(

587 settings.max_booking_hours

588 ):

589 errors.append(

590 _(

591 "Maximum booking duration is %(h)s h.",

592 h=settings.max_booking_hours,

593 )

594 )

595

596 if errors:

597 for msg in errors:

598 flash(msg, "danger")

599 return render_template(

600 "reservations/form.html",

601 aircraft=ac,

602 reservation=r,

603 settings=settings,

604 prefill_start="",

605 )

606

607 hourly_rate, estimated_cost = _compute_cost(

608 (end_dt - start_dt).total_seconds() / 3600, settings

609 )

610

611 _is_new_reservation = r is None

612 if r is None:

613 r = Reservation(

614 aircraft_id=ac.id,

615 pilot_user_id=session["user_id"],

616 status=ReservationStatus.PENDING,

617 )

618 db.session.add(r)

619

620 r.start_dt = start_dt

621 r.end_dt = end_dt

622 r.notes = notes

623 r.hourly_rate = hourly_rate

624 r.estimated_cost = estimated_cost

625 db.session.commit()

626

627 if _is_new_reservation:

628 try:

629 from models import NotificationType # pyright: ignore[reportMissingImports]

630 from services.notification_service import dispatch # pyright: ignore[reportMissingImports]

631

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

633 if tu:

634 dispatch(

635 NotificationType.RESERVATION_REQUEST,

636 tu.tenant_id,

637 {

638 "subject": f"New booking request — {ac.registration}",

639 "notification_title": f"New booking request: {ac.registration}",

640 "notification_message": f"A new booking request was submitted for {ac.registration} from {r.start_dt.strftime('%Y-%m-%d %H:%M')} to {r.end_dt.strftime('%Y-%m-%d %H:%M')} UTC.",

641 "details": [

642 ("Aircraft", ac.registration),

643 ("Start", r.start_dt.strftime("%Y-%m-%d %H:%M UTC")),

644 ("End", r.end_dt.strftime("%Y-%m-%d %H:%M UTC")),

645 ],

646 },

647 )

648 except Exception:

649 import logging as _log

650

651 _log.getLogger(__name__).exception(

652 "Failed to dispatch reservation request notification"

653 )

654

655 flash(_("Reservation saved."), "success")

656 return redirect(url_for("reservations.calendar_view", aircraft_id=ac.id))