300 lines
12 KiB
Python
300 lines
12 KiB
Python
"""Self-update from the Gitea releases page.
|
|
|
|
Polls `<UPDATE_REPO_API>/releases/latest` for a tag newer than
|
|
``bolt_pet.__version__`` and, if there is one, moves the checkout to that tag
|
|
and restarts the pet. The install is expected to be a git clone (which is how
|
|
it's deployed), so "download the update" is just `git fetch` + `git checkout`
|
|
— atomic, and the previous ref is one command away if anything goes wrong.
|
|
|
|
Safety rules, in the order they're enforced:
|
|
|
|
1. **A dirty working tree is never touched.** Local edits are skipped over,
|
|
not stashed — the pet silently discarding your work-in-progress would be
|
|
far worse than running an old version.
|
|
2. **Everything after checkout is guarded.** Dependency install and an import
|
|
smoke test both run before the restart; if either fails, the checkout is
|
|
rolled back to the exact ref that was live before (branch name if we were
|
|
on one, otherwise the commit) and the deps reinstalled from it.
|
|
3. **The restart only happens once the new code imports.** So a broken
|
|
release costs you a rollback and a log line, not a pet that won't start.
|
|
|
|
The git side goes through an injectable *run* callable — ``(args) ->
|
|
(returncode, output)`` — so the whole apply/rollback dance is unit-tested
|
|
against a fake git rather than a real repo. Version comparison and release
|
|
parsing are pure functions for the same reason.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Callable, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
from . import config
|
|
|
|
# (returncode, combined stdout+stderr)
|
|
GitResult = Tuple[int, str]
|
|
GitRunner = Callable[[list], GitResult]
|
|
|
|
_GIT_TIMEOUT_SECONDS = 300
|
|
|
|
|
|
class UpdateError(Exception):
|
|
"""Raised when an update can't be applied. If it's raised *after* the
|
|
checkout moved, the rollback has already run."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Release:
|
|
tag: str
|
|
name: str
|
|
body: str
|
|
prerelease: bool
|
|
|
|
|
|
# ── pure logic ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def parse_version(tag: str) -> tuple:
|
|
"""``"v1.2.3"`` -> ``(1, 2, 3)``. Leading "v" optional; a trailing
|
|
suffix ends the parse (``"1.2.3-beta1"`` -> ``(1, 2, 3)``), so a
|
|
prerelease of a version compares equal to it rather than sorting
|
|
randomly. Junk parses to ``()``, which is never newer than anything."""
|
|
parts: list[int] = []
|
|
for chunk in (tag or "").strip().lstrip("vV").split("."):
|
|
digits = ""
|
|
for char in chunk:
|
|
if not char.isdigit():
|
|
break
|
|
digits += char
|
|
if not digits:
|
|
break
|
|
parts.append(int(digits))
|
|
return tuple(parts)
|
|
|
|
|
|
def is_newer(candidate: str, current: str) -> bool:
|
|
"""True if *candidate* is a strictly newer version than *current*.
|
|
Compares zero-padded, so 1.2 == 1.2.0 and 1.2.1 > 1.2."""
|
|
new, old = parse_version(candidate), parse_version(current)
|
|
if not new:
|
|
return False
|
|
width = max(len(new), len(old))
|
|
return new + (0,) * (width - len(new)) > old + (0,) * (width - len(old))
|
|
|
|
|
|
def release_from_payload(payload: dict) -> Optional[Release]:
|
|
"""Gitea's release JSON -> Release, or None if it's a draft or has no
|
|
tag. /releases/latest already excludes drafts and prereleases, but the
|
|
same parser is used for the full list."""
|
|
if not isinstance(payload, dict) or payload.get("draft"):
|
|
return None
|
|
tag = str(payload.get("tag_name") or "").strip()
|
|
if not tag:
|
|
return None
|
|
return Release(
|
|
tag=tag,
|
|
name=str(payload.get("name") or tag),
|
|
body=str(payload.get("body") or ""),
|
|
prerelease=bool(payload.get("prerelease")),
|
|
)
|
|
|
|
|
|
# ── talking to Gitea ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def fetch_latest_release(api_url: str = None, token: str = None, timeout: float = 15.0) -> Optional[Release]:
|
|
"""Newest published release, or None if the repo has no releases yet
|
|
(a fresh repo 404s here, which is not an error worth logging every hour).
|
|
Raises UpdateError if the server is unreachable or answers with junk."""
|
|
api_url = (api_url if api_url is not None else config.UPDATE_REPO_API).rstrip("/")
|
|
if not api_url:
|
|
raise UpdateError("UPDATE_REPO_API is not set")
|
|
token = config.UPDATE_TOKEN if token is None else token
|
|
headers = {"Authorization": f"token {token}"} if token else {}
|
|
try:
|
|
response = requests.get(f"{api_url}/releases/latest", headers=headers, timeout=timeout)
|
|
except Exception as exc:
|
|
raise UpdateError(f"couldn't reach the releases API: {exc}") from exc
|
|
if response.status_code == 404:
|
|
return None
|
|
try:
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except Exception as exc:
|
|
raise UpdateError(f"bad response from the releases API: {exc}") from exc
|
|
return release_from_payload(payload)
|
|
|
|
|
|
def check_for_update(current_version: str = None, **kwargs) -> Optional[Release]:
|
|
"""The whole "is there anything new?" question in one call. Returns the
|
|
Release to move to, or None if we're already current."""
|
|
from . import __version__
|
|
|
|
current = __version__ if current_version is None else current_version
|
|
release = fetch_latest_release(**kwargs)
|
|
if release is None or not is_newer(release.tag, current):
|
|
return None
|
|
return release
|
|
|
|
|
|
# ── git ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def git_runner(repo: Path = None) -> GitRunner:
|
|
repo = Path(repo or config.HERE)
|
|
|
|
def run(args: list) -> GitResult:
|
|
try:
|
|
completed = subprocess.run(
|
|
["git", *args], cwd=str(repo), capture_output=True,
|
|
text=True, timeout=_GIT_TIMEOUT_SECONDS,
|
|
)
|
|
except Exception as exc:
|
|
return 1, f"git {' '.join(args)} failed to start: {exc}"
|
|
return completed.returncode, ((completed.stdout or "") + (completed.stderr or "")).strip()
|
|
|
|
return run
|
|
|
|
|
|
def is_git_clone(run: GitRunner) -> bool:
|
|
return run(["rev-parse", "--git-dir"])[0] == 0
|
|
|
|
|
|
def working_tree_dirty(run: GitRunner) -> bool:
|
|
code, output = run(["status", "--porcelain"])
|
|
return code != 0 or bool(output.strip())
|
|
|
|
|
|
def current_ref(run: GitRunner) -> str:
|
|
"""The branch name if we're on one, else the commit SHA — i.e. whatever
|
|
`git checkout` needs to put things back exactly as they were."""
|
|
code, output = run(["symbolic-ref", "--quiet", "--short", "HEAD"])
|
|
if code == 0 and output.strip():
|
|
return output.strip()
|
|
code, output = run(["rev-parse", "HEAD"])
|
|
if code != 0 or not output.strip():
|
|
raise UpdateError("couldn't work out the current git ref")
|
|
return output.strip()
|
|
|
|
|
|
def _requirements_changed(run: GitRunner, before: str, after: str) -> bool:
|
|
code, output = run(["diff", "--name-only", before, after, "--", "requirements.txt"])
|
|
return code == 0 and bool(output.strip())
|
|
|
|
|
|
def _install_deps(repo: Path) -> None:
|
|
completed = subprocess.run(
|
|
[sys.executable, "-m", "pip", "install", "-r", "requirements.txt"],
|
|
cwd=str(repo), capture_output=True, text=True, timeout=_GIT_TIMEOUT_SECONDS,
|
|
)
|
|
if completed.returncode != 0:
|
|
raise UpdateError(f"pip install failed: {(completed.stderr or '')[-500:]}")
|
|
|
|
|
|
def _smoke_test(repo: Path) -> None:
|
|
"""Import the freshly checked-out package in a *subprocess* — this one
|
|
still has the old modules loaded, so importing here would prove nothing.
|
|
Catches the common broken release (syntax error, missing dependency)
|
|
before we hand the session over to it."""
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", "import bolt_pet; import bolt_pet.controller"],
|
|
cwd=str(repo), capture_output=True, text=True, timeout=120,
|
|
)
|
|
if completed.returncode != 0:
|
|
raise UpdateError(f"the new version failed to import: {(completed.stderr or '')[-500:]}")
|
|
|
|
|
|
def apply_update(
|
|
tag: str,
|
|
run: GitRunner = None,
|
|
repo: Path = None,
|
|
on_log: Callable[[str], None] = lambda _msg: None,
|
|
install_deps: bool = None,
|
|
verify: Callable[[Path], None] = None,
|
|
) -> str:
|
|
"""Move the checkout to *tag*, rolling back to where it was if anything
|
|
downstream of the checkout fails. Returns the ref we came from (handy for
|
|
logging / a manual `git checkout` back). Raises UpdateError otherwise."""
|
|
repo = Path(repo or config.HERE)
|
|
run = run or git_runner(repo)
|
|
install_deps = config.UPDATE_INSTALL_DEPS if install_deps is None else install_deps
|
|
verify = _smoke_test if verify is None else verify
|
|
|
|
if not is_git_clone(run):
|
|
raise UpdateError("not a git clone — auto-update only works on a git checkout")
|
|
if working_tree_dirty(run):
|
|
raise UpdateError("working tree has local changes — skipping (nothing was touched)")
|
|
|
|
previous = current_ref(run)
|
|
code, output = run(["fetch", "--tags", "--prune", config.UPDATE_GIT_REMOTE])
|
|
if code != 0:
|
|
raise UpdateError(f"git fetch failed: {output}")
|
|
|
|
code, output = run(["checkout", "--force", f"tags/{tag}"])
|
|
if code != 0:
|
|
raise UpdateError(f"git checkout {tag} failed: {output}")
|
|
on_log(f"Checked out {tag} (was {previous}).")
|
|
|
|
# Past this point every failure has to put the checkout back.
|
|
try:
|
|
if install_deps and _requirements_changed(run, previous, f"tags/{tag}"):
|
|
on_log("requirements.txt changed — installing.")
|
|
_install_deps(repo)
|
|
verify(repo)
|
|
except UpdateError as exc:
|
|
_rollback(run, previous, repo, on_log, install_deps)
|
|
raise UpdateError(f"{exc} — rolled back to {previous}") from exc
|
|
except Exception as exc: # a verify() that blows up is still a failed update
|
|
_rollback(run, previous, repo, on_log, install_deps)
|
|
raise UpdateError(f"update failed ({exc}) — rolled back to {previous}") from exc
|
|
|
|
return previous
|
|
|
|
|
|
def _rollback(
|
|
run: GitRunner,
|
|
previous: str,
|
|
repo: Path,
|
|
on_log: Callable[[str], None],
|
|
install_deps: bool,
|
|
) -> None:
|
|
"""Best-effort return to *previous*. Never raises — it's already running
|
|
inside a failure path, and the caller's UpdateError is the thing worth
|
|
surfacing. A rollback that itself fails gets its own loud log line,
|
|
because that's the one case needing a human."""
|
|
code, output = run(["checkout", "--force", previous])
|
|
if code != 0:
|
|
on_log(f"ROLLBACK FAILED — the checkout is stranded. Run: git checkout {previous} ({output})")
|
|
return
|
|
on_log(f"Rolled back to {previous}.")
|
|
if install_deps:
|
|
try:
|
|
_install_deps(repo)
|
|
except Exception as exc:
|
|
on_log(f"Rolled back, but reinstalling the old requirements failed: {exc}")
|
|
|
|
|
|
# ── restart ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def restart() -> None:
|
|
"""Replace this process with a fresh `python -m bolt_pet`.
|
|
|
|
execv rather than spawn-and-exit so there's no window with two pets
|
|
holding the same mic, and no orphan if the parent dies first. Never
|
|
returns when it works; callers should have shut the Qt app and released
|
|
the audio device before calling it.
|
|
|
|
chdir first because `-m bolt_pet` resolves the package from the working
|
|
directory: the pet may well have been launched from somewhere else
|
|
(autostart entry, run.sh invoked by path), and the new process has to
|
|
land on the checkout the update was just applied to."""
|
|
os.chdir(str(config.HERE))
|
|
os.execv(sys.executable, [sys.executable, "-m", "bolt_pet"])
|