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

1import logging 

2import os 

3from datetime import UTC, datetime, timedelta 

4 

5from extensions import _rate_limiting_disabled # pyright: ignore[reportMissingImports] 

6from extensions import limiter as _limiter 

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

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

9from flask_babel import ( 

10 get_locale as _get_locale, # pyright: ignore[reportMissingImports] 

11) 

12from models import DemoSlot, User, db 

13 

14log = logging.getLogger(__name__) 

15 

16demo_bp = Blueprint("demo", __name__) 

17 

18_DEFAULT_BUSY_WINDOW = 30 

19 

20 

21def _busy_window_minutes() -> int: 

22 try: 

23 return int( 

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

25 ) 

26 except ValueError: 

27 return _DEFAULT_BUSY_WINDOW 

28 

29 

30@demo_bp.before_app_request 

31def _fix_stale_demo_session() -> None: 

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

33 

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

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

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

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

38 seamlessly re-enter the same sandbox number. 

39 """ 

40 user_id = session.get("user_id") 

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

42 return 

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

44 session.pop("user_id", None) 

45 

46 

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

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

49def enter() -> ResponseReturnValue: 

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

51 

52 # Restore existing slot if still valid 

53 existing_slot_id = session.get("demo_slot_id") 

54 if existing_slot_id: 

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

56 if slot: 

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

58 session["demo_role"] = role 

59 session.permanent = True 

60 _touch_slot(slot) 

61 return redirect(url_for("index")) 

62 

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

64 visitor_lang = str(_get_locale()) 

65 

66 # Assign the least-recently-used slot 

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

68 if slot is None: 

69 return redirect(url_for("index")) 

70 

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

72 window = _busy_window_minutes() 

73 cutoff = datetime.now(UTC).replace(tzinfo=None) - timedelta(minutes=window) 

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

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

76 

77 session.clear() 

78 session["demo_slot_id"] = slot.id 

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

80 session["demo_role"] = role 

81 session["language"] = visitor_lang 

82 session.permanent = True 

83 _touch_slot(slot) 

84 return redirect(url_for("index")) 

85 

86 

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

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

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

90 return int(slot.renter_user_id) 

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

92 return int(slot.maintenance_user_id) 

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

94 return int(slot.viewer_user_id) 

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

96 return int(slot.sole_pilot_user_id) 

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

98 return int(slot.sole_operator_user_id) 

99 if role == "shared_ownership" and slot.shared_ownership_user_id: 

100 return int(slot.shared_ownership_user_id) 

101 return int(slot.user_id) 

102 

103 

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

105 slot.last_activity_at = datetime.now(UTC) 

106 db.session.commit() 

107 

108 

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

110def next_wipe() -> ResponseReturnValue: 

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

112 from flask import jsonify 

113 

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

115 

116 

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

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

119 cutoff = datetime.now(UTC) - timedelta(minutes=window_minutes) 

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