diff --git a/src/octopal/cli/main.py b/src/octopal/cli/main.py index 5fe4e7a..2e34859 100644 --- a/src/octopal/cli/main.py +++ b/src/octopal/cli/main.py @@ -3,15 +3,19 @@ import asyncio import json import logging +import os +import re import shutil import subprocess import threading import time from collections import deque from datetime import UTC, datetime +from itertools import zip_longest from pathlib import Path from typing import Any +import httpx import typer from rich import box from rich.align import Align @@ -65,6 +69,11 @@ console = Console() logger = logging.getLogger(__name__) +_DEFAULT_RELEASE_REPO = "pmbstyle/Octopal" +_VERSION_CHECK_CACHE_NAME = "version_check.json" +_VERSION_CHECK_TTL_SECONDS = 6 * 60 * 60 +_VERSION_CHECK_TIMEOUT_SECONDS = 1.5 + @app.command() def configure() -> None: @@ -79,6 +88,287 @@ def _build_telegram_bot(token: str) -> Any: return Bot(token=token) +def _normalize_release_version(raw_version: str) -> str | None: + cleaned = raw_version.strip() + if not cleaned: + return None + + if cleaned.lower().startswith("v"): + cleaned = cleaned[1:] + + if not re.fullmatch(r"\d+(?:\.\d+)*", cleaned): + return None + return cleaned + + +def _version_key(raw_version: str) -> tuple[int, ...] | None: + normalized = _normalize_release_version(raw_version) + if normalized is None: + return None + return tuple(int(part) for part in normalized.split(".")) + + +def _is_remote_version_newer(local_version: str, remote_version: str) -> bool: + local_key = _version_key(local_version) + remote_key = _version_key(remote_version) + if local_key is None or remote_key is None: + return False + + for local_part, remote_part in zip_longest(local_key, remote_key, fillvalue=0): + if remote_part > local_part: + return True + if remote_part < local_part: + return False + return False + + +def _project_root() -> Path: + return Path(__file__).resolve().parents[3] + + +def _guess_update_command() -> str: + if (_project_root() / ".git").exists(): + return "git pull && uv sync" + return "install the latest GitHub release" + + +def _detect_release_repo_slug() -> str: + try: + result = subprocess.run( + ["git", "config", "--get", "remote.origin.url"], + cwd=_project_root(), + capture_output=True, + text=True, + encoding="utf-8", + timeout=0.75, + check=False, + ) + except Exception: + return _DEFAULT_RELEASE_REPO + + remote_url = result.stdout.strip() + if not remote_url: + return _DEFAULT_RELEASE_REPO + + https_match = re.search(r"github\.com[:/](?P[^/\s]+/[^/\s]+?)(?:\.git)?$", remote_url) + if not https_match: + return _DEFAULT_RELEASE_REPO + return https_match.group("slug") + + +def _version_check_cache_path(settings: Settings) -> Path: + return settings.state_dir / _VERSION_CHECK_CACHE_NAME + + +def _read_cached_release_info(settings: Settings, repo_slug: str) -> tuple[str, str] | None: + cache_path = _version_check_cache_path(settings) + if not cache_path.exists(): + return None + + try: + payload = json.loads(cache_path.read_text(encoding="utf-8")) + except Exception: + return None + + if payload.get("repo") != repo_slug: + return None + + checked_at_raw = str(payload.get("checked_at") or "").strip() + cached_version = _normalize_release_version(str(payload.get("version") or "")) + cached_url = str(payload.get("url") or "").strip() + if not checked_at_raw or cached_version is None or not cached_url: + return None + + try: + checked_at = datetime.fromisoformat(checked_at_raw) + except ValueError: + return None + + if checked_at.tzinfo is None: + checked_at = checked_at.replace(tzinfo=UTC) + if (datetime.now(UTC) - checked_at).total_seconds() > _VERSION_CHECK_TTL_SECONDS: + return None + + return cached_version, cached_url + + +def _write_cached_release_info(settings: Settings, repo_slug: str, version: str, url: str) -> None: + cache_path = _version_check_cache_path(settings) + cache_path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "repo": repo_slug, + "version": version, + "url": url, + "checked_at": datetime.now(UTC).isoformat(), + } + cache_path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + + +def _fetch_latest_release_info_from_github(repo_slug: str) -> tuple[str, str] | None: + api_url = f"https://api.github.com/repos/{repo_slug}/releases/latest" + headers = { + "Accept": "application/vnd.github+json", + "User-Agent": "octopal-version-check", + } + timeout = httpx.Timeout(_VERSION_CHECK_TIMEOUT_SECONDS, connect=0.5) + + try: + with httpx.Client(timeout=timeout, headers=headers, follow_redirects=True) as client: + response = client.get(api_url) + if response.status_code == 404: + return None + response.raise_for_status() + except Exception: + return None + + try: + payload = response.json() + except ValueError: + return None + + latest_version = _normalize_release_version(str(payload.get("tag_name") or payload.get("name") or "")) + if latest_version is None: + return None + + release_url = str(payload.get("html_url") or f"https://github.com/{repo_slug}/releases").strip() + return latest_version, release_url + + +def _get_latest_release_info(settings: Settings) -> tuple[str, str] | None: + repo_slug = _detect_release_repo_slug() + cached = _read_cached_release_info(settings, repo_slug) + if cached is not None: + return cached + + fetched = _fetch_latest_release_info_from_github(repo_slug) + if fetched is None: + return None + + latest_version, release_url = fetched + try: + _write_cached_release_info(settings, repo_slug, latest_version, release_url) + except Exception: + logger.debug("Failed to cache latest release info", exc_info=True) + return latest_version, release_url + + +def _maybe_warn_about_newer_release(settings: Settings) -> None: + if os.environ.get("OCTOPAL_SKIP_VERSION_CHECK") == "1": + return + + from octopal import __version__ + + latest_release = _get_latest_release_info(settings) + if latest_release is None: + return + + latest_version, release_url = latest_release + if not _is_remote_version_newer(__version__, latest_version): + return + + console.print( + f"[bold yellow]Update available:[/bold yellow] Octopal v{latest_version} is out " + f"(local v{__version__})." + ) + console.print(f" [dim]Release:[/dim] [cyan]{release_url}[/cyan]") + console.print(f" [dim]Suggested update:[/dim] [magenta]{_guess_update_command()}[/magenta]") + + +def _run_capture(command: list[str], *, cwd: Path, timeout: float = 10.0) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + cwd=cwd, + capture_output=True, + text=True, + encoding="utf-8", + timeout=timeout, + check=False, + ) + + +def _list_meaningful_worktree_changes(project_root: Path) -> list[str] | None: + diff = _run_capture(["git", "diff", "--name-only"], cwd=project_root) + if diff.returncode != 0: + return None + + changes: list[str] = [] + for raw_path in diff.stdout.splitlines(): + path = raw_path.strip() + if not path: + continue + + stats = _run_capture(["git", "diff", "--numstat", "--", path], cwd=project_root) + if stats.returncode != 0: + return None + + stats_lines = [line.strip() for line in stats.stdout.splitlines() if line.strip()] + if not stats_lines: + continue + + is_meaningful = False + for line in stats_lines: + parts = line.split("\t") + if len(parts) < 3: + is_meaningful = True + break + + added, deleted = parts[0].strip(), parts[1].strip() + if added == "0" and deleted == "0": + continue + + # Binary/content changes often show `-` counters and should still block update. + is_meaningful = True + break + + if is_meaningful: + changes.append(path) + + return changes + + +def _git_checkout_ready_for_update(project_root: Path) -> tuple[bool, str | None]: + if not (project_root / ".git").exists(): + return False, "This install is not a git checkout." + + if shutil.which("git") is None: + return False, "`git` is not installed or not on PATH." + + status = _run_capture(["git", "status", "--porcelain"], cwd=project_root) + if status.returncode != 0: + detail = status.stderr.strip() or status.stdout.strip() or "unknown git status error" + return False, f"Could not inspect git status: {detail}" + + if status.stdout.strip(): + meaningful_changes = _list_meaningful_worktree_changes(project_root) + if meaningful_changes is None: + return False, "Could not determine whether local changes are content changes or mode-only changes." + if meaningful_changes: + names = ", ".join(meaningful_changes[:3]) + if len(meaningful_changes) > 3: + names += ", ..." + return False, f"Working tree has local content changes: {names}. Commit or stash them first." + + return True, None + + +def _perform_git_update(project_root: Path) -> tuple[bool, str]: + if shutil.which("uv") is None: + return False, "`uv` is not installed or not on PATH." + + pull = _run_capture(["git", "pull", "--ff-only"], cwd=project_root, timeout=60.0) + if pull.returncode != 0: + detail = pull.stderr.strip() or pull.stdout.strip() or "git pull failed" + return False, f"`git pull --ff-only` failed: {detail}" + + sync = _run_capture(["uv", "sync"], cwd=project_root, timeout=120.0) + if sync.returncode != 0: + detail = sync.stderr.strip() or sync.stdout.strip() or "uv sync failed" + return False, f"`uv sync` failed: {detail}" + + pull_summary = pull.stdout.strip() or "Already up to date." + return True, pull_summary + + def _build_gateway_app(*args: Any, **kwargs: Any) -> Any: from octopal.gateway.app import build_app @@ -422,6 +712,8 @@ def start( console.print("Use [magenta]octopal stop[/magenta] first, then start again.") raise typer.Exit(code=1) + _maybe_warn_about_newer_release(settings) + if not foreground: print_banner() _start_background() @@ -502,6 +794,7 @@ def _start_background() -> None: env = os.environ.copy() existing_pp = env.get("PYTHONPATH", "") env["PYTHONPATH"] = f"{src_dir}{os.pathsep}{existing_pp}" if existing_pp else str(src_dir) + env["OCTOPAL_SKIP_VERSION_CHECK"] = "1" # Redirect output to a file for debugging log_dir = project_root / "data" / "logs" @@ -663,6 +956,35 @@ def version() -> None: console.print(f"Octopal [bold cyan]v{__version__}[/bold cyan]") +@app.command() +def update() -> None: + """Update Octopal from the current git checkout.""" + project_root = _project_root() + ready, reason = _git_checkout_ready_for_update(project_root) + if not ready: + console.print(f"[bold red]Update unavailable:[/bold red] {reason}") + raise typer.Exit(code=1) + + running_pids = list_octopal_runtime_pids() + if running_pids: + console.print("[yellow]Octopal is running right now.[/yellow]") + console.print("Update will continue, but restart after it finishes to pick up the new code.") + + with console.status("[bold cyan]Updating Octopal...[/bold cyan]", spinner="dots"): + ok, detail = _perform_git_update(project_root) + + if not ok: + console.print(f"[bold red]Update failed:[/bold red] {detail}") + raise typer.Exit(code=1) + + console.print("[bold green][V] Octopal updated.[/bold green]") + console.print(f" [dim]Git:[/dim] {detail}") + if running_pids: + console.print(" [dim]Next step:[/dim] [magenta]uv run octopal restart[/magenta]") + else: + console.print(" [dim]Next step:[/dim] [magenta]uv run octopal start[/magenta]") + + @app.command() def status() -> None: config_ok = True diff --git a/tests/test_cli_update.py b/tests/test_cli_update.py new file mode 100644 index 0000000..b4c44f2 --- /dev/null +++ b/tests/test_cli_update.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import subprocess + +from typer.testing import CliRunner + +from octopal.cli.main import _git_checkout_ready_for_update, app + +runner = CliRunner() + + +def test_update_rejects_dirty_git_checkout(monkeypatch, tmp_path) -> None: + monkeypatch.setattr("octopal.cli.main._project_root", lambda: tmp_path) + monkeypatch.setattr("octopal.cli.main._git_checkout_ready_for_update", lambda _root: (False, "dirty tree")) + + result = runner.invoke(app, ["update"]) + + assert result.exit_code == 1 + assert "Update unavailable:" in result.stdout + assert "dirty tree" in result.stdout + + +def test_update_runs_git_pull_and_uv_sync(monkeypatch, tmp_path) -> None: + monkeypatch.setattr("octopal.cli.main._project_root", lambda: tmp_path) + monkeypatch.setattr("octopal.cli.main._git_checkout_ready_for_update", lambda _root: (True, None)) + monkeypatch.setattr("octopal.cli.main.list_octopal_runtime_pids", lambda: []) + monkeypatch.setattr( + "octopal.cli.main._perform_git_update", + lambda _root: (True, "Already up to date."), + ) + + result = runner.invoke(app, ["update"]) + + assert result.exit_code == 0 + assert "Octopal updated." in result.stdout + assert "Already up to date." in result.stdout + assert "uv run octopal start" in result.stdout + + +def test_update_warns_when_runtime_is_active(monkeypatch, tmp_path) -> None: + monkeypatch.setattr("octopal.cli.main._project_root", lambda: tmp_path) + monkeypatch.setattr("octopal.cli.main._git_checkout_ready_for_update", lambda _root: (True, None)) + monkeypatch.setattr("octopal.cli.main.list_octopal_runtime_pids", lambda: [12345]) + monkeypatch.setattr( + "octopal.cli.main._perform_git_update", + lambda _root: (True, "Updating abc..def"), + ) + + result = runner.invoke(app, ["update"]) + + assert result.exit_code == 0 + assert "Octopal is running right now." in result.stdout + assert "uv run octopal restart" in result.stdout + + +def test_git_checkout_ready_allows_mode_only_changes(monkeypatch, tmp_path) -> None: + (tmp_path / ".git").mkdir() + + def fake_run_capture(command: list[str], *, cwd, timeout=10.0): + if command == ["git", "status", "--porcelain"]: + return subprocess.CompletedProcess(command, 0, stdout=" M scripts/bootstrap.sh\n", stderr="") + if command == ["git", "diff", "--name-only"]: + return subprocess.CompletedProcess(command, 0, stdout="scripts/bootstrap.sh\n", stderr="") + if command == ["git", "diff", "--numstat", "--", "scripts/bootstrap.sh"]: + return subprocess.CompletedProcess(command, 0, stdout="0\t0\tscripts/bootstrap.sh\n", stderr="") + raise AssertionError(f"unexpected command: {command}") + + monkeypatch.setattr("octopal.cli.main.shutil.which", lambda _name: "/usr/bin/git") + monkeypatch.setattr("octopal.cli.main._run_capture", fake_run_capture) + + assert _git_checkout_ready_for_update(tmp_path) == (True, None) + + +def test_git_checkout_ready_blocks_real_content_changes(monkeypatch, tmp_path) -> None: + (tmp_path / ".git").mkdir() + + def fake_run_capture(command: list[str], *, cwd, timeout=10.0): + if command == ["git", "status", "--porcelain"]: + return subprocess.CompletedProcess(command, 0, stdout=" M scripts/bootstrap.sh\n", stderr="") + if command == ["git", "diff", "--name-only"]: + return subprocess.CompletedProcess(command, 0, stdout="scripts/bootstrap.sh\n", stderr="") + if command == ["git", "diff", "--numstat", "--", "scripts/bootstrap.sh"]: + return subprocess.CompletedProcess(command, 0, stdout="3\t1\tscripts/bootstrap.sh\n", stderr="") + raise AssertionError(f"unexpected command: {command}") + + monkeypatch.setattr("octopal.cli.main.shutil.which", lambda _name: "/usr/bin/git") + monkeypatch.setattr("octopal.cli.main._run_capture", fake_run_capture) + + ok, reason = _git_checkout_ready_for_update(tmp_path) + assert ok is False + assert "scripts/bootstrap.sh" in str(reason) diff --git a/tests/test_cli_version_check.py b/tests/test_cli_version_check.py new file mode 100644 index 0000000..536e1cc --- /dev/null +++ b/tests/test_cli_version_check.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta + +from octopal.cli.main import ( + _VERSION_CHECK_TTL_SECONDS, + _get_latest_release_info, + _is_remote_version_newer, + _maybe_warn_about_newer_release, + _normalize_release_version, +) +from octopal.infrastructure.config.settings import Settings + + +def _build_settings(tmp_path) -> Settings: + return Settings( + TELEGRAM_BOT_TOKEN="123:abc", + OCTOPAL_STATE_DIR=tmp_path / "state", + OCTOPAL_WORKSPACE_DIR=tmp_path / "workspace", + ) + + +def test_normalize_release_version_strips_v_prefix() -> None: + assert _normalize_release_version("v2026.04.15") == "2026.04.15" + assert _normalize_release_version("2026.04.15.1") == "2026.04.15.1" + assert _normalize_release_version("release-2026.04.15") is None + + +def test_remote_version_newer_compares_date_based_versions() -> None: + assert _is_remote_version_newer("2026.04.14", "2026.04.15") is True + assert _is_remote_version_newer("2026.04.14", "2026.04.14.1") is True + assert _is_remote_version_newer("2026.04.14.1", "2026.04.14") is False + assert _is_remote_version_newer("2026.04.14", "2026.04.14") is False + + +def test_get_latest_release_info_uses_fresh_cache(tmp_path, monkeypatch) -> None: + settings = _build_settings(tmp_path) + cache_path = settings.state_dir / "version_check.json" + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text( + json.dumps( + { + "repo": "owner/repo", + "version": "2026.04.15", + "url": "https://example.test/releases/tag/v2026.04.15", + "checked_at": datetime.now(UTC).isoformat(), + } + ), + encoding="utf-8", + ) + + monkeypatch.setattr("octopal.cli.main._detect_release_repo_slug", lambda: "owner/repo") + monkeypatch.setattr( + "octopal.cli.main._fetch_latest_release_info_from_github", + lambda _repo_slug: (_ for _ in ()).throw(AssertionError("network fetch should not run")), + ) + + assert _get_latest_release_info(settings) == ( + "2026.04.15", + "https://example.test/releases/tag/v2026.04.15", + ) + + +def test_get_latest_release_info_refreshes_stale_cache(tmp_path, monkeypatch) -> None: + settings = _build_settings(tmp_path) + cache_path = settings.state_dir / "version_check.json" + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text( + json.dumps( + { + "repo": "owner/repo", + "version": "2026.04.14", + "url": "https://example.test/releases/tag/v2026.04.14", + "checked_at": (datetime.now(UTC) - timedelta(seconds=_VERSION_CHECK_TTL_SECONDS + 1)).isoformat(), + } + ), + encoding="utf-8", + ) + + monkeypatch.setattr("octopal.cli.main._detect_release_repo_slug", lambda: "owner/repo") + monkeypatch.setattr( + "octopal.cli.main._fetch_latest_release_info_from_github", + lambda _repo_slug: ("2026.04.15", "https://example.test/releases/tag/v2026.04.15"), + ) + + assert _get_latest_release_info(settings) == ( + "2026.04.15", + "https://example.test/releases/tag/v2026.04.15", + ) + + +def test_warns_when_new_release_is_available(tmp_path, monkeypatch) -> None: + settings = _build_settings(tmp_path) + printed: list[str] = [] + + monkeypatch.setattr( + "octopal.cli.main._get_latest_release_info", + lambda _settings: ("2026.04.15", "https://example.test/releases/tag/v2026.04.15"), + ) + monkeypatch.setattr("octopal.cli.main.console.print", lambda message="", *args, **kwargs: printed.append(str(message))) + + _maybe_warn_about_newer_release(settings) + + assert any("Update available:" in line for line in printed) + assert any("2026.04.15" in line for line in printed)