mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
feat(cli): add headroom update command and release banner (#1088)
## Description Adds a `headroom update` self-update command and a passive "update available" banner, so users no longer need to remember the right `pip`/`pipx`/`uv` incantation for their environment, and long-running proxies get nudged when they drift behind a release. Closes #1087 ## Type of Change - [x] New feature (non-breaking change that adds functionality) - [x] Documentation update ## Changes Made - `headroom/cli/update.py` — `headroom update` command. `detect_install_method()` resolves the install (git checkout, editable, Docker, pipx, uv tool, venv/conda, `pip --user`, externally-managed system Python per PEP 668, writable global) and builds the matching upgrade. pip path always uses `sys.executable -m pip` so it can't touch the wrong interpreter. Refuses with guidance where self-update is unsafe. Flags: `--check`, `--yes`, `--pre`, `--extras`. - `headroom/update_check.py` — best-effort PyPI check (stdlib `urllib`, no new dep). Split into a daemon-thread probe that caches to `~/.headroom/update_check.json` (≤ once/day) and a cache-only `format_update_notice()`. Opt-out `HEADROOM_UPDATE_CHECK=off`; skipped in `--stateless`, CI, Docker, checkouts. - `headroom/cli/main.py`, `headroom/cli/__init__.py` — register `update`; fire the background check from the group callback (skipped for `update`). - `headroom/cli/proxy.py` — render the one-line notice after the startup banner (best-effort, never blocks). - `README.md` — "Updating" section + opt-out env var. - Tests: `tests/test_update_check.py`, `tests/test_cli_update.py`. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ pytest tests/test_update_check.py tests/test_cli_update.py -q 42 passed in 1.63s $ ruff check headroom/cli/update.py headroom/update_check.py headroom/cli/main.py All checks passed! $ mypy headroom/update_check.py headroom/cli/update.py Success: no issues found in 2 source files ``` ## Real Behavior Proof - Environment: macOS, Python 3.11, source checkout - Exact command / steps: `python -m headroom.cli update --help`; `detect_install_method()` in the checkout - Observed result: command + flags render; in a checkout `detect_install_method()` returns `kind=checkout, can_self_update=False` ("update with `git pull`") and `format_update_notice()` returns `None` (dev tree not nagged) - Not tested: live PyPI fetch and a real pipx/uv-tool upgrade on this machine (covered by unit tests with mocked `urllib`/`subprocess`) ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [x] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - CHANGELOG.md is release-please-managed, so it is intentionally not hand-edited (N/A above). - Update check uses stdlib `urllib` because `httpx` lives only in the `[proxy]` extra — the base CLI must stay dependency-light.
This commit is contained in:
Focused Instability
GitHub
parent
a554c3a0e6
commit
26be2c39cb
9 changed files with 1353 additions and 0 deletions
17
README.md
17
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`
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ from . import ( # noqa: F401
|
|||
perf,
|
||||
proxy,
|
||||
tools,
|
||||
update,
|
||||
wrap,
|
||||
)
|
||||
from .main import main
|
||||
|
|
|
|||
|
|
@ -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
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# -----------------------------------------------------------------------
|
||||
|
|
|
|||
341
headroom/cli/update.py
Normal file
341
headroom/cli/update.py
Normal file
|
|
@ -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"]
|
||||
303
headroom/update_check.py
Normal file
303
headroom/update_check.py
Normal file
|
|
@ -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",
|
||||
]
|
||||
177
tests/test_cli_update.py
Normal file
177
tests/test_cli_update.py
Normal file
|
|
@ -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
|
||||
186
tests/test_update_check.py
Normal file
186
tests/test_update_check.py
Normal file
|
|
@ -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
|
||||
304
tests/test_update_helpers.py
Normal file
304
tests/test_update_helpers.py
Normal file
|
|
@ -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"
|
||||
Loading…
Add table
Add a link
Reference in a new issue