Coverage for app/services/version_service.py: 100%

1""" 

2Version-check service — GitHub Pages versions list, AppSetting cache, background thread. 

3 

4Kept in services/ so both init.py (thread start) and config/routes.py 

5(force-refresh endpoint) can import from here without creating a cycle. 

6""" 

7 

8from datetime import UTC 

9from typing import Any 

10from urllib.parse import urlparse 

11 

12_VERSION_CHECK_HOST = "e2jk.github.io" 

13_VERSIONS_JSON_URL = f"https://{_VERSION_CHECK_HOST}/OpenHangar/versions.json" 

14 

15 

16def fetch_versions() -> list[str]: 

17 """Fetch the ordered list of all published versions from GitHub Pages. Returns [] on error.""" 

18 import json 

19 import urllib.error 

20 import urllib.request 

21 

22 class _StrictRedirect(urllib.request.HTTPRedirectHandler): 

23 """Block any redirect that leaves the allowed host.""" 

24 

25 def redirect_request( 

26 self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str 

27 ) -> Any: 

28 host = urlparse(newurl).netloc 

29 if host != _VERSION_CHECK_HOST: 

30 raise urllib.error.URLError( 

31 f"version-check redirect to {host!r} blocked" 

32 ) 

33 return super().redirect_request(req, fp, code, msg, headers, newurl) 

34 

35 opener = urllib.request.build_opener(_StrictRedirect) 

36 req = urllib.request.Request( 

37 _VERSIONS_JSON_URL, 

38 headers={"User-Agent": "OpenHangar-version-check"}, 

39 ) 

40 try: 

41 with opener.open(req, timeout=10) as resp: # nosec B310 

42 data = json.loads(resp.read()) 

43 return data if isinstance(data, list) else [] 

44 except Exception: # noqa: BLE001 -- non-critical background check, network/JSON errors mean "no update info" 

45 return [] 

46 

47 

48def fetch_latest_version() -> str | None: 

49 """Return the most recent published version, or None on error.""" 

50 versions = fetch_versions() 

51 return versions[0] if versions else None 

52 

53 

54def upsert_app_setting(db_session: Any, key: str, value: str) -> None: 

55 from models import AppSetting # pyright: ignore[reportMissingImports] 

56 

57 setting = db_session.get(AppSetting, key) 

58 if setting: 

59 setting.value = value 

60 else: 

61 db_session.add(AppSetting(key=key, value=value)) 

62 

63 

64def _persist_update_flag(db_session: Any, current: str, latest: str | None) -> None: 

65 """Compute update_available and write it to AppSetting. Silently ignores errors.""" 

66 try: 

67 from packaging.version import Version # pyright: ignore[reportMissingImports] 

68 

69 available = bool( 

70 latest and current != "development" and Version(latest) > Version(current) 

71 ) 

72 except Exception: # noqa: BLE001 -- docstring: silently ignores errors, cosmetic UI flag only 

73 available = False 

74 upsert_app_setting(db_session, "update_available", "true" if available else "false") 

75 

76 

77def run_version_check(app: Any) -> None: 

78 """Check GitHub Pages for the versions list and cache results in AppSetting.""" 

79 import json 

80 import os 

81 from datetime import datetime, timedelta 

82 

83 from models import AppSetting, db # pyright: ignore[reportMissingImports] 

84 

85 with app.app_context(): 

86 last = db.session.get(AppSetting, "version_last_checked_at") 

87 if last and last.value: 

88 try: 

89 if datetime.now(UTC) - datetime.fromisoformat(last.value) < timedelta( 

90 hours=23 

91 ): 

92 return 

93 except ValueError: 

94 pass # malformed stored timestamp — proceed with the check 

95 

96 versions = fetch_versions() 

97 upsert_app_setting( 

98 db.session, 

99 "version_last_checked_at", 

100 datetime.now(UTC).isoformat(), 

101 ) 

102 if versions: 

103 upsert_app_setting(db.session, "latest_version", versions[0]) 

104 upsert_app_setting(db.session, "all_versions", json.dumps(versions)) 

105 _persist_update_flag( 

106 db.session, 

107 os.environ.get("OPENHANGAR_VERSION", "development"), 

108 versions[0], 

109 ) 

110 db.session.commit() 

111 

112 

113def startup_recompute_update_flag(app: Any) -> None: 

114 """Recompute update_available against the currently-running version. 

115 

116 Called once at startup so that a freshly-deployed container immediately 

117 reflects the correct flag value without waiting for the next scheduled 

118 version check (which has a 0–6 h random delay). 

119 """ 

120 import os 

121 

122 from models import AppSetting, db # pyright: ignore[reportMissingImports] 

123 

124 with app.app_context(): 

125 latest_s = db.session.get(AppSetting, "latest_version") 

126 latest = latest_s.value if latest_s else None 

127 _persist_update_flag( 

128 db.session, os.environ.get("OPENHANGAR_VERSION", "development"), latest 

129 ) 

130 db.session.commit() 

131 

132 

133def version_check_loop(app: Any, _sleep_fn: Any = None) -> None: 

134 """Daemon thread body — random startup delay then every 24 h.""" 

135 import random 

136 import time as _time 

137 

138 sleep = _sleep_fn if _sleep_fn is not None else _time.sleep 

139 sleep(random.randint(0, 6 * 3600)) 

140 while True: 

141 try: 

142 run_version_check(app) 

143 except Exception: 

144 app.logger.exception("Version check failed; will retry in 24 h") 

145 sleep(24 * 3600) 

146 

147 

148def start_version_check_thread(app: Any) -> None: 

149 import threading 

150 

151 threading.Thread( 

152 target=startup_recompute_update_flag, 

153 args=(app,), 

154 daemon=True, 

155 name="version-flag-startup", 

156 ).start() 

157 threading.Thread( 

158 target=version_check_loop, 

159 args=(app,), 

160 daemon=True, 

161 name="version-check", 

162 ).start()