feat(cli): add agnes self-upgrade with smoke test + rollback
Reuses cli.update_check.check() for the version probe — extended with bypass_disabled=True so explicit user-typed self-upgrade is not silenced by AGNES_NO_UPDATE_CHECK (which is for the implicit warning loop). Install path: uv tool install --force when uv is on PATH; otherwise curl + pip via sys.executable (NOT system python3, NOT --user — both would land outside the agnes venv and silently no-op the upgrade). Smoke test execs the binary at the install-resolved path (uv tool dir joined with agnes-the-ai-analyst/bin/agnes, or sys.executable's sibling agnes for pip) — never via shutil.which, which can resolve a stale shadow on PATH and produce a false-positive smoke pass on the OLD version. Smoke also asserts --version output contains info.latest via PEP 440 Version() equality (so 0.40.0 does not falsely match 0.40.10). On smoke fail: rollback to last_known_good.json (written only after a previous run's smoke passed). Rollback rc is captured and surfaced on stderr if it also fails. First-ever upgrade or unrecoverable rollback prints the canonical bootstrap recovery: curl -fsSL <server>/cli/install.sh | bash. AGNES_SELF_UPGRADE_IN_PROGRESS=1 is set for the duration of the run and propagated to the smoke-test subprocess. Layer B's _check_version_headers honors the sentinel and skips the < min hard-stop, so an in-flight upgrade can never sys.exit(2) itself. --force invalidates the update_check cache BEFORE probing. --force + offline = exit 1 with explicit stderr (without --force, offline is silent). --quiet suppresses progress output but never gags failure stderr.
This commit is contained in:
ZdenekSrotyr
parent
d93eda7de3
commit
630e224578
5 changed files with 627 additions and 2 deletions
288
cli/commands/self_upgrade.py
Normal file
288
cli/commands/self_upgrade.py
Normal file
|
|
@ -0,0 +1,288 @@
|
|||
"""`agnes self-upgrade` — pull the wheel from the server, reinstall, smoke-test,
|
||||
roll back on failure."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Optional, Union
|
||||
|
||||
import typer
|
||||
|
||||
from cli.config import _config_dir, get_server_url
|
||||
from cli.update_check import UpdateInfo, check, format_outdated_notice
|
||||
|
||||
self_upgrade_app = typer.Typer(
|
||||
name="self-upgrade",
|
||||
help="Reinstall the CLI from the server's currently-shipped wheel.",
|
||||
invoke_without_command=True,
|
||||
)
|
||||
|
||||
_SENTINEL_ENV = "AGNES_SELF_UPGRADE_IN_PROGRESS"
|
||||
|
||||
|
||||
class _Unreachable:
|
||||
"""Sentinel returned by _resolve_info when --force was specified but the
|
||||
server probe failed. Distinguishes 'explicitly requested an upgrade and
|
||||
we couldn't reach the server' (exit 1, stderr) from 'no upgrade needed'
|
||||
(exit 0, silent)."""
|
||||
|
||||
|
||||
_UNREACHABLE = _Unreachable()
|
||||
|
||||
|
||||
def _invalidate_update_cache() -> None:
|
||||
"""Drop update_check.json so the next CLI invocation re-probes /cli/latest."""
|
||||
(_config_dir() / "update_check.json").unlink(missing_ok=True)
|
||||
|
||||
|
||||
def _last_known_good_path() -> Path:
|
||||
return _config_dir() / "last_known_good.json"
|
||||
|
||||
|
||||
def _read_last_known_good() -> Optional[str]:
|
||||
p = _last_known_good_path()
|
||||
if not p.exists():
|
||||
return None
|
||||
try:
|
||||
return json.loads(p.read_text(encoding="utf-8")).get("download_url")
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def _record_last_known_good(download_url: str) -> None:
|
||||
p = _last_known_good_path()
|
||||
try:
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(json.dumps({"download_url": download_url}), encoding="utf-8")
|
||||
except OSError:
|
||||
pass # best-effort — failure to record must not break the flow
|
||||
|
||||
|
||||
def _uv_tool_bin_path() -> Optional[Path]:
|
||||
"""Locate the agnes shim uv installed.
|
||||
|
||||
Tries `uv tool dir --bin` (uv >= 0.5). Falls back to uv's documented
|
||||
default install location on older uv where `--bin` is rejected.
|
||||
"""
|
||||
bin_dir: Optional[Path] = None
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["uv", "tool", "dir", "--bin"], capture_output=True, text=True, timeout=5,
|
||||
)
|
||||
if out.returncode == 0:
|
||||
bin_dir = Path(out.stdout.strip())
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
bin_dir = None
|
||||
|
||||
if bin_dir is None:
|
||||
if sys.platform == "win32":
|
||||
appdata = os.environ.get("APPDATA")
|
||||
if appdata:
|
||||
bin_dir = Path(appdata) / "uv" / "tools" / "bin"
|
||||
else:
|
||||
bin_dir = Path.home() / ".local" / "bin"
|
||||
|
||||
if bin_dir is None or not bin_dir.exists():
|
||||
return None
|
||||
|
||||
for name in ("agnes.exe", "agnes"):
|
||||
candidate = bin_dir / name
|
||||
if candidate.exists():
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _pip_bin_path() -> Optional[Path]:
|
||||
"""`<venv>/bin/agnes` (POSIX) or `<venv>\\Scripts\\agnes.exe` (Windows)."""
|
||||
parent = Path(sys.executable).parent
|
||||
name = "agnes.exe" if sys.platform == "win32" else "agnes"
|
||||
candidate = parent / name
|
||||
return candidate if candidate.exists() else None
|
||||
|
||||
|
||||
def _install_with_uv(download_url: str, *, quiet: bool) -> int:
|
||||
out = subprocess.DEVNULL if quiet else None
|
||||
return subprocess.run(
|
||||
["uv", "tool", "install", "--force", download_url], stdout=out
|
||||
).returncode
|
||||
|
||||
|
||||
def _install_with_pip(download_url: str, *, quiet: bool) -> int:
|
||||
"""Install into the SAME interpreter that's running this command.
|
||||
|
||||
sys.executable resolves to the venv that owns the live `agnes` binary.
|
||||
`python3` would PATH-resolve to system python on macOS, landing the
|
||||
wheel outside the agnes venv. `--user` is wrong inside a uv-tool venv
|
||||
(targets ~/.local outside the venv).
|
||||
"""
|
||||
out = subprocess.DEVNULL if quiet else None
|
||||
with tempfile.TemporaryDirectory(prefix="agnes_cli.") as td:
|
||||
wheel_path = Path(td) / "agnes.whl"
|
||||
rc = subprocess.run(
|
||||
["curl", "-fsSL", "-o", str(wheel_path), download_url], stdout=out
|
||||
).returncode
|
||||
if rc != 0:
|
||||
return rc
|
||||
return subprocess.run(
|
||||
[sys.executable, "-m", "pip", "install",
|
||||
"--force-reinstall", "--no-deps", str(wheel_path)],
|
||||
stdout=out,
|
||||
).returncode
|
||||
|
||||
|
||||
def _smoke_test_new_binary(install_method: str, expected_version: str) -> tuple[bool, str]:
|
||||
"""Exec `<install-path>/agnes --version` and confirm it boots AND reports
|
||||
the expected version. Resolves the binary at the install-method-specific
|
||||
path rather than via PATH — defends against a stale shadow ahead of the
|
||||
freshly-installed binary in $PATH."""
|
||||
binary = _uv_tool_bin_path() if install_method == "uv" else _pip_bin_path()
|
||||
if binary is None:
|
||||
return False, f"agnes binary not found at expected {install_method} install path"
|
||||
try:
|
||||
env = {**os.environ, "AGNES_NO_UPDATE_CHECK": "1", _SENTINEL_ENV: "1"}
|
||||
out = subprocess.run(
|
||||
[str(binary), "--version"],
|
||||
capture_output=True, text=True, timeout=10, env=env,
|
||||
)
|
||||
if out.returncode != 0:
|
||||
return False, f"exit {out.returncode}: {out.stderr.strip()[:200]}"
|
||||
# Use Version() equality (PEP 440-aware) so "0.40.0" doesn't match "0.40.10".
|
||||
from packaging.version import InvalidVersion, Version
|
||||
tokens = out.stdout.strip().split()
|
||||
actual_str = tokens[-1] if tokens else ""
|
||||
try:
|
||||
if Version(actual_str) != Version(expected_version):
|
||||
return False, (
|
||||
f"version mismatch: expected {expected_version}, "
|
||||
f"got {actual_str}"
|
||||
)
|
||||
except InvalidVersion:
|
||||
return False, f"unparseable version output: {out.stdout.strip()[:80]}"
|
||||
return True, out.stdout.strip()
|
||||
except (subprocess.TimeoutExpired, OSError) as e:
|
||||
return False, f"{type(e).__name__}: {e}"
|
||||
|
||||
|
||||
def _resolve_info(force: bool) -> Union[UpdateInfo, _Unreachable, None]:
|
||||
"""Returns:
|
||||
UpdateInfo — install this wheel
|
||||
_UNREACHABLE — --force specified, server probe failed
|
||||
None — nothing to do (current, or offline without --force)
|
||||
"""
|
||||
if force:
|
||||
_invalidate_update_cache()
|
||||
info = check(get_server_url(), bypass_disabled=True)
|
||||
if info is None:
|
||||
return _UNREACHABLE if force else None
|
||||
if not info.download_url:
|
||||
return None
|
||||
if not force and not info.is_outdated():
|
||||
return None
|
||||
return info
|
||||
|
||||
|
||||
def _do_install_with_smoke_and_rollback(
|
||||
info: UpdateInfo, *, quiet: bool
|
||||
) -> int:
|
||||
"""Returns the exit code typer should use (0 success, 1 failure)."""
|
||||
prior_url = _read_last_known_good() # may be None on first upgrade
|
||||
|
||||
if shutil.which("uv"):
|
||||
rc = _install_with_uv(info.download_url, quiet=quiet)
|
||||
method = "uv"
|
||||
else:
|
||||
rc = _install_with_pip(info.download_url, quiet=quiet)
|
||||
method = "pip"
|
||||
|
||||
if rc != 0:
|
||||
sys.stderr.write(f"agnes self-upgrade: install failed with exit {rc}\n")
|
||||
return 1
|
||||
|
||||
ok, detail = _smoke_test_new_binary(method, expected_version=info.latest)
|
||||
if not ok:
|
||||
sys.stderr.write(
|
||||
f"agnes self-upgrade: new binary failed smoke test ({detail}).\n"
|
||||
)
|
||||
server = get_server_url().rstrip("/")
|
||||
bootstrap_recovery = f" Manual recovery: curl -fsSL {server}/cli/install.sh | bash\n"
|
||||
if prior_url and prior_url != info.download_url:
|
||||
sys.stderr.write(f" rolling back to {prior_url}\n")
|
||||
rb_rc = (
|
||||
_install_with_uv(prior_url, quiet=True)
|
||||
if method == "uv"
|
||||
else _install_with_pip(prior_url, quiet=True)
|
||||
)
|
||||
if rb_rc != 0:
|
||||
sys.stderr.write(
|
||||
f" rollback ALSO failed (rc={rb_rc}); CLI is in a broken state.\n"
|
||||
)
|
||||
sys.stderr.write(bootstrap_recovery)
|
||||
else:
|
||||
sys.stderr.write(
|
||||
" no prior wheel URL on record; rollback skipped.\n"
|
||||
)
|
||||
sys.stderr.write(bootstrap_recovery)
|
||||
return 1
|
||||
|
||||
# Convention: record then invalidate. No correctness consequence either way.
|
||||
_record_last_known_good(info.download_url)
|
||||
_invalidate_update_cache()
|
||||
if not quiet:
|
||||
typer.echo(f"agnes self-upgrade: installed {info.latest}", err=True)
|
||||
return 0
|
||||
|
||||
|
||||
@self_upgrade_app.callback()
|
||||
def self_upgrade(
|
||||
quiet: bool = typer.Option(
|
||||
False, "--quiet",
|
||||
help="Suppress progress output. Failures still surface on stderr.",
|
||||
),
|
||||
check_only: bool = typer.Option(
|
||||
False, "--check-only",
|
||||
help="Print status, don't install. Exit 1 if outdated.",
|
||||
),
|
||||
force: bool = typer.Option(
|
||||
False, "--force",
|
||||
help="Reinstall the server's current wheel even when already on the latest version.",
|
||||
),
|
||||
) -> None:
|
||||
# Snapshot any prior sentinel so we restore (rather than destroy) it
|
||||
# in finally — we own the namespace but a wrapper could legitimately
|
||||
# set it.
|
||||
prior_sentinel = os.environ.get(_SENTINEL_ENV)
|
||||
os.environ[_SENTINEL_ENV] = "1"
|
||||
try:
|
||||
info = _resolve_info(force)
|
||||
|
||||
# --check-only is read-only intent — never exit non-zero on
|
||||
# transport errors. If unreachable, treat as "can't tell, current"
|
||||
# and exit 0 silently.
|
||||
if check_only:
|
||||
if isinstance(info, _Unreachable) or info is None or not info.is_outdated():
|
||||
raise typer.Exit(0)
|
||||
typer.echo(format_outdated_notice(info), err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
if isinstance(info, _Unreachable):
|
||||
sys.stderr.write(
|
||||
f"agnes self-upgrade: cannot reach {get_server_url()}/cli/latest\n"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
if info is None:
|
||||
raise typer.Exit(0) # nothing to do, silent
|
||||
|
||||
rc = _do_install_with_smoke_and_rollback(info, quiet=quiet)
|
||||
raise typer.Exit(rc)
|
||||
finally:
|
||||
if prior_sentinel is None:
|
||||
os.environ.pop(_SENTINEL_ENV, None)
|
||||
else:
|
||||
os.environ[_SENTINEL_ENV] = prior_sentinel
|
||||
|
|
@ -33,6 +33,7 @@ from cli.commands.status import status_app
|
|||
from cli.commands.admin import admin_app
|
||||
from cli.commands.diagnose import diagnose_app
|
||||
from cli.commands.skills import skills_app
|
||||
from cli.commands.self_upgrade import self_upgrade_app
|
||||
from cli.commands.setup import setup_app
|
||||
from cli.commands.server import server_app
|
||||
from cli.commands.explore import explore_app
|
||||
|
|
@ -115,6 +116,7 @@ app.add_typer(status_app, name="status")
|
|||
app.add_typer(admin_app, name="admin")
|
||||
app.add_typer(diagnose_app, name="diagnose")
|
||||
app.add_typer(skills_app, name="skills")
|
||||
app.add_typer(self_upgrade_app, name="self-upgrade")
|
||||
app.add_typer(setup_app, name="setup")
|
||||
app.add_typer(server_app, name="server")
|
||||
app.add_typer(explore_app, name="explore")
|
||||
|
|
|
|||
|
|
@ -114,13 +114,23 @@ def _fetch_latest(server_url: str) -> Optional[dict]:
|
|||
return None
|
||||
|
||||
|
||||
def check(server_url: Optional[str]) -> Optional[UpdateInfo]:
|
||||
def check(
|
||||
server_url: Optional[str], *, bypass_disabled: bool = False
|
||||
) -> Optional[UpdateInfo]:
|
||||
"""Return UpdateInfo if a check ran (cached or fresh), else None.
|
||||
|
||||
Silent on every failure path: no server configured, CLI package not
|
||||
installed, network down, malformed response, cache unreadable.
|
||||
|
||||
`bypass_disabled=True` ignores `AGNES_NO_UPDATE_CHECK`. The env var
|
||||
silences the implicit warning loop in the root callback; an explicit
|
||||
user-typed `agnes self-upgrade` is not the implicit loop and must
|
||||
still probe. Default keeps existing call sites (root callback) silent
|
||||
when the env var is set.
|
||||
"""
|
||||
if is_disabled() or not server_url:
|
||||
if not bypass_disabled and is_disabled():
|
||||
return None
|
||||
if not server_url:
|
||||
return None
|
||||
|
||||
installed = _installed_version()
|
||||
|
|
|
|||
|
|
@ -35,6 +35,27 @@ def test_check_returns_none_when_server_url_missing(tmp_config):
|
|||
assert update_check.check(None) is None # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_check_bypass_disabled_overrides_env(monkeypatch, tmp_config):
|
||||
"""`AGNES_NO_UPDATE_CHECK=1` silences the implicit warning loop, but
|
||||
explicit callers (e.g. `agnes self-upgrade`) pass `bypass_disabled=True`
|
||||
and must NOT become a silent no-op."""
|
||||
from cli import update_check
|
||||
|
||||
monkeypatch.setenv("AGNES_NO_UPDATE_CHECK", "1")
|
||||
payload = {
|
||||
"version": "9.9.9",
|
||||
"wheel_filename": "x.whl",
|
||||
"download_url_path": "/cli/wheel/x.whl",
|
||||
}
|
||||
with patch("cli.update_check._installed_version", return_value="2.0.0"):
|
||||
with patch("cli.update_check._fetch_latest", return_value=payload):
|
||||
# Default: env var wins, returns None.
|
||||
assert update_check.check("http://server.test") is None
|
||||
# Bypass: env var ignored.
|
||||
info = update_check.check("http://server.test", bypass_disabled=True)
|
||||
assert info is not None and info.latest == "9.9.9"
|
||||
|
||||
|
||||
def test_check_returns_none_when_installed_version_unknown(tmp_config):
|
||||
from cli import update_check
|
||||
with patch("cli.update_check._installed_version", return_value="unknown"):
|
||||
|
|
|
|||
304
tests/test_self_upgrade.py
Normal file
304
tests/test_self_upgrade.py
Normal file
|
|
@ -0,0 +1,304 @@
|
|||
"""Tests for `agnes self-upgrade` — install path, smoke test, rollback
|
||||
(with rc capture), recursion barrier, --force offline failure, AGNES_NO_UPDATE_CHECK
|
||||
bypass for explicit upgrades, --quiet stderr behavior, version-mismatch
|
||||
smoke detection."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from cli.main import app
|
||||
from cli.update_check import UpdateInfo
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _ensure_no_sentinel_leak(monkeypatch):
|
||||
"""Pytest test order is not guaranteed; explicitly clear the recursion
|
||||
sentinel before every test so a leaked value from a prior test doesn't
|
||||
produce a false-positive 'cleared on exit' assertion."""
|
||||
monkeypatch.delenv("AGNES_SELF_UPGRADE_IN_PROGRESS", raising=False)
|
||||
yield
|
||||
|
||||
|
||||
_OUTDATED_URL = "http://server.test/cli/wheel/agnes-0.40.0-py3-none-any.whl"
|
||||
_PRIOR_URL = "http://server.test/cli/wheel/agnes-0.35.0-py3-none-any.whl"
|
||||
|
||||
|
||||
def _outdated_info():
|
||||
return UpdateInfo(installed="0.30.0", latest="0.40.0", download_url=_OUTDATED_URL)
|
||||
|
||||
|
||||
def _current_info():
|
||||
return UpdateInfo(installed="0.40.0", latest="0.40.0", download_url=None)
|
||||
|
||||
|
||||
def _smoke_pass():
|
||||
return (True, "agnes 0.40.0")
|
||||
|
||||
|
||||
def _smoke_fail():
|
||||
return (False, "exit 1: ImportError: cannot import name 'foo'")
|
||||
|
||||
|
||||
def test_check_only_when_outdated_exits_1():
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()):
|
||||
result = runner.invoke(app, ["self-upgrade", "--check-only"])
|
||||
assert result.exit_code == 1
|
||||
assert "out of date" in result.output
|
||||
|
||||
|
||||
def test_check_only_when_current_exits_0():
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_current_info()):
|
||||
result = runner.invoke(app, ["self-upgrade", "--check-only"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_when_current_short_circuits_no_install():
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_current_info()), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run:
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
mock_run.assert_not_called()
|
||||
|
||||
|
||||
def test_uv_path_when_uv_available():
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good"), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache"):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
args = mock_run.call_args_list[0].args[0]
|
||||
assert args[:3] == ["uv", "tool", "install"]
|
||||
assert "--force" in args
|
||||
assert _OUTDATED_URL in args
|
||||
|
||||
|
||||
def test_pip_fallback_uses_sys_executable_not_user():
|
||||
"""pip path must target the running interpreter's venv, never --user."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value=None), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good"), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache"):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list]
|
||||
assert any(cmd[0] == "curl" for cmd in cmds), cmds
|
||||
pip_cmd = next(cmd for cmd in cmds if "pip" in cmd)
|
||||
assert pip_cmd[0] == sys.executable, pip_cmd
|
||||
assert "--force-reinstall" in pip_cmd
|
||||
assert "--user" not in pip_cmd
|
||||
|
||||
|
||||
def test_force_invalidates_cache_before_check():
|
||||
"""--force must drop the cached download_url before probing /cli/latest."""
|
||||
fresh_current_with_url = UpdateInfo(installed="0.40.0", latest="0.40.0",
|
||||
download_url=_OUTDATED_URL)
|
||||
with patch("cli.commands.self_upgrade._invalidate_update_cache") as mock_invalidate, \
|
||||
patch("cli.commands.self_upgrade.check", return_value=fresh_current_with_url) as mock_check, \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good"):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade", "--force"])
|
||||
assert result.exit_code == 0
|
||||
assert mock_invalidate.call_count == 2
|
||||
mock_check.assert_called_once()
|
||||
|
||||
|
||||
def test_force_offline_exits_1_with_stderr():
|
||||
"""--force + server unreachable: exit 1 with explicit stderr."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=None), \
|
||||
patch("cli.commands.self_upgrade.get_server_url",
|
||||
return_value="http://server.test"), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache"):
|
||||
result = runner.invoke(app, ["self-upgrade", "--force"])
|
||||
assert result.exit_code == 1
|
||||
assert "cannot reach" in result.stderr
|
||||
assert "server.test" in result.stderr
|
||||
|
||||
|
||||
def test_offline_without_force_is_silent():
|
||||
"""No --force, server unreachable: exit 0 silently from self-upgrade
|
||||
itself. (The root callback's warning loop in cli/main.py may still emit
|
||||
`[update] …` to stderr — that's a separate code path; this test only
|
||||
pins that self-upgrade does not add a `cannot reach …` error.)"""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache"):
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
assert "cannot reach" not in result.stderr
|
||||
assert "self-upgrade:" not in result.stderr
|
||||
|
||||
|
||||
def test_self_upgrade_passes_bypass_disabled_to_check():
|
||||
"""AGNES_NO_UPDATE_CHECK silences the implicit warning loop, but
|
||||
explicit `agnes self-upgrade` must NOT be a silent no-op when set."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_current_info()) as mock_check:
|
||||
result = runner.invoke(app, ["self-upgrade", "--check-only"])
|
||||
assert result.exit_code == 0
|
||||
kwargs = mock_check.call_args.kwargs
|
||||
assert kwargs.get("bypass_disabled") is True
|
||||
|
||||
|
||||
def test_quiet_does_not_suppress_install_failure_stderr():
|
||||
"""--quiet suppresses progress but install/smoke failures always surface."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None):
|
||||
mock_run.return_value = MagicMock(returncode=42)
|
||||
result = runner.invoke(app, ["self-upgrade", "--quiet"])
|
||||
assert result.exit_code == 1
|
||||
assert "install failed" in result.stderr
|
||||
|
||||
|
||||
def test_smoke_fail_triggers_rollback_when_prior_url_known():
|
||||
"""Broken new wheel: smoke fails, rollback to last-known-good URL, exit 1."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=_PRIOR_URL), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good") as mock_record:
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 1
|
||||
urls_installed = [
|
||||
arg for c in mock_run.call_args_list
|
||||
for arg in c.args[0] if isinstance(arg, str) and arg.startswith("http")
|
||||
]
|
||||
assert _OUTDATED_URL in urls_installed
|
||||
assert _PRIOR_URL in urls_installed
|
||||
mock_record.assert_not_called()
|
||||
assert "smoke test" in result.stderr
|
||||
|
||||
|
||||
def test_smoke_fail_with_rollback_failure_surfaces_rc():
|
||||
"""Forward install ok, smoke fail, rollback ALSO fails: stderr surfaces rc + recovery."""
|
||||
install_results = [MagicMock(returncode=0), MagicMock(returncode=99)]
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run", side_effect=install_results), \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=_PRIOR_URL), \
|
||||
patch("cli.commands.self_upgrade.get_server_url",
|
||||
return_value="http://server.test"):
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 1
|
||||
assert "rollback ALSO failed" in result.stderr
|
||||
assert "rc=99" in result.stderr
|
||||
assert "/cli/install.sh" in result.stderr
|
||||
|
||||
|
||||
def test_smoke_fail_no_prior_url_prints_install_sh_recovery():
|
||||
"""First-ever upgrade with no rollback target: stderr points at bootstrap path."""
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_fail()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade.get_server_url",
|
||||
return_value="http://server.test"):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 1
|
||||
assert "/cli/install.sh" in result.stderr
|
||||
assert "server.test" in result.stderr
|
||||
|
||||
|
||||
def test_smoke_pass_records_last_known_good_then_invalidates_cache():
|
||||
"""Convention: record before invalidate."""
|
||||
call_order = []
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run") as mock_run, \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", return_value=_smoke_pass()), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good",
|
||||
side_effect=lambda url: call_order.append(("record", url))), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache",
|
||||
side_effect=lambda: call_order.append(("invalidate", None))):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
record_idx = next(i for i, c in enumerate(call_order) if c[0] == "record")
|
||||
invalidate_idx = next(i for i, c in enumerate(call_order) if c[0] == "invalidate")
|
||||
assert record_idx < invalidate_idx, call_order
|
||||
assert call_order[record_idx] == ("record", _OUTDATED_URL)
|
||||
|
||||
|
||||
def test_self_upgrade_propagates_sentinel_to_smoke_subprocess():
|
||||
"""The sentinel is set in os.environ during the run and cleared in finally."""
|
||||
captured_envs = []
|
||||
|
||||
def _fake_smoke(method, expected_version):
|
||||
env = {**os.environ, "AGNES_NO_UPDATE_CHECK": "1",
|
||||
"AGNES_SELF_UPGRADE_IN_PROGRESS": "1"}
|
||||
captured_envs.append(env)
|
||||
return _smoke_pass()
|
||||
|
||||
with patch("cli.commands.self_upgrade.check", return_value=_outdated_info()), \
|
||||
patch("cli.commands.self_upgrade.shutil.which", return_value="/usr/local/bin/uv"), \
|
||||
patch("cli.commands.self_upgrade.subprocess.run",
|
||||
return_value=MagicMock(returncode=0)), \
|
||||
patch("cli.commands.self_upgrade._smoke_test_new_binary", side_effect=_fake_smoke), \
|
||||
patch("cli.commands.self_upgrade._read_last_known_good", return_value=None), \
|
||||
patch("cli.commands.self_upgrade._record_last_known_good"), \
|
||||
patch("cli.commands.self_upgrade._invalidate_update_cache"):
|
||||
result = runner.invoke(app, ["self-upgrade"])
|
||||
assert result.exit_code == 0
|
||||
assert captured_envs and captured_envs[0]["AGNES_SELF_UPGRADE_IN_PROGRESS"] == "1"
|
||||
assert os.environ.get("AGNES_SELF_UPGRADE_IN_PROGRESS") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("install_method,patch_target", [
|
||||
("uv", "_uv_tool_bin_path"),
|
||||
("pip", "_pip_bin_path"),
|
||||
])
|
||||
def test_smoke_test_detects_version_mismatch(install_method, patch_target):
|
||||
"""Smoke test execs binary at install path (NOT shutil.which) and checks
|
||||
Version equality (NOT substring). Parametrized over uv + pip."""
|
||||
from pathlib import Path
|
||||
from cli.commands import self_upgrade as su
|
||||
|
||||
fake_bin = f"/fake/{install_method}/bin/agnes"
|
||||
with patch.object(su, patch_target, return_value=Path(fake_bin)), \
|
||||
patch.object(su.subprocess, "run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.30.0\n", stderr="")
|
||||
ok, detail = su._smoke_test_new_binary(install_method, expected_version="0.40.0")
|
||||
assert ok is False
|
||||
assert "version mismatch" in detail
|
||||
assert "0.40.0" in detail and "0.30.0" in detail
|
||||
assert mock_run.call_args.args[0][0] == fake_bin
|
||||
|
||||
|
||||
def test_smoke_test_passes_with_pep440_local_version():
|
||||
"""Use Version() comparison, not substring (so "0.40.0" doesn't match "0.40.10")."""
|
||||
from pathlib import Path
|
||||
from cli.commands import self_upgrade as su
|
||||
|
||||
with patch.object(su, "_uv_tool_bin_path", return_value=Path("/fake/agnes")), \
|
||||
patch.object(su.subprocess, "run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.40.0\n", stderr="")
|
||||
ok, _ = su._smoke_test_new_binary("uv", expected_version="0.40.0")
|
||||
assert ok is True
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="agnes 0.40.10\n", stderr="")
|
||||
ok, detail = su._smoke_test_new_binary("uv", expected_version="0.40.0")
|
||||
assert ok is False
|
||||
assert "version mismatch" in detail
|
||||
Loading…
Reference in a new issue