Coverage for app/services/version_service.py: 100%
84 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""
2Version-check service — GitHub Pages versions list, AppSetting cache, background thread.
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"""
8from datetime import UTC
9from typing import Any
10from urllib.parse import urlparse
12_VERSION_CHECK_HOST = "e2jk.github.io"
13_VERSIONS_JSON_URL = f"https://{_VERSION_CHECK_HOST}/OpenHangar/versions.json"
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
22 class _StrictRedirect(urllib.request.HTTPRedirectHandler):
23 """Block any redirect that leaves the allowed host."""
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)
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 []
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
54def upsert_app_setting(db_session: Any, key: str, value: str) -> None:
55 from models import AppSetting # pyright: ignore[reportMissingImports]
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))
64def is_newer_version(current: str, latest: "str | None") -> bool:
65 """True when `latest` is a real, newer release than `current`.
67 Shared by the cached update_available flag below (config page banner)
68 and by the auto-upgrade check (services/backup_scheduler.py), so both
69 agree on what counts as "an update is available". Never raises — a
70 malformed version string just means "no update".
71 """
72 try:
73 from packaging.version import Version # pyright: ignore[reportMissingImports]
75 return bool(
76 latest and current != "development" and Version(latest) > Version(current)
77 )
78 except Exception: # noqa: BLE001 -- cosmetic comparison, malformed input means "no update"
79 return False
82def _persist_update_flag(db_session: Any, current: str, latest: str | None) -> None:
83 """Compute update_available and write it to AppSetting. Silently ignores errors."""
84 available = is_newer_version(current, latest)
85 upsert_app_setting(db_session, "update_available", "true" if available else "false")
88def run_version_check(app: Any) -> None:
89 """Check GitHub Pages for the versions list and cache results in AppSetting."""
90 import json
91 import os
92 from datetime import datetime, timedelta
94 from models import AppSetting, db # pyright: ignore[reportMissingImports]
96 with app.app_context():
97 last = db.session.get(AppSetting, "version_last_checked_at")
98 if last and last.value:
99 try:
100 if datetime.now(UTC) - datetime.fromisoformat(last.value) < timedelta(
101 hours=23
102 ):
103 return
104 except ValueError:
105 pass # malformed stored timestamp — proceed with the check
107 versions = fetch_versions()
108 upsert_app_setting(
109 db.session,
110 "version_last_checked_at",
111 datetime.now(UTC).isoformat(),
112 )
113 if versions:
114 upsert_app_setting(db.session, "latest_version", versions[0])
115 upsert_app_setting(db.session, "all_versions", json.dumps(versions))
116 _persist_update_flag(
117 db.session,
118 os.environ.get("OPENHANGAR_VERSION", "development"),
119 versions[0],
120 )
121 db.session.commit()
124def startup_recompute_update_flag(app: Any) -> None:
125 """Recompute update_available against the currently-running version.
127 Called once at startup so that a freshly-deployed container immediately
128 reflects the correct flag value without waiting for the next scheduled
129 version check (which has a 0–6 h random delay).
130 """
131 import os
133 from models import AppSetting, db # pyright: ignore[reportMissingImports]
135 with app.app_context():
136 latest_s = db.session.get(AppSetting, "latest_version")
137 latest = latest_s.value if latest_s else None
138 _persist_update_flag(
139 db.session, os.environ.get("OPENHANGAR_VERSION", "development"), latest
140 )
141 db.session.commit()
144def version_check_loop(app: Any, _sleep_fn: Any = None) -> None:
145 """Daemon thread body — random startup delay then every 24 h."""
146 import random
147 import time as _time
149 sleep = _sleep_fn if _sleep_fn is not None else _time.sleep
150 sleep(random.randint(0, 6 * 3600))
151 while True:
152 try:
153 run_version_check(app)
154 except Exception:
155 app.logger.exception("Version check failed; will retry in 24 h")
156 sleep(24 * 3600)
159def start_version_check_thread(app: Any) -> None:
160 import threading
162 threading.Thread(
163 target=startup_recompute_update_flag,
164 args=(app,),
165 daemon=True,
166 name="version-flag-startup",
167 ).start()
168 threading.Thread(
169 target=version_check_loop,
170 args=(app,),
171 daemon=True,
172 name="version-check",
173 ).start()