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

1import logging 

2import os 

3from datetime import datetime, timedelta, timezone 

4 

5from flask import Blueprint, redirect, render_template, request, session, url_for 

6 

7from extensions import _rate_limiting_disabled, limiter as _limiter # pyright: ignore[reportMissingImports] 

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

9from flask_babel import get_locale as _get_locale # pyright: ignore[reportMissingImports] 

10 

11from models import DemoSlot, User, db 

12 

13log = logging.getLogger(__name__) 

14 

15demo_bp = Blueprint("demo", __name__) 

16 

17_DEFAULT_BUSY_WINDOW = 30 

18 

19 

20def _busy_window_minutes() -> int: 

21 try: 

22 return int( 

23 os.environ.get("OPENHANGAR_DEMO_BUSY_WINDOW_MINUTES", _DEFAULT_BUSY_WINDOW) 

24 ) 

25 except ValueError: 

26 return _DEFAULT_BUSY_WINDOW 

27 

28 

29@demo_bp.before_app_request 

30def _fix_stale_demo_session() -> None: 

31 """After a demo wipe, user_id in session may point to a deleted user. 

32 

33 The seed() function deletes old users and creates new ones (sequences are 

34 not reset), so stale user_ids from before the wipe no longer exist. Clear 

35 the stale user_id so the next page load shows the landing page with a 

36 fresh-entry prompt. The demo_slot_id is preserved so the visitor can 

37 seamlessly re-enter the same sandbox number. 

38 """ 

39 user_id = session.get("user_id") 

40 if not user_id or not session.get("demo_slot_id"): 

41 return 

42 if db.session.get(User, user_id) is None: 

43 session.pop("user_id", None) 

44 

45 

46@demo_bp.route("/demo/enter", methods=["POST"]) 

47@_limiter.limit("3 per minute", exempt_when=_rate_limiting_disabled) 

48def enter() -> ResponseReturnValue: 

49 role = request.form.get("role", "owner") # "owner" or "renter" 

50 

51 # Restore existing slot if still valid 

52 existing_slot_id = session.get("demo_slot_id") 

53 if existing_slot_id: 

54 slot = db.session.get(DemoSlot, existing_slot_id) 

55 if slot: 

56 session["user_id"] = _slot_user_id(slot, role) 

57 session["demo_role"] = role 

58 session.permanent = True 

59 _touch_slot(slot) 

60 return redirect(url_for("index")) 

61 

62 # Capture visitor locale (Accept-Language or manual switch) before session wipe 

63 visitor_lang = str(_get_locale()) 

64 

65 # Assign the least-recently-used slot 

66 slot = DemoSlot.query.order_by(DemoSlot.last_activity_at.asc().nullsfirst()).first() 

67 if slot is None: 

68 return redirect(url_for("index")) 

69 

70 # If even the LRU slot is still warm, all slots are actively in use 

71 window = _busy_window_minutes() 

72 cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(minutes=window) 

73 if slot.last_activity_at and slot.last_activity_at >= cutoff: 

74 return render_template("demo_full.html"), 503 

75 

76 session.clear() 

77 session["demo_slot_id"] = slot.id 

78 session["user_id"] = _slot_user_id(slot, role) 

79 session["demo_role"] = role 

80 session["language"] = visitor_lang 

81 session.permanent = True 

82 _touch_slot(slot) 

83 return redirect(url_for("index")) 

84 

85 

86def _slot_user_id(slot: DemoSlot, role: str) -> int: 

87 """Return the correct user_id for the requested role in this slot.""" 

88 if role in ("renter", "pilot") and slot.renter_user_id: 

89 return int(slot.renter_user_id) 

90 if role == "maintenance" and slot.maintenance_user_id: 

91 return int(slot.maintenance_user_id) 

92 if role == "viewer" and slot.viewer_user_id: 

93 return int(slot.viewer_user_id) 

94 if role == "sole_pilot" and slot.sole_pilot_user_id: 

95 return int(slot.sole_pilot_user_id) 

96 if role == "sole_operator" and slot.sole_operator_user_id: 

97 return int(slot.sole_operator_user_id) 

98 return int(slot.user_id) 

99 

100 

101def _touch_slot(slot: DemoSlot) -> None: 

102 slot.last_activity_at = datetime.now(timezone.utc) 

103 db.session.commit() 

104 

105 

106@demo_bp.route("/demo/next-wipe") 

107def next_wipe() -> ResponseReturnValue: 

108 """Return the scheduled next wipe time for the browser reload-detection logic.""" 

109 from flask import jsonify # noqa: PLC0415 

110 

111 return jsonify({"next_wipe": os.environ.get("OPENHANGAR_DEMO_NEXT_WIPE_UTC")}), 200 

112 

113 

114def demo_has_recent_activity(window_minutes: int = 20) -> bool: 

115 """Return True if any slot had activity within *window_minutes*.""" 

116 cutoff = datetime.now(timezone.utc) - timedelta(minutes=window_minutes) 

117 return int(DemoSlot.query.filter(DemoSlot.last_activity_at >= cutoff).count()) > 0