From 0cfd42e56204e57d8886819c86b7a98e982108bc Mon Sep 17 00:00:00 2001 From: ethernet Date: Thu, 24 Sep 2026 11:38:52 -0400 Subject: [PATCH] refactor(cli): split source_check.check_for_updates into cache, channel, fetch helpers check_for_updates had grown to ~CC 75 over 128 lines, mixing install gating, checkout probing, cache freshness, channel resolution, remote-tip fetch and result shaping. It now reads as orchestration over cohesive helpers: _unsupported_reason, _read_checkout (_Checkout facts), _cached_status / _write_cache, _resolve_channel, _check_branch (_branch_remote, _heal_deleted_branch, _behind_count). Every function is below CC 30 (check_for_updates 26). Behaviour, public signature and return shape are unchanged; module-level seams tests patch (_git_stdout, _git_ok, _branch_tip, resolve_source_target, time, ...) are still read from this module at call time. --- hermes_cli/source_check.py | 272 ++++++++++++++++++++++++------------- 1 file changed, 178 insertions(+), 94 deletions(-) diff --git a/hermes_cli/source_check.py b/hermes_cli/source_check.py index bab00823a6..bda26d17ab 100644 --- a/hermes_cli/source_check.py +++ b/hermes_cli/source_check.py @@ -9,6 +9,7 @@ import logging import os import subprocess import time +from dataclasses import dataclass from pathlib import Path from typing import Optional from urllib.parse import quote @@ -183,6 +184,159 @@ def _commits(payload: dict | None) -> list[dict]: return rows[::-1] +@dataclass(frozen=True) +class _Checkout: + """Read-only facts about the checkout under test, gathered once per check.""" + root: Path + git: str + embedded: Optional[str] + head: Optional[str] + current_branch: Optional[str] + origin: str + repository: Optional[str] + dirty: bool + + +def _read_json(path: Path): + return _quiet(lambda: json.loads(path.read_text(encoding="utf-8-sig"))) + + +def _unsupported_reason(stamp: dict, root: Path, *, explicit_root: bool, embedded: Optional[str]) -> Optional[dict]: + """Fields explaining why this install cannot self-update from Git, or None when it can.""" + from hermes_cli.config import detect_install_method + from hermes_cli.update_contract import COMMIT_BUILD_UPDATE_MESSAGE + + if stamp.get("source") == "commit-build": + return {"reason": "commit-build", "message": COMMIT_BUILD_UPDATE_MESSAGE} + if stamp.get("payload") in {"bundled", "light", "runtime"} or ( + not explicit_root and detect_install_method(root) in {"docker", "apt"}): + return {"reason": "not-a-git-checkout"} + if not embedded and not (root / ".git").exists(): + return {"reason": "not-a-git-checkout", "message": "This install has no git checkout to update."} + if stamp.get("updateMechanism") not in (None, "self") and not embedded: + return {"reason": "update-root-steward-owned-git-tree", + "message": "This installation is managed by its install method; use its updater.", "advice": "git pull"} + return None + + +def _read_checkout(root: Path, git: str, embedded: Optional[str]) -> _Checkout: + # An embedded revision has no checkout to ask: it always tracks the official repository. + head = embedded or _git_stdout(["rev-parse", "HEAD"], cwd=root, git=git) + current_branch = None if embedded else _git_stdout(["rev-parse", "--abbrev-ref", "HEAD"], cwd=root, git=git) + origin = "" if embedded else (_git_stdout(["remote", "get-url", "origin"], cwd=root, git=git) or "") + match = _GITHUB_ORIGIN.fullmatch(origin) + repository = OFFICIAL_REPOSITORY if embedded else (match[1] if match else None) + dirty = False if embedded else bool(_git_stdout(["status", "--porcelain"], cwd=root, git=git)) + return _Checkout(root, git, embedded, head, current_branch, origin, repository, dirty) + + +def _configured_branch(desktop_config) -> Optional[str]: + value = desktop_config.get("branch") if isinstance(desktop_config, dict) else None + return (value.strip() or None) if isinstance(value, str) else None + + +def _checked_out_branch(current_branch: Optional[str], fallback: Optional[str]) -> Optional[str]: + """The checkout's branch, or ``fallback`` when detached or unreadable.""" + return current_branch if current_branch and current_branch != "HEAD" else fallback + + +def _cached_status(cache_file: Path, identity: dict, now: float) -> Optional[dict]: + """A still-fresh supported status cached for exactly this identity, else None. + + Failures expire sooner so a transient network error does not hide updates for a day. + """ + cached = _read_json(cache_file) + if not (isinstance(cached, dict) and cached.get("identity") == identity + and isinstance(cached.get("status"), dict) and cached["status"].get("supported") is True): + return None + status = cached.get("status", {}) + ttl = _UPDATE_CHECK_FAILURE_CACHE_SECONDS if status.get("error") else _UPDATE_CHECK_CACHE_SECONDS + ts = cached.get("ts") + return status if isinstance(ts, (float, int)) and 0 <= now - ts < ttl else None + + +def _write_cache(cache_file: Path, identity: dict, now: float, result: dict) -> None: + try: + from utils import atomic_json_write + cache_file.parent.mkdir(parents=True, exist_ok=True) + atomic_json_write(cache_file, {"identity": identity, "ts": now, "status": result}) + except OSError as exc: + logger.debug("Could not cache source check: %s", exc) + + +def _resolve_channel(result: dict, channel: str, co: _Checkout): + """Resolve a release channel's target into ``result``; the SourceTarget, or None on error. + + A target with a pinned commit is final; one without names a branch to follow instead. + """ + try: + source_target = resolve_source_target(channel, [co.git] if not co.embedded else None, co.root, + repository=co.repository or OFFICIAL_REPOSITORY) + except (OSError, ValueError, subprocess.SubprocessError) as exc: + result.update(error="release-unavailable", message=f"Could not resolve the {channel} source channel: {exc}") + return None + if source_target.commit: + target = source_target.commit + result.pop("branch", None) + result.update(channel=channel, targetSha=target, updateAvailable=co.head != target, + behind=0 if co.head == target else UPDATE_AVAILABLE_NO_COUNT, + sourceVersion=source_target.version, buildId=source_target.build_id) + if source_target.retired: + result["retirement"] = {"destination": source_target.channel, "sourceOnly": True} + return source_target + + +def _branch_remote(co: _Checkout, selected_branch: str) -> str: + official_ssh = (co.repository and co.repository.lower() == OFFICIAL_REPOSITORY.lower() + and co.origin.lower().startswith(("git@", "ssh://"))) + # The public official repo does not require the user's SSH credentials. + # Forks must keep their own origin, including its authentication. + return (f"https://github.com/{OFFICIAL_REPOSITORY}.git" + if co.embedded or (official_ssh and selected_branch != "main") else "origin") + + +def _heal_deleted_branch(branch_config_path: Path, desktop_config: dict) -> None: + """Point the Desktop branch setting back at main after its branch was deleted upstream.""" + # Do not overwrite a concurrent choice made while the network probe ran. + if _read_json(branch_config_path) == desktop_config: + from utils import atomic_json_write + atomic_json_write(branch_config_path, {**desktop_config, "branch": "main"}) + + +def _behind_count(co: _Checkout, target: str) -> tuple[int, list[dict]]: + """``(behind, commits)`` for ``target``: local ancestry first, then the GitHub compare API.""" + if co.head == target or (not co.embedded and _git_ok( + ["merge-base", "--is-ancestor", target, co.head], cwd=co.root, git=co.git)): + return 0, [] + if co.repository: + payload = _github_compare(co.head, target, co.repository) + ahead = (payload or {}).get("ahead_by") + if isinstance(ahead, int) and not isinstance(ahead, bool) and ahead >= 0: + return ahead, (_quiet(lambda: _commits(payload), []) if ahead else []) + return UPDATE_AVAILABLE_NO_COUNT, [] + + +def _check_branch(result: dict, co: _Checkout, selected_branch: str, *, + heal: Optional[tuple[Path, dict]]) -> None: + """Compare the checkout with ``selected_branch``'s remote tip, falling back to main if it was deleted.""" + result["branch"] = selected_branch + remote = _branch_remote(co, selected_branch) + target, missing, failure = _branch_tip(co.repository, selected_branch, co.root, co.git, remote) + if missing and selected_branch != "main": + result["branch"] = "main" + if heal: + _heal_deleted_branch(*heal) + target, _, failure = _branch_tip(co.repository, "main", co.root, co.git, remote if co.embedded else "origin") + if target is None: + result.update(error="fetch-failed", + message=f"Could not resolve the remote branch tip: {failure}" if failure + else "Could not resolve the remote branch tip.") + return + behind, commits = _behind_count(co, target) + result["commits"] = commits + result.update(targetSha=target, behind=behind, updateAvailable=behind != 0) + + def check_for_updates(*, install_root: Path | None = None, home: Path | None = None, branch: str | None = None, channel: str | None = None, cache_path: Path | None = None, branch_config_path: Path | None = None, @@ -193,124 +347,54 @@ def check_for_updates(*, install_root: Path | None = None, home: Path | None = N Only the default (running installation) may use HERMES_REVISION. An explicit target must never inherit the host process's embedded revision or stamp. """ - from hermes_cli.config import detect_install_method, get_project_root, require_readable_config_before_write + from hermes_cli.config import get_project_root, require_readable_config_before_write from hermes_cli.steward import read_install_stamp from hermes_cli.update_channel import install_id, resolve_update_channel from hermes_cli.release_channels import validate_name - from hermes_cli.update_contract import COMMIT_BUILD_UPDATE_MESSAGE embedded = (os.environ.get("HERMES_REVISION") or None) if install_root is None else None root = Path(install_root if install_root is not None else get_project_root()).resolve() home = Path(home if home is not None else get_hermes_home()).resolve() - stamp = read_install_stamp(root) result = {"supported": False, "hermesRoot": str(root), "behind": None, "commits": []} - if stamp.get("source") == "commit-build": - return {**result, "reason": "commit-build", "message": COMMIT_BUILD_UPDATE_MESSAGE} - if stamp.get("payload") in {"bundled", "light", "runtime"} or (install_root is None and detect_install_method(root) in {"docker", "apt"}): - return {**result, "reason": "not-a-git-checkout"} - if not embedded and not (root / ".git").exists(): - return {**result, "reason": "not-a-git-checkout", - "message": "This install has no git checkout to update."} - if stamp.get("updateMechanism") not in (None, "self") and not embedded: - return {**result, "reason": "update-root-steward-owned-git-tree", - "message": "This installation is managed by its install method; use its updater.", "advice": "git pull"} + unsupported = _unsupported_reason(read_install_stamp(root), root, + explicit_root=install_root is not None, embedded=embedded) + if unsupported: + return {**result, **unsupported} config = require_readable_config_before_write(home / "config.yaml") if passive and (config.get("updates") or {}).get("check") is False: return {**result, "reason": "disabled"} channel = resolve_update_channel(config, root) if channel is None else validate_name(channel) - head = embedded or _git_stdout(["rev-parse", "HEAD"], cwd=root, git=git) - current_branch = None if embedded else _git_stdout(["rev-parse", "--abbrev-ref", "HEAD"], cwd=root, git=git) - desktop_config = _quiet(lambda: json.loads(branch_config_path.read_text(encoding="utf-8-sig"))) if branch_config_path else None - configured_branch = desktop_config.get("branch") if isinstance(desktop_config, dict) else None - if isinstance(configured_branch, str): - configured_branch = configured_branch.strip() or None - else: - configured_branch = None - selected_branch = branch or configured_branch or (current_branch if current_branch and current_branch != "HEAD" else "main") - origin = "" if embedded else (_git_stdout(["remote", "get-url", "origin"], cwd=root, git=git) or "") - match = _GITHUB_ORIGIN.fullmatch(origin) - repository = OFFICIAL_REPOSITORY if embedded else (match[1] if match else None) - dirty = False if embedded else bool(_git_stdout(["status", "--porcelain"], cwd=root, git=git)) - result.update(supported=True, currentSha=head, currentBranch=current_branch, dirty=dirty) + co = _read_checkout(root, git, embedded) + desktop_config = _read_json(branch_config_path) if branch_config_path else None + configured_branch = _configured_branch(desktop_config) + selected_branch = branch or configured_branch or _checked_out_branch(co.current_branch, "main") + result.update(supported=True, currentSha=co.head, currentBranch=co.current_branch, dirty=co.dirty) if channel != "main": result["channel"] = channel else: result["branch"] = selected_branch - identity = {"root": str(root), "home": str(home), "head": head, "origin": origin, "branch": selected_branch, + identity = {"root": str(root), "home": str(home), "head": co.head, "origin": co.origin, "branch": selected_branch, "channel": channel, "embedded": embedded, "branchOverride": branch is not None, "channelProtocol": 1} cache_file = Path(cache_path) if cache_path is not None else home / "source-checks" / f"{install_id(root)}.json" - cached = _quiet(lambda: json.loads(cache_file.read_text(encoding="utf-8-sig"))) now = time.time() - if (not force and isinstance(cached, dict) and cached.get("identity") == identity - and isinstance(cached.get("status"), dict) and cached["status"].get("supported") is True): - status = cached.get("status", {}) - ttl = _UPDATE_CHECK_FAILURE_CACHE_SECONDS if status.get("error") else _UPDATE_CHECK_CACHE_SECONDS - ts = cached.get("ts") - if isinstance(ts, (float, int)) and 0 <= now - ts < ttl: - return {**status, "dirty": dirty, "currentBranch": current_branch} + cached = None if force else _cached_status(cache_file, identity, now) + if cached is not None: + return {**cached, "dirty": co.dirty, "currentBranch": co.current_branch} result["fetchedAt"] = int(now * 1000) source_target = None - if not _is_full_sha(head): + if not _is_full_sha(co.head): result.update(error="head-unavailable", message="Could not read the installed revision.") elif branch is None: - try: - source_target = resolve_source_target(channel, [git] if not embedded else None, root, - repository=repository or OFFICIAL_REPOSITORY) - except (OSError, ValueError, subprocess.SubprocessError) as exc: - result.update(error="release-unavailable", message=f"Could not resolve the {channel} source channel: {exc}") - else: - if source_target.commit: - target = source_target.commit - result.pop("branch", None) - result.update(channel=channel, targetSha=target, updateAvailable=head != target, - behind=0 if head == target else UPDATE_AVAILABLE_NO_COUNT, - sourceVersion=source_target.version, buildId=source_target.build_id) - if source_target.retired: - result["retirement"] = {"destination": source_target.channel, "sourceOnly": True} - else: - # The record supplies a default, not permission to leave the user's branch. - selected_branch = configured_branch or ( - current_branch if current_branch and current_branch != "HEAD" else source_target.branch) + source_target = _resolve_channel(result, channel, co) + if source_target is not None and not source_target.commit: + # The record supplies a default, not permission to leave the user's branch. + selected_branch = configured_branch or _checked_out_branch(co.current_branch, source_target.branch) if "error" not in result and (source_target is None or source_target.branch is not None): - result["branch"] = selected_branch - official_ssh = (repository and repository.lower() == OFFICIAL_REPOSITORY.lower() - and origin.lower().startswith(("git@", "ssh://"))) - # The public official repo does not require the user's SSH credentials. - # Forks must keep their own origin, including its authentication. - remote = (f"https://github.com/{OFFICIAL_REPOSITORY}.git" - if embedded or (official_ssh and selected_branch != "main") else "origin") - target, missing, failure = _branch_tip(repository, selected_branch, root, git, remote) - if missing and selected_branch != "main": - result["branch"] = "main" - if branch_config_path and not branch and configured_branch == selected_branch: - # Do not overwrite a concurrent choice made while the network probe ran. - current_config = _quiet(lambda: json.loads(branch_config_path.read_text(encoding="utf-8-sig"))) - if current_config == desktop_config: - from utils import atomic_json_write - atomic_json_write(branch_config_path, {**desktop_config, "branch": "main"}) - target, _, failure = _branch_tip(repository, "main", root, git, remote if embedded else "origin") - if target is None: - result.update(error="fetch-failed", - message=f"Could not resolve the remote branch tip: {failure}" if failure - else "Could not resolve the remote branch tip.") - else: - behind = UPDATE_AVAILABLE_NO_COUNT - if head == target or (not embedded and _git_ok( - ["merge-base", "--is-ancestor", target, head], cwd=root, git=git)): - behind = 0 - elif repository: - payload = _github_compare(head, target, repository) - ahead = (payload or {}).get("ahead_by") - if isinstance(ahead, int) and not isinstance(ahead, bool) and ahead >= 0: - behind = ahead - result["commits"] = _quiet(lambda: _commits(payload), []) if behind else [] - result.update(targetSha=target, behind=behind, updateAvailable=behind != 0) - try: - from utils import atomic_json_write - cache_file.parent.mkdir(parents=True, exist_ok=True) - atomic_json_write(cache_file, {"identity": identity, "ts": now, "status": result}) - except OSError as exc: - logger.debug("Could not cache source check: %s", exc) + # Only a Desktop-configured branch the caller did not override is healed. + heal = branch_config_path and not branch and configured_branch == selected_branch + _check_branch(result, co, selected_branch, + heal=(branch_config_path, desktop_config) if heal else None) + _write_cache(cache_file, identity, now, result) return result