mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aa7a097037 | ||
|
|
df80fad960 | ||
|
|
a0bddbc13b | ||
|
|
b5e82edb83 |
@@ -5,6 +5,7 @@ Strix Agent Interface
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from datetime import UTC, datetime
|
||||
@@ -32,6 +33,13 @@ from strix.config.models import (
|
||||
from strix.core.paths import run_dir_for, runtime_state_dir
|
||||
from strix.interface.cli import run_cli
|
||||
from strix.interface.tui import run_tui
|
||||
from strix.interface.update_check import (
|
||||
is_binary_install,
|
||||
notify_update,
|
||||
prompt_update_if_available,
|
||||
self_update,
|
||||
start_background_check,
|
||||
)
|
||||
from strix.interface.utils import (
|
||||
assign_workspace_subdirs,
|
||||
build_final_stats_text,
|
||||
@@ -447,6 +455,14 @@ Examples:
|
||||
version=f"strix {get_version()}",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--update",
|
||||
action="store_true",
|
||||
help="Update strix to the latest version and exit. Self-updates the "
|
||||
"standalone binary install; for pip/pipx/uv installs, prints the "
|
||||
"matching upgrade command instead.",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"-t",
|
||||
"--target",
|
||||
@@ -565,6 +581,9 @@ Examples:
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.update:
|
||||
sys.exit(0 if self_update() else 1)
|
||||
|
||||
if args.instruction and args.instruction_file:
|
||||
parser.error(
|
||||
"Cannot specify both --instruction and --instruction-file. Use one or the other."
|
||||
@@ -788,6 +807,8 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
|
||||
"[#60a5fa]discord.gg/strix-ai[/]"
|
||||
)
|
||||
console.print()
|
||||
if not args.non_interactive:
|
||||
notify_update(console)
|
||||
|
||||
|
||||
def pull_docker_image() -> None:
|
||||
@@ -851,6 +872,12 @@ def main() -> None:
|
||||
if args.config:
|
||||
apply_config_override(validate_config_file(args.config))
|
||||
|
||||
start_background_check()
|
||||
if not args.non_interactive and prompt_update_if_available(Console()):
|
||||
if is_binary_install() and sys.platform != "win32":
|
||||
os.execv(sys.executable, sys.argv) # noqa: S606 # nosec B606
|
||||
sys.exit(0)
|
||||
|
||||
check_docker_installed()
|
||||
pull_docker_image()
|
||||
|
||||
|
||||
@@ -0,0 +1,389 @@
|
||||
"""Update notifications and self-update for the strix CLI.
|
||||
|
||||
Follows the pattern used by tools like gh, uv, and pip: a background,
|
||||
rate-limited (once per 24h) check against the release source, a cached
|
||||
result in ``~/.strix``, a non-intrusive notice with the upgrade command
|
||||
for the detected install method, and a ``strix --update`` self-update
|
||||
path for the standalone binary install.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
import requests
|
||||
from rich.console import Console
|
||||
from rich.prompt import Prompt
|
||||
|
||||
from strix.telemetry._common import get_version
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
GITHUB_REPO = "usestrix/strix"
|
||||
PYPI_PACKAGE = "strix-agent"
|
||||
CHECK_INTERVAL_SECONDS = 24 * 60 * 60
|
||||
REQUEST_TIMEOUT_SECONDS = 5
|
||||
|
||||
_CACHE_PATH = Path.home() / ".strix" / "update-check.json"
|
||||
|
||||
_background_thread: threading.Thread | None = None
|
||||
|
||||
|
||||
def _is_disabled() -> bool:
|
||||
return bool(os.environ.get("STRIX_NO_UPDATE_CHECK")) or any(
|
||||
os.environ.get(key)
|
||||
for key in ("CI", "GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "BUILDKITE", "CIRCLECI")
|
||||
)
|
||||
|
||||
|
||||
def is_binary_install() -> bool:
|
||||
return bool(getattr(sys, "frozen", False))
|
||||
|
||||
|
||||
def get_install_method() -> str:
|
||||
if is_binary_install():
|
||||
return "binary"
|
||||
prefix = str(Path(sys.prefix)).replace("\\", "/")
|
||||
if "/pipx/" in prefix or prefix.endswith("/pipx"):
|
||||
return "pipx"
|
||||
if "/uv/tools/" in prefix:
|
||||
return "uv"
|
||||
return "pip"
|
||||
|
||||
|
||||
def get_upgrade_command(method: str | None = None) -> str:
|
||||
method = method or get_install_method()
|
||||
commands = {
|
||||
"binary": "strix --update",
|
||||
"pipx": "pipx upgrade strix-agent",
|
||||
"uv": "uv tool upgrade strix-agent",
|
||||
"pip": "pip install --upgrade strix-agent",
|
||||
}
|
||||
return commands[method]
|
||||
|
||||
|
||||
def _parse_version(value: str) -> tuple[int, ...] | None:
|
||||
parts = value.strip().lstrip("v").split(".")
|
||||
try:
|
||||
return tuple(int(part) for part in parts)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _is_newer(latest: str, current: str) -> bool:
|
||||
latest_parts = _parse_version(latest)
|
||||
current_parts = _parse_version(current)
|
||||
if latest_parts is None or current_parts is None:
|
||||
return False
|
||||
return latest_parts > current_parts
|
||||
|
||||
|
||||
def _fetch_latest_version() -> str | None:
|
||||
try:
|
||||
if is_binary_install():
|
||||
response = requests.get(
|
||||
f"https://api.github.com/repos/{GITHUB_REPO}/releases/latest",
|
||||
timeout=REQUEST_TIMEOUT_SECONDS,
|
||||
)
|
||||
response.raise_for_status()
|
||||
tag = response.json().get("tag_name", "")
|
||||
return tag.lstrip("v") or None
|
||||
response = requests.get(
|
||||
f"https://pypi.org/pypi/{PYPI_PACKAGE}/json",
|
||||
timeout=REQUEST_TIMEOUT_SECONDS,
|
||||
)
|
||||
response.raise_for_status()
|
||||
version = response.json().get("info", {}).get("version")
|
||||
return str(version) if version else None
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("update check failed", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def _fetch_asset_digest(version: str, filename: str) -> str | None:
|
||||
"""Return the expected sha256 (hex) for a release asset, if the API provides one."""
|
||||
try:
|
||||
response = requests.get(
|
||||
f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/v{version}",
|
||||
timeout=REQUEST_TIMEOUT_SECONDS,
|
||||
)
|
||||
response.raise_for_status()
|
||||
for asset in response.json().get("assets", []):
|
||||
if asset.get("name") == filename:
|
||||
digest = asset.get("digest") or ""
|
||||
if digest.startswith("sha256:"):
|
||||
return digest.removeprefix("sha256:")
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("release asset digest lookup failed", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def _sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as f:
|
||||
for chunk in iter(lambda: f.read(1 << 20), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _read_cache() -> dict[str, object]:
|
||||
try:
|
||||
with _CACHE_PATH.open(encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
if isinstance(data, dict):
|
||||
return cast("dict[str, object]", data)
|
||||
except Exception: # noqa: BLE001, S110
|
||||
pass # nosec B110
|
||||
return {}
|
||||
|
||||
|
||||
def _write_cache(**fields: object) -> None:
|
||||
try:
|
||||
cache = _read_cache()
|
||||
cache.update(fields)
|
||||
_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
_CACHE_PATH.write_text(json.dumps(cache), encoding="utf-8")
|
||||
except Exception: # noqa: BLE001, S110
|
||||
pass # nosec B110
|
||||
|
||||
|
||||
def skip_version(version: str) -> None:
|
||||
"""Remember not to prompt again for this version (newer releases still notify)."""
|
||||
_write_cache(skipped_version=version)
|
||||
|
||||
|
||||
def _refresh_cache() -> None:
|
||||
latest = _fetch_latest_version()
|
||||
if latest:
|
||||
_write_cache(latest_version=latest, checked_at=time.time())
|
||||
|
||||
|
||||
def start_background_check() -> None:
|
||||
"""Refresh the cached latest-version info in a daemon thread (at most once per 24h)."""
|
||||
global _background_thread # noqa: PLW0603
|
||||
if _is_disabled():
|
||||
return
|
||||
cache = _read_cache()
|
||||
checked_at = cache.get("checked_at")
|
||||
if isinstance(checked_at, int | float) and time.time() - checked_at < CHECK_INTERVAL_SECONDS:
|
||||
return
|
||||
_background_thread = threading.Thread(target=_refresh_cache, daemon=True)
|
||||
_background_thread.start()
|
||||
|
||||
|
||||
def get_available_update(*, respect_skip: bool = True) -> str | None:
|
||||
"""Return the newer version from the cache, or None if up to date / unknown."""
|
||||
if _is_disabled():
|
||||
return None
|
||||
if _background_thread is not None:
|
||||
_background_thread.join(timeout=0.2)
|
||||
cache = _read_cache()
|
||||
latest = cache.get("latest_version")
|
||||
current = get_version()
|
||||
if not isinstance(latest, str) or current == "unknown" or not _is_newer(latest, current):
|
||||
return None
|
||||
if respect_skip and cache.get("skipped_version") == latest:
|
||||
return None
|
||||
return latest
|
||||
|
||||
|
||||
def notify_update(console: Console) -> None:
|
||||
latest = get_available_update()
|
||||
if not latest:
|
||||
return
|
||||
console.print(
|
||||
f"[#eab308]A new version of strix is available:[/] "
|
||||
f"[dim]{get_version()}[/] [dim]→[/] [bold #22c55e]{latest}[/]"
|
||||
f" [dim]·[/] [#60a5fa]{get_upgrade_command()}[/]"
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
def run_package_upgrade(console: Console, method: str) -> bool:
|
||||
"""Upgrade a package-manager install by running its upgrade command."""
|
||||
command = get_upgrade_command(method).split()
|
||||
console.print(f"[dim]Running[/] [#60a5fa]{' '.join(command)}[/]")
|
||||
try:
|
||||
result = subprocess.run(command, check=False) # noqa: S603
|
||||
except OSError as e:
|
||||
console.print(f"[bold red]Update failed:[/] {e}")
|
||||
return False
|
||||
if result.returncode != 0:
|
||||
console.print(
|
||||
f"[bold red]Update failed[/] [dim](exit code {result.returncode}).[/] "
|
||||
f"Run it manually: [#60a5fa]{get_upgrade_command(method)}[/]"
|
||||
)
|
||||
return False
|
||||
console.print("[#22c55e]✓ strix updated — restart the scan to use the new version[/]")
|
||||
return True
|
||||
|
||||
|
||||
def prompt_update_if_available(console: Console) -> bool:
|
||||
"""Offer an interactive update before a scan starts.
|
||||
|
||||
Returns True if strix was updated (caller should re-exec / exit).
|
||||
"""
|
||||
latest = get_available_update()
|
||||
if not latest or not sys.stdin.isatty() or not sys.stdout.isatty():
|
||||
return False
|
||||
console.print()
|
||||
console.print(
|
||||
f"[#eab308]A new version of strix is available:[/] "
|
||||
f"[dim]{get_version()}[/] [dim]→[/] [bold #22c55e]{latest}[/]"
|
||||
)
|
||||
console.print(
|
||||
"[dim] y — update now n — not now (ask again next run) s — skip this version[/]"
|
||||
)
|
||||
choice = Prompt.ask("Update strix?", choices=["y", "n", "s"], default="n")
|
||||
console.print()
|
||||
if choice == "s":
|
||||
skip_version(latest)
|
||||
return False
|
||||
if choice != "y":
|
||||
return False
|
||||
method = get_install_method()
|
||||
if method == "binary":
|
||||
return self_update(console, version=latest)
|
||||
return run_package_upgrade(console, method)
|
||||
|
||||
|
||||
def _release_target() -> str | None:
|
||||
raw_os = platform.system().lower()
|
||||
os_name = {"darwin": "macos", "linux": "linux", "windows": "windows"}.get(raw_os)
|
||||
arch = platform.machine().lower()
|
||||
arch = {"aarch64": "arm64", "amd64": "x86_64"}.get(arch, arch)
|
||||
if os_name is None:
|
||||
return None
|
||||
target = f"{os_name}-{arch}"
|
||||
supported = {"linux-x86_64", "macos-x86_64", "macos-arm64", "windows-x86_64"}
|
||||
return target if target in supported else None
|
||||
|
||||
|
||||
def _download_and_replace(version: str, target: str, console: Console) -> bool:
|
||||
is_windows = target.startswith("windows")
|
||||
archive_ext = ".zip" if is_windows else ".tar.gz"
|
||||
filename = f"strix-{version}-{target}{archive_ext}"
|
||||
url = f"https://github.com/{GITHUB_REPO}/releases/download/v{version}/{filename}"
|
||||
binary_name = f"strix-{version}-{target}" + (".exe" if is_windows else "")
|
||||
current_exe = Path(sys.executable).resolve()
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp_dir = Path(tmp)
|
||||
archive_path = tmp_dir / filename
|
||||
console.print(f"[dim]Downloading[/] {url}")
|
||||
with requests.get( # nosec B113
|
||||
url,
|
||||
stream=True,
|
||||
timeout=REQUEST_TIMEOUT_SECONDS * 12,
|
||||
) as response:
|
||||
response.raise_for_status()
|
||||
with archive_path.open("wb") as f:
|
||||
for chunk in response.iter_content(chunk_size=1 << 20):
|
||||
f.write(chunk)
|
||||
|
||||
expected_digest = _fetch_asset_digest(version, filename)
|
||||
if expected_digest:
|
||||
actual_digest = _sha256_file(archive_path)
|
||||
if actual_digest != expected_digest:
|
||||
raise RuntimeError(
|
||||
f"checksum mismatch for {filename}: "
|
||||
f"expected sha256 {expected_digest}, got {actual_digest}"
|
||||
)
|
||||
else:
|
||||
console.print("[dim yellow]No published checksum available; skipping verification[/]")
|
||||
|
||||
if is_windows:
|
||||
with zipfile.ZipFile(archive_path) as zf:
|
||||
zf.extract(binary_name, tmp_dir)
|
||||
else:
|
||||
with tarfile.open(archive_path, "r:gz") as tf:
|
||||
tf.extract(binary_name, tmp_dir, filter="data")
|
||||
|
||||
new_binary = tmp_dir / binary_name
|
||||
new_binary.chmod(new_binary.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
||||
|
||||
staged = current_exe.with_name(current_exe.name + ".new")
|
||||
try:
|
||||
shutil.copy2(new_binary, staged)
|
||||
if is_windows:
|
||||
# Windows can't replace a running executable in place; move it aside first.
|
||||
old = current_exe.with_name(current_exe.name + ".old")
|
||||
old.unlink(missing_ok=True)
|
||||
current_exe.rename(old)
|
||||
try:
|
||||
staged.replace(current_exe)
|
||||
except Exception:
|
||||
old.rename(current_exe)
|
||||
raise
|
||||
else:
|
||||
staged.replace(current_exe)
|
||||
except Exception:
|
||||
staged.unlink(missing_ok=True)
|
||||
raise
|
||||
return True
|
||||
|
||||
|
||||
def self_update(console: Console | None = None, version: str | None = None) -> bool:
|
||||
"""Replace the running standalone binary with the latest release.
|
||||
|
||||
Returns True on success. For package-manager installs this only
|
||||
prints the right upgrade command and returns False.
|
||||
"""
|
||||
console = console or Console()
|
||||
|
||||
if not is_binary_install():
|
||||
method = get_install_method()
|
||||
console.print(
|
||||
f"[#eab308]This strix was installed via {method};[/] "
|
||||
f"upgrade it with: [#60a5fa]{get_upgrade_command(method)}[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
latest = version or _fetch_latest_version()
|
||||
if not latest:
|
||||
console.print("[bold red]Could not determine the latest strix version.[/]")
|
||||
return False
|
||||
|
||||
current = get_version()
|
||||
if current != "unknown" and not _is_newer(latest, current):
|
||||
console.print(f"[#22c55e]strix {current} is already the latest version.[/]")
|
||||
return True
|
||||
|
||||
target = _release_target()
|
||||
if not target:
|
||||
console.print(
|
||||
f"[bold red]No prebuilt binary for this platform "
|
||||
f"({platform.system()}/{platform.machine()}).[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
try:
|
||||
_download_and_replace(latest, target, console)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.debug("self-update failed", exc_info=True)
|
||||
console.print(f"[bold red]Update failed:[/] {e}")
|
||||
console.print(
|
||||
"[dim]You can reinstall manually with:[/] "
|
||||
"[#60a5fa]curl -sSL https://strix.ai/install | bash[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
_write_cache(latest_version=latest, checked_at=time.time())
|
||||
console.print(f"[#22c55e]✓ Updated strix to {latest}[/]")
|
||||
return True
|
||||
@@ -0,0 +1,172 @@
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import platform
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from rich.console import Console
|
||||
|
||||
from strix.interface import update_check
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(update_check, "_CACHE_PATH", tmp_path / "update-check.json")
|
||||
monkeypatch.setattr(update_check, "_background_thread", None)
|
||||
monkeypatch.delenv("STRIX_NO_UPDATE_CHECK", raising=False)
|
||||
for key in ("CI", "GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "BUILDKITE", "CIRCLECI"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("latest", "current", "expected"),
|
||||
[
|
||||
("1.2.0", "1.1.0", True),
|
||||
("1.1.0", "1.1.0", False),
|
||||
("1.0.9", "1.1.0", False),
|
||||
("2.0.0", "1.99.99", True),
|
||||
("1.10.0", "1.9.0", True),
|
||||
("v1.2.0", "1.1.0", True),
|
||||
("not-a-version", "1.1.0", False),
|
||||
("1.2.0", "unknown", False),
|
||||
],
|
||||
)
|
||||
def test_is_newer(latest: str, current: str, expected: bool) -> None:
|
||||
assert update_check._is_newer(latest, current) is expected
|
||||
|
||||
|
||||
def test_get_available_update_from_cache(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "9.9.9", "checked_at": time.time()})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
assert update_check.get_available_update() == "9.9.9"
|
||||
|
||||
|
||||
def test_get_available_update_up_to_date(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "1.0.0", "checked_at": time.time()})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
assert update_check.get_available_update() is None
|
||||
|
||||
|
||||
def test_get_available_update_disabled_by_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "9.9.9", "checked_at": time.time()})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
monkeypatch.setenv("STRIX_NO_UPDATE_CHECK", "1")
|
||||
assert update_check.get_available_update() is None
|
||||
|
||||
|
||||
def test_get_available_update_disabled_in_ci(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "9.9.9", "checked_at": time.time()})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
monkeypatch.setenv("CI", "true")
|
||||
assert update_check.get_available_update() is None
|
||||
|
||||
|
||||
def test_get_available_update_corrupt_cache(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text("{not json")
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
assert update_check.get_available_update() is None
|
||||
|
||||
|
||||
def test_background_check_skipped_when_fresh(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "1.0.0", "checked_at": time.time()})
|
||||
)
|
||||
called = False
|
||||
|
||||
def fake_refresh() -> None:
|
||||
nonlocal called
|
||||
called = True
|
||||
|
||||
monkeypatch.setattr(update_check, "_refresh_cache", fake_refresh)
|
||||
update_check.start_background_check()
|
||||
assert update_check._background_thread is None
|
||||
assert called is False
|
||||
|
||||
|
||||
def test_background_check_runs_when_stale(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "1.0.0", "checked_at": time.time() - 2 * 24 * 60 * 60})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "_fetch_latest_version", lambda: "1.2.3")
|
||||
update_check.start_background_check()
|
||||
assert update_check._background_thread is not None
|
||||
update_check._background_thread.join(timeout=5)
|
||||
cache = json.loads(update_check._CACHE_PATH.read_text())
|
||||
assert cache["latest_version"] == "1.2.3"
|
||||
|
||||
|
||||
def test_skipped_version_suppresses_update(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps({"latest_version": "9.9.9", "checked_at": time.time()})
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
update_check.skip_version("9.9.9")
|
||||
assert update_check.get_available_update() is None
|
||||
assert update_check.get_available_update(respect_skip=False) == "9.9.9"
|
||||
|
||||
|
||||
def test_newer_release_overrides_skipped_version(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update_check._CACHE_PATH.write_text(
|
||||
json.dumps(
|
||||
{"latest_version": "9.9.10", "checked_at": time.time(), "skipped_version": "9.9.9"}
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
assert update_check.get_available_update() == "9.9.10"
|
||||
|
||||
|
||||
def test_write_cache_preserves_existing_fields() -> None:
|
||||
update_check.skip_version("9.9.9")
|
||||
update_check._write_cache(latest_version="1.2.3", checked_at=123.0)
|
||||
cache = json.loads(update_check._CACHE_PATH.read_text())
|
||||
assert cache == {"latest_version": "1.2.3", "checked_at": 123.0, "skipped_version": "9.9.9"}
|
||||
|
||||
|
||||
def test_get_upgrade_command_all_methods() -> None:
|
||||
assert update_check.get_upgrade_command("binary") == "strix --update"
|
||||
assert update_check.get_upgrade_command("pipx") == "pipx upgrade strix-agent"
|
||||
assert update_check.get_upgrade_command("uv") == "uv tool upgrade strix-agent"
|
||||
assert update_check.get_upgrade_command("pip") == "pip install --upgrade strix-agent"
|
||||
|
||||
|
||||
def test_self_update_non_binary_prints_command(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(update_check, "is_binary_install", lambda: False)
|
||||
buffer = io.StringIO()
|
||||
assert update_check.self_update(Console(file=buffer)) is False
|
||||
assert "upgrade" in buffer.getvalue()
|
||||
|
||||
|
||||
def test_self_update_already_latest(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(update_check, "is_binary_install", lambda: True)
|
||||
monkeypatch.setattr(update_check, "_fetch_latest_version", lambda: "1.0.0")
|
||||
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
|
||||
assert update_check.self_update() is True
|
||||
|
||||
|
||||
def test_sha256_file(tmp_path: Path) -> None:
|
||||
path = tmp_path / "blob"
|
||||
path.write_bytes(b"strix")
|
||||
assert update_check._sha256_file(path) == hashlib.sha256(b"strix").hexdigest()
|
||||
|
||||
|
||||
def test_release_target(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(platform, "system", lambda: "Linux")
|
||||
monkeypatch.setattr(platform, "machine", lambda: "x86_64")
|
||||
assert update_check._release_target() == "linux-x86_64"
|
||||
|
||||
monkeypatch.setattr(platform, "system", lambda: "Darwin")
|
||||
monkeypatch.setattr(platform, "machine", lambda: "arm64")
|
||||
assert update_check._release_target() == "macos-arm64"
|
||||
|
||||
monkeypatch.setattr(platform, "machine", lambda: "riscv64")
|
||||
assert update_check._release_target() is None
|
||||
Reference in New Issue
Block a user