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 typing import Any 

9from urllib.parse import urlparse 

10 

11_VERSION_CHECK_HOST = "e2jk.github.io" 

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

13 

14 

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

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

17 import json 

18 import urllib.error 

19 import urllib.request 

20 

21 class _StrictRedirect(urllib.request.HTTPRedirectHandler): 

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

23 

24 def redirect_request( 

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

26 ) -> Any: 

27 host = urlparse(newurl).netloc 

28 if host != _VERSION_CHECK_HOST: 

29 raise urllib.error.URLError( 

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

31 ) 

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

33 

34 opener = urllib.request.build_opener(_StrictRedirect) 

35 req = urllib.request.Request( 

36 _VERSIONS_JSON_URL, 

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

38 ) 

39 try: 

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

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

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

43 except Exception: 

44 return [] 

45 

46 

47def fetch_latest_version() -> str | None: 

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

49 versions = fetch_versions() 

50 return versions[0] if versions else None 

51 

52 

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

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

55 

56 setting = db_session.get(AppSetting, key) 

57 if setting: 

58 setting.value = value 

59 else: 

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

61 

62 

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

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

65 import json 

66 from datetime import datetime, timedelta, timezone 

67 

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

69 

70 with app.app_context(): 

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

72 if last and last.value: 

73 try: 

74 if datetime.now(timezone.utc) - datetime.fromisoformat( 

75 last.value 

76 ) < timedelta(hours=23): 

77 return 

78 except ValueError: 

79 pass # malformed stored timestamp — proceed with the check 

80 

81 versions = fetch_versions() 

82 upsert_app_setting( 

83 db.session, 

84 "version_last_checked_at", 

85 datetime.now(timezone.utc).isoformat(), 

86 ) 

87 if versions: 

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

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

90 db.session.commit() 

91 

92 

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

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

95 import random 

96 import time as _time 

97 

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

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

100 while True: 

101 try: 

102 run_version_check(app) 

103 except Exception: 

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

105 sleep(24 * 3600) 

106 

107 

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

109 import threading 

110 

111 t = threading.Thread( 

112 target=version_check_loop, 

113 args=(app,), 

114 daemon=True, 

115 name="version-check", 

116 ) 

117 t.start()