diff --git a/README.md b/README.md index 4f108b25b..c254b1512 100644 --- a/README.md +++ b/README.md @@ -305,6 +305,23 @@ pipx install --python python3.13 "headroom-ai[all]" → [Installation guide](https://headroom-docs.vercel.app/docs/installation) — Docker tags, persistent service, PowerShell, devcontainers. +### Updating + +```bash +headroom update # detects pip / pipx / uv tool and upgrades in place +headroom update --check # report the latest release without upgrading +headroom update --pre # include pre-releases +``` + +`headroom update` figures out how Headroom was installed (pip/venv, `pip --user`, +pipx, uv tool) and runs the matching upgrade across macOS, Linux, and Windows. +For git checkouts, editable installs, Docker images, and externally-managed +system Pythons (PEP 668) it prints the correct manual step instead of guessing. + +The proxy also shows a one-line "update available" notice on startup. It checks +PyPI at most once a day, in the background, and never blocks. Opt out with +`HEADROOM_UPDATE_CHECK=off` (also skipped in `--stateless` mode and CI). + ### Corporate / SSL-inspection environments If `pip install "headroom-ai[all]"` fails with `CERTIFICATE_VERIFY_FAILED` diff --git a/headroom/cli/__init__.py b/headroom/cli/__init__.py index 626ea1d3d..883b04c46 100644 --- a/headroom/cli/__init__.py +++ b/headroom/cli/__init__.py @@ -24,6 +24,7 @@ from . import ( # noqa: F401 perf, proxy, tools, + update, wrap, ) from .main import main diff --git a/headroom/cli/main.py b/headroom/cli/main.py index 429d0b2a1..9059b967e 100644 --- a/headroom/cli/main.py +++ b/headroom/cli/main.py @@ -28,9 +28,21 @@ def main(ctx: click.Context) -> None: headroom proxy Start the optimization proxy headroom memory list List stored memories headroom memory stats Show memory statistics + headroom update Update Headroom to the latest release """ ctx.ensure_object(dict) + # Fire a rate-limited, opt-out background check for newer releases so other + # surfaces (e.g. the proxy banner) can show an "update available" notice. + # Never blocks, never raises; skipped for `update` (it checks explicitly). + if ctx.invoked_subcommand != "update": + try: + from headroom.update_check import maybe_check_async + + maybe_check_async() + except Exception: # noqa: BLE001 — update check must never break the CLI + pass + # Import subcommands - these register themselves with the main group def _register_commands() -> None: @@ -50,6 +62,7 @@ def _register_commands() -> None: perf, # noqa: F401 proxy, # noqa: F401 tools, # noqa: F401 + update, # noqa: F401 wrap, # noqa: F401 ) diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py index dc216c3ab..bf5e229e1 100644 --- a/headroom/cli/proxy.py +++ b/headroom/cli/proxy.py @@ -1107,6 +1107,17 @@ Endpoints: Press Ctrl+C to stop. """) + # Surface an "update available" notice (reads cache only; no network here). + # Best-effort: a broken update check must never block proxy startup. + try: + from headroom.update_check import format_update_notice + + _update_notice = format_update_notice() + if _update_notice: + click.echo(f"\n{_update_notice}\n") + except Exception: # noqa: BLE001 — banner must never crash startup + pass + # ----------------------------------------------------------------------- # Option E: start embedding server sidecar if requested # ----------------------------------------------------------------------- diff --git a/headroom/cli/update.py b/headroom/cli/update.py new file mode 100644 index 000000000..93975201b --- /dev/null +++ b/headroom/cli/update.py @@ -0,0 +1,341 @@ +"""`headroom update` — self-update across the supported install methods. + +Detects how Headroom was installed and runs the matching upgrade command, or +refuses with clear guidance when in-tool self-update isn't appropriate (git +checkout, editable install, Docker image, system package manager). + +The upgrade always runs through ``sys.executable -m pip`` for the pip path so +it can never touch a different interpreter than the one actually running +Headroom. +""" + +from __future__ import annotations + +import json +import os +import shlex +import subprocess +import sys +from dataclasses import dataclass + +import click + +from headroom.update_check import ( + PACKAGE_NAME, + fetch_latest_version, + installed_version, + write_cache, +) + +from .main import main + + +@dataclass(frozen=True) +class InstallMethod: + """How Headroom is installed and how (or whether) to upgrade it.""" + + kind: str + can_self_update: bool + argv: list[str] | None = None + guidance: str | None = None + + +def _is_source_checkout() -> bool: + try: + from headroom._version import _source_root + + return _source_root() is not None + except Exception: + return False + + +def _is_editable_install() -> bool: + """Detect a PEP 660 editable install via its ``direct_url.json``.""" + try: + from importlib.metadata import distribution + + raw = distribution(PACKAGE_NAME).read_text("direct_url.json") + if not raw: + return False + data = json.loads(raw) + dir_info = data.get("dir_info") + return bool(isinstance(dir_info, dict) and dir_info.get("editable")) + except Exception: + return False + + +def _in_docker() -> bool: + try: + from pathlib import Path + + return Path("/.dockerenv").exists() or bool( + os.environ.get("HEADROOM_IN_DOCKER", "").strip() + ) + except Exception: + return False + + +def _in_virtualenv() -> bool: + """True inside a venv/virtualenv or a conda environment.""" + if getattr(sys, "prefix", "") != getattr(sys, "base_prefix", ""): + return True + # Conda envs often share prefix==base_prefix for `base`; pip -U is still safe + # inside any conda env, so treat an active CONDA_PREFIX as an environment. + return bool(os.environ.get("CONDA_PREFIX", "").strip()) + + +def _norm(path: str | os.PathLike[str] | None) -> str: + """Resolve + lowercase a path for cross-platform substring matching.""" + if not path: + return "" + try: + from pathlib import Path + + return str(Path(path).resolve()).replace("\\", "/").lower() + except Exception: + return str(path).replace("\\", "/").lower() + + +def _package_location() -> str | None: + """Normalized base directory the headroom-ai distribution is installed in.""" + try: + from importlib.metadata import distribution + + return _norm(str(distribution(PACKAGE_NAME).locate_file(""))) + except Exception: + return None + + +def _user_site() -> str | None: + try: + import site + + return _norm(site.getusersitepackages()) + except Exception: + return None + + +def _is_user_site_install(location: str | None) -> bool: + user = _user_site() + if not user or not location: + return False + # Path-segment containment so "/.../site" never matches "/.../site-packages". + return location == user or location.startswith(user.rstrip("/") + "/") + + +def _format_cmd(argv: list[str]) -> str: + """Render argv as a copy-pasteable shell string (handles spaces in paths).""" + if sys.platform.startswith("win"): + return subprocess.list2cmdline(argv) + return shlex.join(argv) + + +def _is_externally_managed() -> bool: + """Detect a PEP 668 ``EXTERNALLY-MANAGED`` marker (Homebrew, Debian, etc.).""" + try: + import sysconfig + from pathlib import Path + + for key in ("stdlib", "platstdlib", "purelib"): + base = sysconfig.get_path(key) + if base and Path(base, "EXTERNALLY-MANAGED").exists(): + return True + except Exception: + return False + return False + + +def _spec(extras: str | None) -> str: + extras = (extras or "").strip().strip("[]") + return f"{PACKAGE_NAME}[{extras}]" if extras else PACKAGE_NAME + + +def _managed_env_guidance() -> str: + if sys.platform == "darwin": + hint = "`brew upgrade headroom-ai` (if installed via Homebrew), or reinstall with pipx" + elif sys.platform.startswith("win"): + hint = "reinstall with pipx (`pipx install headroom-ai`) or use a virtualenv" + else: + hint = "use your distro package manager, or reinstall with pipx / a virtualenv" + return ( + "Headroom is installed in an externally-managed system Python (PEP 668). " + f"Don't pip into it — {hint}." + ) + + +def detect_install_method(extras: str | None = None) -> InstallMethod: + """Classify the install and build the appropriate upgrade plan. + + Resolution order (first match wins), covering every supported install path + on macOS / Linux / Windows: + + 1. git checkout → refuse (`git pull`) + 2. editable install → refuse (reinstall from source) + 3. Docker → refuse (pull a new image) + 4. pipx → `pipx upgrade` + 5. uv tool → `uv tool upgrade` + 6. venv / virtualenv / conda → `sys.executable -m pip install -U` + 7. user-site (`pip --user`) → `sys.executable -m pip install -U --user` + 8. externally-managed system Python (PEP 668) → refuse with guidance + 9. writable global Python → `sys.executable -m pip install -U` (last resort) + """ + if _is_source_checkout(): + return InstallMethod( + kind="checkout", + can_self_update=False, + guidance="Running from a source checkout — update with `git pull`.", + ) + if _is_editable_install(): + return InstallMethod( + kind="editable", + can_self_update=False, + guidance=( + "Editable install detected — update your source tree, or " + "reinstall with `pip install -U --force-reinstall .`." + ), + ) + if _in_docker(): + return InstallMethod( + kind="docker", + can_self_update=False, + guidance=( + "Running inside a container — pull a newer Headroom image instead of self-updating." + ), + ) + + # pipx / uv tool own their venvs; match against executable, prefix, and the + # distribution location so detection works regardless of platform layout. + haystack = "::".join( + filter( + None, + ( + _norm(getattr(sys, "executable", "")), + _norm(getattr(sys, "prefix", "")), + _package_location(), + ), + ) + ) + + pipx_home = _norm(os.environ.get("PIPX_HOME")) + if "/pipx/venvs/" in haystack or "/pipx/" in haystack or (pipx_home and pipx_home in haystack): + return InstallMethod( + kind="pipx", + can_self_update=True, + argv=["pipx", "upgrade", PACKAGE_NAME], + ) + + uv_tool_dir = _norm(os.environ.get("UV_TOOL_DIR")) + if "/uv/tools/" in haystack or (uv_tool_dir and uv_tool_dir in haystack): + return InstallMethod( + kind="uv-tool", + can_self_update=True, + argv=["uv", "tool", "upgrade", PACKAGE_NAME], + ) + + if _in_virtualenv(): + return InstallMethod( + kind="pip", + can_self_update=True, + argv=[sys.executable, "-m", "pip", "install", "-U", _spec(extras)], + ) + + location = _package_location() + if _is_user_site_install(location): + return InstallMethod( + kind="pip-user", + can_self_update=True, + argv=[sys.executable, "-m", "pip", "install", "-U", "--user", _spec(extras)], + ) + + if _is_externally_managed(): + return InstallMethod( + kind="system", + can_self_update=False, + guidance=_managed_env_guidance(), + ) + + # Writable global interpreter (e.g. Windows python.org, some Linux setups): + # pip -U works without admin in the common case; if not, the manual command + # is surfaced on failure. + return InstallMethod( + kind="pip", + can_self_update=True, + argv=[sys.executable, "-m", "pip", "install", "-U", _spec(extras)], + ) + + +@main.command("update") +@click.option("--check", "check_only", is_flag=True, help="Report only; do not upgrade.") +@click.option("-y", "--yes", "assume_yes", is_flag=True, help="Skip the confirmation prompt.") +@click.option("--pre", "allow_pre", is_flag=True, help="Include pre-releases.") +@click.option( + "--extras", + default=None, + help="Re-request extras for the pip path, e.g. 'all' or 'proxy'.", +) +def update(check_only: bool, assume_yes: bool, allow_pre: bool, extras: str | None) -> None: + """Update Headroom to the latest release. + + Detects pipx / uv tool / pip installs and runs the right upgrade. Refuses + (with guidance) for git checkouts, editable installs, Docker, and system + Python. + """ + current = installed_version() + + click.echo("Checking PyPI for the latest Headroom release...") + latest = fetch_latest_version(allow_pre=allow_pre) + if latest is None: + raise click.ClickException("Could not reach PyPI to check for updates. Try again later.") + + # Refresh the banner cache as a side effect of an explicit check. + write_cache(latest) + + if current: + from packaging.version import InvalidVersion, Version + + try: + if Version(latest) <= Version(current): + click.echo(f"Headroom is up to date ({current}).") + return + except InvalidVersion: + pass + click.echo(f"Update available: {current} → {latest}") + else: + click.echo(f"Latest Headroom release: {latest}") + + method = detect_install_method(extras) + + if not method.can_self_update: + click.echo(method.guidance or "Automatic update is not available for this install.") + return + + assert method.argv is not None + cmd_str = _format_cmd(method.argv) + click.echo(f"Upgrade command: {cmd_str}") + + if check_only: + return + + if not assume_yes and not click.confirm("Proceed with the upgrade?", default=True): + click.echo("Aborted.") + return + + click.echo(f"Running: {cmd_str}") + try: + result = subprocess.run(method.argv) # noqa: S603 — argv built from a fixed allowlist + except FileNotFoundError: + raise click.ClickException( + f"`{method.argv[0]}` was not found on PATH. Install it or upgrade manually: {cmd_str}" + ) from None + + if result.returncode != 0: + raise click.ClickException( + f"Upgrade failed (exit {result.returncode}). Run manually: {cmd_str}" + ) + + # ASCII-only output — emoji can raise UnicodeEncodeError on some Windows consoles. + click.echo(f"Headroom upgraded to {latest}.") + click.echo("Restart any running `headroom proxy` to pick up the new version.") + + +__all__ = ["InstallMethod", "detect_install_method", "update"] diff --git a/headroom/update_check.py b/headroom/update_check.py new file mode 100644 index 000000000..b527d9167 --- /dev/null +++ b/headroom/update_check.py @@ -0,0 +1,303 @@ +"""Best-effort "is a newer Headroom released?" check. + +This module is intentionally dependency-light (stdlib + ``packaging`` only — +``httpx`` lives in the ``[proxy]`` extra and must not be required by the base +CLI). It mirrors the telemetry beacon contract: opt-out, cached, fire-and- +forget, and it must never raise into a caller or block startup. + +Two halves, deliberately split so a background thread never races stdout: + +* :func:`maybe_check_async` performs the (rate-limited) network probe on a + daemon thread and writes the result to a cache file. It prints nothing. +* :func:`format_update_notice` reads *only* the cache and returns a one-line + notice string (or ``None``). Callers own the rendering. + +Opt out with ``HEADROOM_UPDATE_CHECK=off``. Also skipped in ``--stateless`` +mode, in CI, inside Docker, and from a git checkout (developers manage their +own tree). +""" + +from __future__ import annotations + +import json +import logging +import os +import threading +import time +import urllib.request +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +PACKAGE_NAME = "headroom-ai" +_PYPI_JSON_URL = f"https://pypi.org/pypi/{PACKAGE_NAME}/json" +_CACHE_FILE = "update_check.json" + +# Probe PyPI at most once per day. +_CHECK_TTL_SECONDS = 86_400 + +_OFF_VALUES = frozenset(("off", "false", "0", "no", "disable", "disabled")) +_TRUE_VALUES = frozenset(("on", "true", "1", "yes", "enable", "enabled")) + + +def _env_off(name: str, default: str = "on") -> bool: + """Return True when env var ``name`` is set to a falsey/off value.""" + return os.environ.get(name, default).strip().lower() in _OFF_VALUES + + +def _env_on(name: str) -> bool: + """Return True when env var ``name`` is set to a truthy/on value.""" + return os.environ.get(name, "").strip().lower() in _TRUE_VALUES + + +def is_update_check_enabled() -> bool: + """Whether the update check / banner should run at all. + + Disabled by ``HEADROOM_UPDATE_CHECK=off``, stateless mode + (``HEADROOM_STATELESS=true``/``1``/``yes``/``on``, matching the proxy's own + parsing), or any CI environment (``CI`` set). + """ + if _env_off("HEADROOM_UPDATE_CHECK"): + return False + if _env_on("HEADROOM_STATELESS"): + return False + if os.environ.get("CI", "").strip(): + return False + return True + + +def _is_source_checkout() -> bool: + """True when running from a git checkout (developers manage their tree).""" + try: + from headroom._version import _source_root + + return _source_root() is not None + except Exception: + return False + + +def _in_docker() -> bool: + """Best-effort container detection — image rebuilds, not self-update.""" + try: + return Path("/.dockerenv").exists() or bool( + os.environ.get("HEADROOM_IN_DOCKER", "").strip() + ) + except Exception: + return False + + +def installed_version() -> str | None: + """Return the *installed-distribution* version, or None. + + Deliberately uses ``importlib.metadata`` rather than + ``headroom._version.get_version()`` — the latter computes a synthetic + version from git history in a checkout, which would produce a meaningless + comparison against PyPI. + """ + try: + from importlib.metadata import PackageNotFoundError, version + + try: + return version(PACKAGE_NAME) + except PackageNotFoundError: + return None + except Exception: + return None + + +def _cache_path() -> Path: + from headroom.paths import workspace_dir + + return workspace_dir() / _CACHE_FILE + + +def read_cache() -> dict[str, Any] | None: + """Return the cached check result, or None if missing/unreadable.""" + try: + path = _cache_path() + if not path.exists(): + return None + with path.open("r", encoding="utf-8") as fh: + data = json.load(fh) + return data if isinstance(data, dict) else None + except Exception: + return None + + +def write_cache(latest_version: str, *, now: float | None = None) -> None: + """Persist the latest-known version + check timestamp. Never raises.""" + try: + from headroom.paths import ensure_workspace_dir + + ensure_workspace_dir() + payload = { + "last_check": now if now is not None else time.time(), + "latest_version": latest_version, + } + path = _cache_path() + tmp = path.with_suffix(".json.tmp") + with tmp.open("w", encoding="utf-8") as fh: + json.dump(payload, fh) + tmp.replace(path) + except Exception: + logger.debug("update_check: failed to write cache", exc_info=True) + + +def _select_latest(data: dict[str, Any], *, allow_pre: bool) -> str | None: + """Pick the newest non-yanked release from a PyPI JSON payload.""" + from packaging.version import InvalidVersion, Version + + releases = data.get("releases") + candidates: list[Version] = [] + if isinstance(releases, dict): + for ver_str, files in releases.items(): + # Skip releases whose every artifact is yanked. + if ( + isinstance(files, list) + and files + and all(isinstance(f, dict) and f.get("yanked") for f in files) + ): + continue + try: + ver = Version(ver_str) + except InvalidVersion: + continue + if ver.is_prerelease and not allow_pre: + continue + candidates.append(ver) + if candidates: + return str(max(candidates)) + + # Fallback to info.version when releases is absent/empty. + info = data.get("info") + if isinstance(info, dict): + ver_str = info.get("version") + if isinstance(ver_str, str): + try: + ver = Version(ver_str) + except InvalidVersion: + return None + if ver.is_prerelease and not allow_pre: + return None + return str(ver) + return None + + +def fetch_latest_version(*, allow_pre: bool = False, timeout: float = 4.0) -> str | None: + """Query the PyPI JSON API for the latest release. Returns None on any error. + + Uses ``urllib`` (stdlib) so the base CLI install needs no HTTP dependency. + """ + try: + req = urllib.request.Request( + _PYPI_JSON_URL, + headers={"Accept": "application/json", "User-Agent": "headroom-update-check"}, + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 — fixed https URL + data = json.loads(resp.read().decode("utf-8")) + return _select_latest(data, allow_pre=allow_pre) + except Exception: + logger.debug("update_check: PyPI fetch failed", exc_info=True) + return None + + +def should_check(now: float | None = None) -> bool: + """True when the cache is stale (older than the TTL) or absent.""" + cache = read_cache() + if not cache: + return True + last = cache.get("last_check") + if not isinstance(last, (int, float)): + return True + now = now if now is not None else time.time() + return (now - last) >= _CHECK_TTL_SECONDS + + +def run_check(*, allow_pre: bool = False, now: float | None = None) -> str | None: + """Probe PyPI and update the cache. Returns the latest version or None. + + Synchronous — used directly by ``headroom update`` and indirectly by + :func:`maybe_check_async`. Honors the enable gate. + """ + if not is_update_check_enabled(): + return None + latest = fetch_latest_version(allow_pre=allow_pre) + if latest: + write_cache(latest, now=now) + return latest + + +def maybe_check_async() -> threading.Thread | None: + """Fire a rate-limited background check on a daemon thread. + + Returns the spawned thread (for tests) or None when the check is gated off, + suppressed (checkout/Docker), or still within the TTL window. Never blocks + and never raises. + """ + try: + if not is_update_check_enabled() or _is_source_checkout() or _in_docker(): + return None + if not should_check(): + return None + + def _worker() -> None: + try: + run_check() + except Exception: + logger.debug("update_check: background check crashed", exc_info=True) + + thread = threading.Thread(target=_worker, name="headroom-update-check", daemon=True) + thread.start() + return thread + except Exception: + logger.debug("update_check: maybe_check_async crashed", exc_info=True) + return None + + +def format_update_notice() -> str | None: + """Return a one-line "update available" notice, or None. + + Reads only the cache (no network). Returns None when the check is disabled, + in a checkout/Docker, when the installed version is unknown, or when already + up to date. + """ + try: + if not is_update_check_enabled() or _is_source_checkout() or _in_docker(): + return None + cache = read_cache() + if not cache: + return None + latest = cache.get("latest_version") + current = installed_version() + if not isinstance(latest, str) or not current: + return None + + from packaging.version import InvalidVersion, Version + + try: + if Version(latest) <= Version(current): + return None + except InvalidVersion: + return None + + # ASCII-only: some Windows consoles can't encode unicode and would raise + # UnicodeEncodeError at the echo site, breaking a "best-effort" banner. + return f"Update available: Headroom {latest} (you have {current}) - run: headroom update" + except Exception: + logger.debug("update_check: format_update_notice crashed", exc_info=True) + return None + + +__all__ = [ + "PACKAGE_NAME", + "fetch_latest_version", + "format_update_notice", + "installed_version", + "is_update_check_enabled", + "maybe_check_async", + "read_cache", + "run_check", + "should_check", + "write_cache", +] diff --git a/tests/test_cli_update.py b/tests/test_cli_update.py new file mode 100644 index 000000000..bad12b3b1 --- /dev/null +++ b/tests/test_cli_update.py @@ -0,0 +1,177 @@ +"""Tests for the `headroom update` command + install-method detection.""" + +from __future__ import annotations + +import sys + +import pytest +from click.testing import CliRunner + +from headroom.cli import update as up +from headroom.cli.main import main + + +@pytest.fixture(autouse=True) +def _clean_env(tmp_path, monkeypatch): + monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path)) + monkeypatch.delenv("PIPX_HOME", raising=False) + monkeypatch.delenv("UV_TOOL_DIR", raising=False) + monkeypatch.delenv("CONDA_PREFIX", raising=False) + # Default: not a checkout / editable / docker / managed install. + monkeypatch.setattr(up, "_is_source_checkout", lambda: False) + monkeypatch.setattr(up, "_is_editable_install", lambda: False) + monkeypatch.setattr(up, "_in_docker", lambda: False) + monkeypatch.setattr(up, "_is_externally_managed", lambda: False) + + +# --------------------------------------------------------------------------- # +# detect_install_method +# --------------------------------------------------------------------------- # +def test_detect_checkout(monkeypatch): + monkeypatch.setattr(up, "_is_source_checkout", lambda: True) + m = up.detect_install_method() + assert m.kind == "checkout" and m.can_self_update is False and "git pull" in m.guidance + + +def test_detect_editable(monkeypatch): + monkeypatch.setattr(up, "_is_editable_install", lambda: True) + m = up.detect_install_method() + assert m.kind == "editable" and m.can_self_update is False + + +def test_detect_docker(monkeypatch): + monkeypatch.setattr(up, "_in_docker", lambda: True) + m = up.detect_install_method() + assert m.kind == "docker" and m.can_self_update is False + + +def test_detect_pipx_by_path(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", "/home/u/.local/pipx/venvs/headroom-ai") + m = up.detect_install_method() + assert m.kind == "pipx" and m.argv == ["pipx", "upgrade", "headroom-ai"] + + +def test_detect_pipx_windows_path(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", r"C:\\Users\\u\\pipx\\venvs\\headroom-ai") + m = up.detect_install_method() + assert m.kind == "pipx" + + +def test_detect_uv_tool(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", "/home/u/.local/share/uv/tools/headroom-ai") + m = up.detect_install_method() + assert m.kind == "uv-tool" and m.argv == ["uv", "tool", "upgrade", "headroom-ai"] + + +def test_detect_venv_uses_current_interpreter(monkeypatch): + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + m = up.detect_install_method() + assert m.kind == "pip" + assert m.argv[:4] == [sys.executable, "-m", "pip", "install"] + assert "-U" in m.argv and "headroom-ai" in m.argv + + +def test_detect_venv_with_extras(monkeypatch): + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + m = up.detect_install_method(extras="all") + assert "headroom-ai[all]" in m.argv + + +def test_detect_user_site(monkeypatch): + monkeypatch.setattr(up, "_in_virtualenv", lambda: False) + monkeypatch.setattr(up, "_package_location", lambda: "/home/u/.local/site") + monkeypatch.setattr(up, "_is_user_site_install", lambda loc: True) + m = up.detect_install_method() + assert m.kind == "pip-user" and "--user" in m.argv + + +def test_detect_externally_managed_refuses(monkeypatch): + monkeypatch.setattr(up, "_in_virtualenv", lambda: False) + monkeypatch.setattr(up, "_is_user_site_install", lambda loc: False) + monkeypatch.setattr(up, "_is_externally_managed", lambda: True) + m = up.detect_install_method() + assert m.kind == "system" and m.can_self_update is False + assert "PEP 668" in m.guidance + + +# --------------------------------------------------------------------------- # +# `headroom update` command +# --------------------------------------------------------------------------- # +def test_update_already_current(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.26.0") + res = CliRunner().invoke(main, ["update"]) + assert res.exit_code == 0 + assert "up to date" in res.output + + +def test_update_check_reports_command_without_running(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + + def _no_run(*a, **k): + raise AssertionError("subprocess.run must not be called with --check") + + monkeypatch.setattr(up.subprocess, "run", _no_run) + res = CliRunner().invoke(main, ["update", "--check"]) + assert res.exit_code == 0 + assert "Update available: 0.26.0 → 0.27.0" in res.output + assert "pip" in res.output and "install" in res.output + + +def test_update_runs_upgrade_with_yes(monkeypatch): + calls = {} + + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + + class _Result: + returncode = 0 + + def _run(argv, *a, **k): + calls["argv"] = argv + return _Result() + + monkeypatch.setattr(up.subprocess, "run", _run) + res = CliRunner().invoke(main, ["update", "--yes"]) + assert res.exit_code == 0 + assert calls["argv"][:4] == [sys.executable, "-m", "pip", "install"] + assert "upgraded to 0.27.0" in res.output + + +def test_update_refuses_in_checkout(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_is_source_checkout", lambda: True) + + def _no_run(*a, **k): + raise AssertionError("must not upgrade a checkout") + + monkeypatch.setattr(up.subprocess, "run", _no_run) + res = CliRunner().invoke(main, ["update", "--yes"]) + assert res.exit_code == 0 + assert "git pull" in res.output + + +def test_update_network_failure(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: None) + res = CliRunner().invoke(main, ["update"]) + assert res.exit_code != 0 + assert "Could not reach PyPI" in res.output + + +def test_update_upgrade_failure_surfaces_command(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + + class _Result: + returncode = 1 + + monkeypatch.setattr(up.subprocess, "run", lambda *a, **k: _Result()) + res = CliRunner().invoke(main, ["update", "--yes"]) + assert res.exit_code != 0 + assert "Upgrade failed" in res.output diff --git a/tests/test_update_check.py b/tests/test_update_check.py new file mode 100644 index 000000000..7f00917ec --- /dev/null +++ b/tests/test_update_check.py @@ -0,0 +1,186 @@ +"""Tests for headroom.update_check (PyPI probe, cache, banner notice).""" + +from __future__ import annotations + +import json +import time + +import pytest + +from headroom import update_check as uc + + +@pytest.fixture(autouse=True) +def _workspace(tmp_path, monkeypatch): + """Point the workspace (cache) dir at a tmp dir and enable the check.""" + monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path)) + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", "on") + monkeypatch.delenv("HEADROOM_STATELESS", raising=False) + monkeypatch.delenv("CI", raising=False) + # Treat tests as a non-checkout, non-docker install by default. + monkeypatch.setattr(uc, "_is_source_checkout", lambda: False) + monkeypatch.setattr(uc, "_in_docker", lambda: False) + return tmp_path + + +# --------------------------------------------------------------------------- # +# enable gate +# --------------------------------------------------------------------------- # +def test_enabled_by_default(monkeypatch): + assert uc.is_update_check_enabled() is True + + +@pytest.mark.parametrize("val", ["off", "false", "0", "no", "disabled"]) +def test_disabled_by_env(monkeypatch, val): + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", val) + assert uc.is_update_check_enabled() is False + + +def test_disabled_in_stateless(monkeypatch): + monkeypatch.setenv("HEADROOM_STATELESS", "1") + assert uc.is_update_check_enabled() is False + + +def test_disabled_in_ci(monkeypatch): + monkeypatch.setenv("CI", "true") + assert uc.is_update_check_enabled() is False + + +# --------------------------------------------------------------------------- # +# _select_latest +# --------------------------------------------------------------------------- # +def test_select_latest_picks_max_stable(): + data = {"releases": {"0.25.0": [{}], "0.26.0": [{}], "0.27.0rc1": [{}]}} + assert uc._select_latest(data, allow_pre=False) == "0.26.0" + + +def test_select_latest_allows_pre(): + data = {"releases": {"0.26.0": [{}], "0.27.0rc1": [{}]}} + assert uc._select_latest(data, allow_pre=True) == "0.27.0rc1" + + +def test_select_latest_skips_fully_yanked(): + data = { + "releases": { + "0.26.0": [{"yanked": False}], + "0.27.0": [{"yanked": True}], + } + } + assert uc._select_latest(data, allow_pre=False) == "0.26.0" + + +def test_select_latest_falls_back_to_info_version(): + data = {"releases": {}, "info": {"version": "0.26.0"}} + assert uc._select_latest(data, allow_pre=False) == "0.26.0" + + +# --------------------------------------------------------------------------- # +# fetch_latest_version +# --------------------------------------------------------------------------- # +def test_fetch_latest_version_parses(monkeypatch): + payload = json.dumps({"releases": {"0.26.0": [{}], "0.27.0": [{}]}}).encode() + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return payload + + monkeypatch.setattr(uc.urllib.request, "urlopen", lambda *a, **k: _Resp()) + assert uc.fetch_latest_version() == "0.27.0" + + +def test_fetch_latest_version_network_error_returns_none(monkeypatch): + def _boom(*a, **k): + raise OSError("no network") + + monkeypatch.setattr(uc.urllib.request, "urlopen", _boom) + assert uc.fetch_latest_version() is None + + +# --------------------------------------------------------------------------- # +# cache + should_check +# --------------------------------------------------------------------------- # +def test_cache_roundtrip(): + uc.write_cache("0.27.0", now=1000.0) + cache = uc.read_cache() + assert cache["latest_version"] == "0.27.0" + assert cache["last_check"] == 1000.0 + + +def test_should_check_true_when_no_cache(): + assert uc.should_check() is True + + +def test_should_check_false_when_fresh(): + now = time.time() + uc.write_cache("0.27.0", now=now) + assert uc.should_check(now=now + 10) is False + + +def test_should_check_true_when_stale(): + now = time.time() + uc.write_cache("0.27.0", now=now) + assert uc.should_check(now=now + uc._CHECK_TTL_SECONDS + 1) is True + + +# --------------------------------------------------------------------------- # +# format_update_notice +# --------------------------------------------------------------------------- # +def test_notice_when_newer(monkeypatch): + uc.write_cache("0.27.0") + monkeypatch.setattr(uc, "installed_version", lambda: "0.26.0") + notice = uc.format_update_notice() + assert notice and "0.27.0" in notice and "headroom update" in notice + + +def test_no_notice_when_current(monkeypatch): + uc.write_cache("0.26.0") + monkeypatch.setattr(uc, "installed_version", lambda: "0.26.0") + assert uc.format_update_notice() is None + + +def test_no_notice_in_checkout(monkeypatch): + uc.write_cache("0.27.0") + monkeypatch.setattr(uc, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(uc, "_is_source_checkout", lambda: True) + assert uc.format_update_notice() is None + + +def test_no_notice_when_disabled(monkeypatch): + uc.write_cache("0.27.0") + monkeypatch.setattr(uc, "installed_version", lambda: "0.26.0") + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", "off") + assert uc.format_update_notice() is None + + +def test_no_notice_when_version_unknown(monkeypatch): + uc.write_cache("0.27.0") + monkeypatch.setattr(uc, "installed_version", lambda: None) + assert uc.format_update_notice() is None + + +# --------------------------------------------------------------------------- # +# maybe_check_async +# --------------------------------------------------------------------------- # +def test_maybe_check_async_writes_cache(monkeypatch): + monkeypatch.setattr(uc, "fetch_latest_version", lambda **k: "0.27.0") + thread = uc.maybe_check_async() + assert thread is not None + thread.join(timeout=5) + assert uc.read_cache()["latest_version"] == "0.27.0" + + +def test_maybe_check_async_skips_in_checkout(monkeypatch): + monkeypatch.setattr(uc, "_is_source_checkout", lambda: True) + assert uc.maybe_check_async() is None + + +def test_maybe_check_async_skips_when_fresh(monkeypatch): + uc.write_cache("0.27.0") + monkeypatch.setattr(uc, "fetch_latest_version", lambda **k: pytest.fail("should not fetch")) + assert uc.maybe_check_async() is None diff --git a/tests/test_update_helpers.py b/tests/test_update_helpers.py new file mode 100644 index 000000000..96f3a0818 --- /dev/null +++ b/tests/test_update_helpers.py @@ -0,0 +1,304 @@ +"""Coverage for update_check / cli.update helper functions and branches.""" + +from __future__ import annotations + +import importlib.metadata as md +import sysconfig + +import pytest +from click.testing import CliRunner + +from headroom import update_check as uc +from headroom.cli import update as up +from headroom.cli.main import main + + +@pytest.fixture(autouse=True) +def _env(tmp_path, monkeypatch): + monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path)) + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", "on") + monkeypatch.delenv("HEADROOM_STATELESS", raising=False) + monkeypatch.delenv("CI", raising=False) + monkeypatch.delenv("HEADROOM_IN_DOCKER", raising=False) + monkeypatch.delenv("CONDA_PREFIX", raising=False) + + +# --------------------------------------------------------------------------- # +# cli.update helpers +# --------------------------------------------------------------------------- # +def test_norm_normalizes_and_lowercases(): + assert up._norm("/Foo/Bar").endswith("/foo/bar") + assert up._norm(None) == "" + assert up._norm(r"C:\\X\\Y").count("\\") == 0 + + +def test_in_virtualenv_via_prefix(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", "/venv") + monkeypatch.setattr(up.sys, "base_prefix", "/usr") + assert up._in_virtualenv() is True + + +def test_in_virtualenv_via_conda(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", "/x") + monkeypatch.setattr(up.sys, "base_prefix", "/x") + monkeypatch.setenv("CONDA_PREFIX", "/opt/conda") + assert up._in_virtualenv() is True + + +def test_in_virtualenv_false(monkeypatch): + monkeypatch.setattr(up.sys, "prefix", "/x") + monkeypatch.setattr(up.sys, "base_prefix", "/x") + assert up._in_virtualenv() is False + + +def test_in_docker_env_flag(monkeypatch): + monkeypatch.setenv("HEADROOM_IN_DOCKER", "1") + assert up._in_docker() is True + + +def test_in_docker_default_false(): + # No HEADROOM_IN_DOCKER and (almost certainly) no /.dockerenv on the runner. + assert isinstance(up._in_docker(), bool) + + +def test_editable_install_true(monkeypatch): + class _D: + def read_text(self, name): + return '{"dir_info": {"editable": true}}' + + monkeypatch.setattr(md, "distribution", lambda name: _D()) + assert up._is_editable_install() is True + + +def test_editable_install_false_when_not_editable(monkeypatch): + class _D: + def read_text(self, name): + return '{"url": "https://pypi.org", "archive_info": {}}' + + monkeypatch.setattr(md, "distribution", lambda name: _D()) + assert up._is_editable_install() is False + + +def test_editable_install_false_when_no_direct_url(monkeypatch): + class _D: + def read_text(self, name): + return None + + monkeypatch.setattr(md, "distribution", lambda name: _D()) + assert up._is_editable_install() is False + + +def test_editable_install_swallows_errors(monkeypatch): + def _boom(name): + raise RuntimeError("nope") + + monkeypatch.setattr(md, "distribution", _boom) + assert up._is_editable_install() is False + + +def test_package_location_handles_missing(monkeypatch): + def _boom(name): + raise md.PackageNotFoundError(name) + + monkeypatch.setattr(md, "distribution", _boom) + assert up._package_location() is None + + +def test_user_site_and_membership(monkeypatch): + monkeypatch.setattr(up, "_user_site", lambda: "/home/u/.local/site") + assert up._is_user_site_install("/home/u/.local/site/headroom_ai") is True + assert up._is_user_site_install("/home/u/.local/site") is True + assert up._is_user_site_install("/usr/lib/python3/site") is False + assert up._is_user_site_install(None) is False + + +def test_user_site_no_sibling_prefix_match(monkeypatch): + # Path-segment containment: "/.../site" must NOT match "/.../site-packages". + monkeypatch.setattr(up, "_user_site", lambda: "/home/u/.local/site") + assert up._is_user_site_install("/home/u/.local/site-packages/x") is False + + +def test_format_cmd_quotes_spaces(monkeypatch): + monkeypatch.setattr(up.sys, "platform", "linux") + out = up._format_cmd(["/path with space/python", "-m", "pip", "install", "-U", "headroom-ai"]) + assert "'/path with space/python'" in out + + +def test_format_cmd_windows(monkeypatch): + monkeypatch.setattr(up.sys, "platform", "win32") + out = up._format_cmd([r"C:\\Program Files\\Python\\python.exe", "-m", "pip"]) + assert "Program Files" in out and out.endswith("-m pip") + + +def test_externally_managed_true(tmp_path, monkeypatch): + (tmp_path / "EXTERNALLY-MANAGED").write_text("[externally-managed]") + monkeypatch.setattr(sysconfig, "get_path", lambda key: str(tmp_path)) + assert up._is_externally_managed() is True + + +def test_externally_managed_false(tmp_path, monkeypatch): + monkeypatch.setattr(sysconfig, "get_path", lambda key: str(tmp_path)) + assert up._is_externally_managed() is False + + +def test_user_site_real_returns_str_or_empty(): + assert isinstance(up._user_site(), str) + + +def test_source_checkout_real_is_bool(): + assert isinstance(up._is_source_checkout(), bool) + + +def test_package_location_real_runs(): + # Either a normalized path string or None, but the call must not raise. + assert up._package_location() is None or isinstance(up._package_location(), str) + + +@pytest.mark.parametrize( + "platform,needle", + [("darwin", "brew"), ("win32", "pipx"), ("linux", "distro")], +) +def test_managed_env_guidance(monkeypatch, platform, needle): + monkeypatch.setattr(up.sys, "platform", platform) + assert needle in up._managed_env_guidance() + + +def test_spec_with_and_without_extras(): + assert up._spec(None) == "headroom-ai" + assert up._spec("all") == "headroom-ai[all]" + assert up._spec("[proxy]") == "headroom-ai[proxy]" + + +# --------------------------------------------------------------------------- # +# cli.update command branches +# --------------------------------------------------------------------------- # +def test_update_aborts_on_decline(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_is_source_checkout", lambda: False) + monkeypatch.setattr(up, "_is_editable_install", lambda: False) + monkeypatch.setattr(up, "_in_docker", lambda: False) + monkeypatch.setattr(up, "_in_virtualenv", lambda: True) + + def _no_run(*a, **k): + raise AssertionError("should not run after decline") + + monkeypatch.setattr(up.subprocess, "run", _no_run) + res = CliRunner().invoke(main, ["update"], input="n\n") + assert res.exit_code == 0 + assert "Aborted" in res.output + + +def test_update_missing_tool_surfaces_command(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_is_source_checkout", lambda: False) + monkeypatch.setattr(up, "_is_editable_install", lambda: False) + monkeypatch.setattr(up, "_in_docker", lambda: False) + monkeypatch.setattr( + up, + "detect_install_method", + lambda extras=None: up.InstallMethod( + kind="pipx", can_self_update=True, argv=["pipx", "upgrade", "headroom-ai"] + ), + ) + + def _missing(*a, **k): + raise FileNotFoundError("pipx") + + monkeypatch.setattr(up.subprocess, "run", _missing) + res = CliRunner().invoke(main, ["update", "--yes"]) + assert res.exit_code != 0 + assert "not found on PATH" in res.output + + +def test_update_externally_managed_refuses_via_command(monkeypatch): + monkeypatch.setattr(up, "installed_version", lambda: "0.26.0") + monkeypatch.setattr(up, "fetch_latest_version", lambda **k: "0.27.0") + monkeypatch.setattr(up, "_is_source_checkout", lambda: False) + monkeypatch.setattr(up, "_is_editable_install", lambda: False) + monkeypatch.setattr(up, "_in_docker", lambda: False) + monkeypatch.setattr(up, "_in_virtualenv", lambda: False) + monkeypatch.setattr(up, "_is_user_site_install", lambda loc: False) + monkeypatch.setattr(up, "_is_externally_managed", lambda: True) + res = CliRunner().invoke(main, ["update", "--yes"]) + assert res.exit_code == 0 + assert "PEP 668" in res.output + + +# --------------------------------------------------------------------------- # +# update_check helpers / branches +# --------------------------------------------------------------------------- # +def test_installed_version_present(monkeypatch): + monkeypatch.setattr(md, "version", lambda name: "0.26.0") + assert uc.installed_version() == "0.26.0" + + +def test_installed_version_not_found(monkeypatch): + def _boom(name): + raise md.PackageNotFoundError(name) + + monkeypatch.setattr(md, "version", _boom) + assert uc.installed_version() is None + + +def test_is_source_checkout_real_is_bool(): + assert isinstance(uc._is_source_checkout(), bool) + + +def test_in_docker_real_is_bool(): + assert isinstance(uc._in_docker(), bool) + + +def test_select_latest_skips_invalid_versions(): + data = {"releases": {"not-a-version": [{}], "0.26.0": [{}]}} + assert uc._select_latest(data, allow_pre=False) == "0.26.0" + + +def test_select_latest_info_fallback_invalid_returns_none(): + data = {"releases": {}, "info": {"version": "not-a-version"}} + assert uc._select_latest(data, allow_pre=False) is None + + +def test_select_latest_info_fallback_prerelease_filtered(): + data = {"releases": {}, "info": {"version": "1.0.0rc1"}} + assert uc._select_latest(data, allow_pre=False) is None + assert uc._select_latest(data, allow_pre=True) == "1.0.0rc1" + + +def test_run_check_disabled_returns_none(monkeypatch): + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", "off") + monkeypatch.setattr(uc, "fetch_latest_version", lambda **k: pytest.fail("no fetch")) + assert uc.run_check() is None + + +def test_maybe_check_async_disabled_returns_none(monkeypatch): + monkeypatch.setenv("HEADROOM_UPDATE_CHECK", "off") + assert uc.maybe_check_async() is None + + +def test_format_update_notice_invalid_versions(monkeypatch): + uc.write_cache("not-a-version") + monkeypatch.setattr(uc, "_is_source_checkout", lambda: False) + monkeypatch.setattr(uc, "_in_docker", lambda: False) + monkeypatch.setattr(uc, "installed_version", lambda: "0.26.0") + assert uc.format_update_notice() is None + + +def test_fetch_latest_version_info_only(monkeypatch): + import json + + payload = json.dumps({"releases": {}, "info": {"version": "0.30.0"}}).encode() + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return payload + + monkeypatch.setattr(uc.urllib.request, "urlopen", lambda *a, **k: _Resp()) + assert uc.fetch_latest_version() == "0.30.0"