diff --git a/README.md b/README.md index f6c41ea..2b8fbe1 100644 --- a/README.md +++ b/README.md @@ -46,10 +46,12 @@ Got a **Gemini** subscription too? Two more backends use it instead of OpenAI: ` ## Upgrade ```bash -image-use update +image-use upgrade # install the latest release and refresh the skill +image-use upgrade --check # only report: image-use 0.29.2 -> 0.30.0 +image-use upgrade --json # the same as JSON ``` -It runs the `skills` manager for you — directly when `skills` is on PATH, through `npx` when it isn't (it usually isn't). Interactive runs check for a newer version at most once a day and upgrade automatically for the next run; failures fall back to a notice listing what changed. `IMAGE_USE_NO_AUTO_UPDATE=1` disables installation but keeps the check and notice, while `IMAGE_USE_NO_UPDATE_CHECK=1` disables both. `--quiet`/`--no-progress` never upgrades in the background. +`upgrade` (alias `update`) installs the newest GitHub release the same way this copy was installed — `skills update` for an `npx skills add` install, `git pull --ff-only` for a clone, a fresh copy of the script for a standalone file — then refreshes every other copy of the skill it finds (Claude Code plugin, clones and copies under `~/.agents/skills`, `~/.claude/skills`, `~/.codex/skills`). Any other command checks for a newer release at most once a day and prints one line to stderr when there is one. `IMAGE_USE_NO_UPDATE_CHECK=1` or the family-wide `USE_NO_UPDATE_CHECK=1` turns the check off; it is also skipped when `CI` is set. Nothing is installed until you run `upgrade`. **On 0.23.1 or earlier?** That self-update only looked for a global `skills` and gave up when it was missing, so it cannot deliver its own fix. Bootstrap once with (installs from before the rename are registered as `chatgpt-imagegen`): @@ -57,7 +59,7 @@ It runs the `skills` manager for you — directly when `skills` is on PATH, thro npx -y skills update chatgpt-imagegen ``` -After that `image-use update` works on its own. +After that `image-use upgrade` works on its own. ## Usage diff --git a/README.zh-CN.md b/README.zh-CN.md index e6e4bf9..b1d4ff0 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -46,10 +46,12 @@ sudo ln -sf image-use /usr/local/bin/chatgpt-imagegen # 可选:保留旧命令 ## 升级 ```bash -image-use update +image-use upgrade # 装最新版并刷新 skill +image-use upgrade --check # 只查不改:image-use 0.29.2 -> 0.30.0 +image-use upgrade --json # 同上,输出 JSON ``` -它会替你调用 `skills` 管理器——PATH 上有 `skills` 就直接用,没有则走 `npx`(通常都没有)。交互式运行每天最多检查一次新版并自动升级,下次运行生效;失败时会退回提醒并列出变更。`IMAGE_USE_NO_AUTO_UPDATE=1` 只关闭自动安装,`IMAGE_USE_NO_UPDATE_CHECK=1` 连检查也关闭。`--quiet`/`--no-progress` 不会在后台升级。 +`upgrade`(别名 `update`)按这份安装的来路装最新的 GitHub Release——`npx skills add` 装的走 `skills update`,git clone 的走 `git pull --ff-only`,单独拷贝的脚本直接换成新版——然后刷新它找到的其他 skill 副本(Claude Code 插件,以及 `~/.agents/skills`、`~/.claude/skills`、`~/.codex/skills` 下的 clone 和拷贝)。其他命令每天最多检查一次新版,有新版就往 stderr 打一行提示。设 `IMAGE_USE_NO_UPDATE_CHECK=1` 或全家通用的 `USE_NO_UPDATE_CHECK=1` 可关闭检查,设了 `CI` 时也不检查。不运行 `upgrade` 就不会安装任何东西。 **还停在 0.23.1 或更早?** 那时的自升级只找全局 `skills`,找不到就放弃,所以它没法把这个修复本身装进来。先手动破一次局(改名前的安装在 skills 里登记的名字是 `chatgpt-imagegen`): @@ -57,7 +59,7 @@ image-use update npx -y skills update chatgpt-imagegen ``` -之后 `image-use update` 就能自己跑了。 +之后 `image-use upgrade` 就能自己跑了。 ## 用法 diff --git a/SKILL.md b/SKILL.md index 469aad8..45375d3 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,6 +1,6 @@ --- name: "image-use" -version: "0.29.2" +version: "0.30.0" description: >- Backend-neutral image generation: create new raster images and looping GIF/WebP animations through the local one-file image-use CLI (formerly @@ -291,17 +291,9 @@ A vague prompt yields a useless figure. Make the prompt describe the figure's ** ## Error handling -**First step for any "which backend / why isn't web working" failure:** run `image-use doctor`. It reports, read-only, the CLI's own version vs. the latest on `main`, whether each backend is set up (codex token; chrome-use installed + version; relay connected; logged-in Chrome profiles), and **which one `auto` would pick** — turning a vague "no logged-in browser" into a precise checklist. +**First step for any "which backend / why isn't web working" failure:** run `image-use doctor`. It reports, read-only, the CLI's own version vs. the latest GitHub release, whether each backend is set up (codex token; chrome-use installed + version; relay connected; logged-in Chrome profiles), and **which one `auto` would pick** — turning a vague "no logged-in browser" into a precise checklist. -**Automatic updates.** `skills` has no scheduler of its own, so an interactive CLI run checks `main` at most once a day. When a newer version exists it invokes the same `skills update` path as the explicit command, then uses the new code on the next run. If automatic installation is unavailable or fails, it falls back to a short stderr notice that **lists what changed** since your version: - -``` -提示:image-use 0.14.0 可用(当前 0.12.0)。更新:image-use update - • 0.14.0:更新提示现在会列出每个新版本改了什么 - • 0.13.0:新增每天一次的新版本提示… -``` - -It never touches stdout and is skipped under `--quiet`/`--no-progress`; `doctor` checks unconditionally and prints the same change list. To turn checking off entirely, set `IMAGE_USE_NO_UPDATE_CHECK=1`. To keep the daily check and notice but disable automatic installation, set `IMAGE_USE_NO_AUTO_UPDATE=1`. When you see the fallback notice, run `image-use update` — it runs the `skills` manager for you, through npx when `skills` isn't on PATH (it usually isn't), so it works without a global install (or re-run the self-heal `curl`). +Update notices are covered under [Upgrade](#upgrade). | Symptom | Cause | Fix | | --- | --- | --- | @@ -319,6 +311,19 @@ It never touches stdout and is skipped under `--quiet`/`--no-progress`; `doctor` | `chatgpt.com rate-limited this account ('Too many requests') …` | (web) The page surface temporarily blocked the account for making requests too quickly | Wait a few minutes. If it fired *before* submit, `auto` mode already fell back to codex; if *after* submit, check the conversation later — the image may still appear there. Don't retry in a loop | | `waiting for a free web/codex slot (max N concurrent …)` | More parallel runs than the backend's concurrency cap | Nothing — the run starts when a slot frees up; queue time doesn't eat `--timeout` | +## Upgrade + +When any `image-use` command prints `image-use X is available`, tell the user and offer to run +`image-use upgrade` (it updates the CLI and this skill). Check without changing anything: +`image-use upgrade --check` (or `--json`). The user may also just say "升级 image-use" / "upgrade image-use". +`image-use update` and the old `chatgpt-imagegen update` do the same thing. + +If the skill came from somewhere `upgrade` can't refresh: +- Claude Code plugin: `claude plugin update image-use@leeguooooo-plugins` +- Whole family: `curl -fsSL https://raw.githubusercontent.com/leeguooooo/plugins/main/upgrade-use-family.sh | sh` + +The notice is one stderr line, checked at most once a day; `IMAGE_USE_NO_UPDATE_CHECK=1` (or the old `CHATGPT_IMAGEGEN_NO_UPDATE_CHECK`, or the family-wide `USE_NO_UPDATE_CHECK`) turns it off. + ## Internals (for maintainers / debugging) **web backend (`run_web`)** diff --git a/image-use b/image-use index 22c6840..018e44c 100755 --- a/image-use +++ b/image-use @@ -67,13 +67,14 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Iterator -__version__ = "0.29.2" +__version__ = "0.30.0" # What's-new, newest first — one terse line per release. The update reminder # surfaces these so a user learns WHAT changed, not just that a newer version # exists. Kept as flat, easy-to-regex comments near the top so the remote copy # can be parsed from the same 8KB prefix the reminder already reads — no extra # file to drift, no import needed. Format: `# WHATSNEW[]: `. +# WHATSNEW[0.30.0]: 按 *-use 家族约定升级——`image-use upgrade`(`update` 仍可用)按安装方式装最新 GitHub Release 并刷新各处 skill,`--check`/`--json` 只查不改;每天一次的新版提示改为 stderr 一行,以 GitHub Release 为准,CI / USE_NO_UPDATE_CHECK / IMAGE_USE_NO_UPDATE_CHECK 可关闭;不再在后台自动升级 # WHATSNEW[0.29.2]: web 后端发送前核对提示词时,把编辑器等价的空白(CRLF/CR 换行、行首空格被转成的不换行空格 U+00A0)视为相同,不再因此误判「文本不一致」而拒绝发送(#39) # WHATSNEW[0.29.1]: SKILL.md 的 description 超过 1024 字符(pi 会拒收整个 skill),精简到限制以内,并加单元测试守住长度和版本号同步 # WHATSNEW[0.29.0]: 修好 web 后端的多行提示词与多参考图——提示词改为整段粘贴(旧的逐键输入会把换行当发送,拆成多条消息),发送前核对编辑器里的完整文本、等所有参考图真正传完服务端才点一次发送,结果不确定时只报告不重发;未发送就中止的运行会清掉自己粘贴的草稿,不再让下一次运行卡在「输入框非空」(#34) @@ -146,15 +147,21 @@ FALLBACK_UA = f"codex_cli_rs/{FALLBACK_VERSION} (Mac OS 26.0.1; arm64) chatgpt-i AUTH_PATH = Path.home() / ".codex" / "auth.json" VERSION_PATH = Path.home() / ".codex" / "version.json" -# Self-update check. `skills add` has no auto-update, so interactive runs check -# main once a day and ask the existing skills manager to refresh the install. -# Failure degrades to the old soft notice. It never touches stdout and is -# silenced by `--no-progress` or IMAGE_USE_NO_UPDATE_CHECK; users can keep -# the check/notice but disable installation with IMAGE_USE_NO_AUTO_UPDATE. -UPDATE_CHECK_URL = ( - "https://raw.githubusercontent.com/leeguooooo/image-use/main/image-use" -) +# Update check — the *-use family upgrade convention +# (https://github.com/leeguooooo/plugins/blob/main/docs/upgrade.md). The latest +# version is the newest non-prerelease GitHub release. Any run checks it at most +# once a day (2 s timeout, cached) and prints one stderr line when it is newer; +# nothing is installed until the user runs `image-use upgrade`. Silenced by CI, +# IMAGE_USE_NO_UPDATE_CHECK (legacy CHATGPT_IMAGEGEN_NO_UPDATE_CHECK) or the +# family-wide USE_NO_UPDATE_CHECK. +UPDATE_REPO = "leeguooooo/image-use" +UPDATE_RELEASES_URL = f"https://api.github.com/repos/{UPDATE_REPO}/releases/latest" +# The CLI file at a given git ref (a release tag), for `upgrade` and WHATSNEW. +UPDATE_RAW_URL = f"https://raw.githubusercontent.com/{UPDATE_REPO}/{{ref}}/image-use" UPDATE_CHECK_INTERVAL = 86400 # seconds; at most one network check per day +UPDATE_NOTICE_TIMEOUT = 2.0 # the daily check must not slow a run down +UPDATE_FETCH_TIMEOUT = 10.0 # an explicit `upgrade` / `doctor` can wait longer +PLUGIN_MARKETPLACE = "leeguooooo-plugins" # Environment variables. The tool was called chatgpt-imagegen until it grew # non-ChatGPT backends; every IMAGE_USE_ variable still honours its old @@ -4562,12 +4569,34 @@ def _copy_ref_into_asset(asset_dir: Path, src: str, existing: list[str]) -> str: return fname -# ---------- self-update reminder ---------- +# ---------- update check ---------- def _update_cache_path() -> Path: - base = os.environ.get("XDG_CONFIG_HOME", "").strip() - root = Path(base) if base else Path.home() / ".config" - return root / CONFIG_DIRNAME / "update-check.json" + """``${XDG_CACHE_HOME:-~/.cache}/image-use/update-check.json``.""" + base = os.environ.get("XDG_CACHE_HOME", "").strip() + root = Path(base) if base else Path.home() / ".cache" + return root / "image-use" / "update-check.json" + + +def _update_check_disabled() -> bool: + """True when CI, IMAGE_USE_NO_UPDATE_CHECK (or its legacy + CHATGPT_IMAGEGEN_ spelling) or USE_NO_UPDATE_CHECK is set.""" + return bool(os.environ.get("CI") + or _env("IMAGE_USE_NO_UPDATE_CHECK") + or os.environ.get("USE_NO_UPDATE_CHECK")) + + +def _should_check_for_update(argv: list[str]) -> bool: + """The daily check runs for every invocation except `upgrade`/`update`, + `doctor` (which does its own fresh check), `--version` and `--help`.""" + if argv and argv[0] in ("upgrade", "update", "doctor"): + return False + for arg in argv: + if arg == "--": + break + if arg in ("-h", "--help", "-V", "--version"): + return False + return True def _parse_whatsnew(text: str) -> dict: @@ -4578,23 +4607,59 @@ def _parse_whatsnew(text: str) -> dict: return out -def _fetch_latest_info(timeout: float = 4.0): - """Best-effort `(latest_version, {ver: note})` from the script on main. +def _github_open(url: str, timeout: float, accept: str): + headers = {"User-Agent": f"image-use/{__version__} update-check", + "Accept": accept} + token = os.environ.get("GITHUB_TOKEN", "").strip() + if token and url.startswith("https://api.github.com/"): + headers["Authorization"] = f"Bearer {token}" + return urllib.request.urlopen( + urllib.request.Request(url, headers=headers), timeout=timeout) + + +def _release_version(tag: Any) -> str | None: + """`v1.2.3` / `1.2.3` → `1.2.3`; anything else → None.""" + m = re.fullmatch(r"v?(\d+(?:\.\d+){1,3})", str(tag or "").strip()) + return m.group(1) if m else None - Reads only an 8KB prefix (both `__version__` and the WHATSNEW lines live at - the top), so it never pulls the whole file. Any network/parse failure returns - `(None, {})` — this is a nicety, never a gate.""" + +def _fetch_latest_release(timeout: float = UPDATE_FETCH_TIMEOUT) -> str | None: + """Version of the newest non-prerelease GitHub release, or None on any + network / API / parse failure.""" try: - req = urllib.request.Request( - UPDATE_CHECK_URL, - headers={"User-Agent": f"chatgpt-imagegen/{__version__} update-check"}, - ) - with urllib.request.urlopen(req, timeout=timeout) as r: + with _github_open(UPDATE_RELEASES_URL, timeout, + "application/vnd.github+json") as r: + data = json.loads(r.read(1 << 20).decode("utf-8")) + except Exception: # noqa: BLE001 + return None + if not isinstance(data, dict) or data.get("draft") or data.get("prerelease"): + return None + return _release_version(data.get("tag_name")) + + +def _fetch_whatsnew(version: str, timeout: float = UPDATE_FETCH_TIMEOUT) -> dict: + """WHATSNEW notes from the CLI at tag ``v`` (8KB prefix only: + `__version__` and the WHATSNEW lines sit at the top). {} on failure.""" + try: + with _github_open(UPDATE_RAW_URL.format(ref=f"v{version}"), timeout, + "text/plain") as r: head = r.read(8192).decode("utf-8", "replace") except Exception: # noqa: BLE001 + return {} + return _parse_whatsnew(head) + + +def _fetch_latest_info(timeout: float = UPDATE_FETCH_TIMEOUT): + """Best-effort `(latest_version, {ver: note})`; `(None, {})` on failure.""" + latest = _fetch_latest_release(timeout) + if not latest: return None, {} - m = re.search(r'__version__\s*=\s*"([\d.]+)"', head) - return (m.group(1) if m else None), _parse_whatsnew(head) + notes = _fetch_whatsnew(latest, timeout) if _is_newer(latest) else {} + return latest, notes + + +def _is_newer(latest: str | None, current: str = __version__) -> bool: + return bool(latest) and _version_tuple(latest) > _version_tuple(current) def _changes_since(notes: dict, base: str = __version__) -> list[tuple[str, str]]: @@ -4605,85 +4670,83 @@ def _changes_since(notes: dict, base: str = __version__) -> list[tuple[str, str] return [(v, notes[v]) for v in vers] -def _format_update_notice(latest: str, notes: dict, max_lines: int = 3) -> str: - """One header line + up to ``max_lines`` `• ver: what-changed` lines.""" - head = (f"提示:image-use {latest} 可用(当前 {__version__})。" - f"更新:image-use update") - changes = _changes_since(notes) - lines = [head] - for ver, note in changes[:max_lines]: - lines.append(f" • {ver}:{note}") - extra = len(changes) - max_lines - if extra > 0: - lines.append(f" • 另有 {extra} 项更新") - return "\n".join(lines) - - -def _installed_script_version() -> str | None: - """Version currently stored at the path this process was launched from.""" +def _write_update_cache(checked_at: float, latest: str | None) -> None: + cache = _update_cache_path() try: - head = Path(__file__).read_text(encoding="utf-8")[:8192] - except OSError: - return None - match = re.search(r'^__version__\s*=\s*"([\d.]+)"', head, re.MULTILINE) - return match.group(1) if match else None - - -def _maybe_notify_update(emit, *, auto_update: bool = False) -> None: - """Once a day, handle a newer version found on main. + cache.parent.mkdir(parents=True, exist_ok=True) + # A per-process temp name: concurrent runs must not interleave writes + # into one shared `.tmp` before the atomic rename. + fd, tmp = tempfile.mkstemp(prefix=cache.name + ".", suffix=".tmp", + dir=str(cache.parent)) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(json.dumps({"checked_at": int(checked_at), "latest": latest})) + os.replace(tmp, cache) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + except Exception: # noqa: BLE001 + pass - Interactive callers pass ``auto_update=True``: use the same skills-manager - path as the explicit ``update`` command and fall back to the ordinary notice - if the runner is missing or fails. Tests and other callers default to the - original notice-only behavior. - Caches the result (timestamp, latest version, and the what's-new notes) - under the config dir so normal runs do no network at all between daily - checks. Fully skipped when IMAGE_USE_NO_UPDATE_CHECK is set. Never - raises.""" - if _env("IMAGE_USE_NO_UPDATE_CHECK"): - return - cache = _update_cache_path() - now = time.time() +def _fetch_latest_release_bounded(timeout: float) -> str | None: + """`_fetch_latest_release` with a hard wall-clock cap. urlopen's timeout + covers socket reads but not DNS resolution, which can stall far longer on + a broken network — so the fetch runs in a daemon thread and is abandoned + (treated as a failure) once ``timeout`` seconds pass.""" + box: list = [] + t = threading.Thread(target=lambda: box.append(_fetch_latest_release(timeout)), + daemon=True) + t.start() + t.join(timeout) + return box[0] if box else None + + +def _cached_latest(now: float | None = None) -> str | None: + """Latest release version, from the cache when it is under a day old, + else from GitHub with a 2 s timeout. A failed fetch still stamps + `checked_at` (keeping the last known version) so an offline machine is + not retried on every call.""" + now = time.time() if now is None else now try: - data = json.loads(cache.read_text(encoding="utf-8")) + data = json.loads(_update_cache_path().read_text(encoding="utf-8")) last = float(data.get("checked_at", 0)) latest = data.get("latest") - notes = data.get("notes") or {} except Exception: # noqa: BLE001 - last, latest, notes = 0.0, None, {} - - if now - last >= UPDATE_CHECK_INTERVAL: - fetched_latest, fetched_notes = _fetch_latest_info() - if fetched_latest: # keep last-known if the fetch failed - latest, notes = fetched_latest, fetched_notes - try: - cache.parent.mkdir(parents=True, exist_ok=True) - cache.write_text(json.dumps( - {"checked_at": now, "latest": latest, "notes": notes}), encoding="utf-8") - except Exception: # noqa: BLE001 - pass - - if not latest or _version_tuple(latest) <= _version_tuple(__version__): + last, latest = 0.0, None + if not isinstance(latest, str) or not _release_version(latest): + latest = None + if 0 <= now - last < UPDATE_CHECK_INTERVAL: + return latest + fetched = _fetch_latest_release_bounded(UPDATE_NOTICE_TIMEOUT) + if fetched: + latest = fetched + _write_update_cache(now, latest) + return latest + + +def _update_notice(latest: str) -> str: + return (f"image-use {latest} is available (you have {__version__}). " + f"Upgrade: image-use upgrade") + + +def _maybe_notify_update(emit=None) -> None: + """Once a day, print one line to stderr when a newer release exists. + Never touches stdout, never installs anything, never raises.""" + if _update_check_disabled(): return - - if auto_update and not _env("IMAGE_USE_NO_AUTO_UPDATE"): - argv = _update_runner() - if argv: - emit(f"检测到 image-use {latest},正在自动升级…") - try: - result = subprocess.run( - argv, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - text=True, timeout=120) - except (OSError, subprocess.TimeoutExpired): - result = None - installed = _installed_script_version() - if (result is not None and result.returncode == 0 and installed - and _version_tuple(installed) >= _version_tuple(latest)): - emit(f"已自动升级到 v{latest};本次仍使用 v{__version__},下次运行生效。") - return - - emit(_format_update_notice(latest, notes)) + try: + latest = _cached_latest() + if _is_newer(latest): + if emit is None: + print(_update_notice(latest), file=sys.stderr) + else: + emit(_update_notice(latest)) + except Exception: # noqa: BLE001 + pass def _default_styles_doc() -> dict: @@ -5432,7 +5495,7 @@ def _doctor_command(argv: list[str]) -> int: latest, notes = _fetch_latest_info() # always fresh here (no throttle) if latest and _version_tuple(latest) > _version_tuple(__version__): line("warn", "version", - f"{__version__} → {latest} 可用 (image-use update)") + f"{__version__} → {latest} 可用 (image-use upgrade)") for ver, note in _changes_since(notes)[:5]: print(f" • {ver}:{note}") else: @@ -5538,7 +5601,7 @@ subcommands: image-use animate PROMPT generate a 4x2 sprite sheet and encode a loop image-use doctor check which backend is ready (and which auto picks) image-use style manage reusable style/character assets (alias: styles) - image-use update upgrade this CLI (alias: upgrade) + image-use upgrade install the latest release (--check, --json; alias: update) docs: https://github.com/leeguooooo/image-use """ @@ -5584,49 +5647,300 @@ def _installed_skill_name() -> str: return SKILL_NAME -def _self_update() -> int: - """`image-use update` / `upgrade` — refresh the installed CLI. +def _git_toplevel(path: Path) -> Path | None: + """Root of the git work tree containing ``path``, or None.""" + if not shutil.which("git"): + return None + try: + r = subprocess.run(["git", "-C", str(path), "rev-parse", "--show-toplevel"], + capture_output=True, text=True, timeout=10) + except (OSError, subprocess.TimeoutExpired): + return None + out = r.stdout.strip() + if r.returncode != 0 or not out: + return None + try: + return Path(out).resolve() + except OSError: + return None - Reserving these words as a subcommand fixes a real trap: without it, - `image-use upgrade` silently became an image prompt (drawing the word - "upgrade" and burning quota). Delivery is the `skills` manager (what the - daily update reminder points at) — run directly when it is on PATH, else - through npx; only when neither exists does this print the command instead - of running it. On success it reports the latest known version — noting that - the running process stays on the old code until the next invocation (this - .py file was just replaced underneath it). - """ - argv = _update_runner() - if argv is None: - print("`skills` 和 `npx` 都不在 PATH 上,无法自动升级。" - "装上 Node 后执行:\n" - f" npx -y skills update {_installed_skill_name()}", file=sys.stderr) - return 1 - print("升级中:" + " ".join(argv), file=sys.stderr) + +def _git_checkout_root(path: Path) -> Path | None: + """``path`` when it is itself the root of a git clone of this repo. + + Only a clone whose root IS the folder counts: a copied skill folder that + happens to sit inside some other repo (a dotfiles repo holding + ~/.claude, say) must never get that repo pulled.""" + top = _git_toplevel(path) + try: + return top if top is not None and top == path.resolve() else None + except OSError: + return None + + +def _git_is_this_repo(path: Path) -> bool: + """True when a remote of the clone at ``path`` is image-use (or its + pre-rename name chatgpt-imagegen).""" + try: + r = subprocess.run(["git", "-C", str(path), "remote", "-v"], + capture_output=True, text=True, timeout=10) + except (OSError, subprocess.TimeoutExpired): + return False + return r.returncode == 0 and bool(re.search( + r"[/:]leeguooooo/(image-use|chatgpt-imagegen)(\.git)?\s", r.stdout)) + + +def _is_plugin_path(path: Path) -> bool: + parts = path.parts + return any(a == ".claude" and b == "plugins" for a, b in zip(parts, parts[1:])) + + +def _plugin_installs() -> list[dict]: + """Claude Code plugin installs of this skill (installed_plugins.json keys + starting `image-use@`, or the pre-rename `chatgpt-imagegen@`).""" + registry = Path.home() / ".claude" / "plugins" / "installed_plugins.json" try: - rc = subprocess.run(argv).returncode + data = json.loads(registry.read_text(encoding="utf-8")) + except Exception: # noqa: BLE001 + return [] + plugins = data.get("plugins") if isinstance(data, dict) else None + if not isinstance(plugins, dict): + plugins = data if isinstance(data, dict) else {} + out = [] + for key, entries in plugins.items(): + if not isinstance(key, str) or not key.startswith( + (f"{SKILL_NAME}@", f"{LEGACY_SKILL_NAME}@")): + continue + entry = entries[0] if isinstance(entries, list) and entries else entries + path = entry.get("installPath") if isinstance(entry, dict) else None + out.append({"channel": "claude-plugin", "path": str(path or registry), + "update": f"claude plugin update {key}"}) + return out + + +def _skill_roots() -> list[Path]: + home = Path.home() + return [home / ".agents" / "skills", home / ".claude" / "skills", + home / ".codex" / "skills"] + + +def _find_skill_installs() -> list[dict]: + """Every installed copy of this skill, with how to refresh it: + `claude-plugin`, `git` (a clone) or `copy` (e.g. `npx skills add`).""" + found = _plugin_installs() + seen: set[str] = set() + for root in _skill_roots(): + for name in (SKILL_NAME, LEGACY_SKILL_NAME): + try: + real = (root / name).resolve() + if not real.is_dir(): + continue + except OSError: + continue + if str(real) in seen: + continue + seen.add(str(real)) + if _git_checkout_root(real) is not None: + found.append({"channel": "git", "path": str(real), + "update": f"git -C {real} pull --ff-only"}) + elif (real / "SKILL.md").is_file(): + found.append({"channel": "copy", "path": str(real), + "update": f"npx skills update {name}"}) + return found + + +def _cli_install_route() -> tuple[str, Path]: + """How this CLI file was installed, from where it lives: + `claude-plugin` (the plugin cache), `git` (a clone), `skills` (a skill + folder the `skills` manager copied) or `file` (a lone copied script).""" + here = Path(__file__).resolve() + folder = here.parent + if _is_plugin_path(here): + return "claude-plugin", folder + # Only a clone of THIS repo: a script sitting at the root of some other + # repo (a dotfiles-managed ~/bin, say) must never get that repo pulled. + if _git_checkout_root(folder) is not None and _git_is_this_repo(folder): + return "git", folder + if (folder / "SKILL.md").is_file(): + return "skills", folder + return "file", here + + +def _replace_script(target: Path, version: str) -> int: + """Swap the lone script at ``target`` for the one at tag ``v``.""" + url = UPDATE_RAW_URL.format(ref=f"v{version}") + try: + with _github_open(url, 30, "text/plain") as r: + data = r.read() + except Exception as e: # noqa: BLE001 + print(f"error: download failed: {url}: {e}", file=sys.stderr) + return 2 + m = re.search(rb'^__version__\s*=\s*"([\d.]+)"', data[:8192], re.MULTILINE) + if not data.startswith(b"#!") or not m or m.group(1).decode() != version: + print(f"error: {url} is not image-use {version}", file=sys.stderr) + return 2 + tmp = target.with_name(f".{target.name}.upgrade") + try: + tmp.write_bytes(data) + os.chmod(tmp, 0o755) + os.replace(tmp, target) except OSError as e: - print(f"error: 无法运行 `skills update`:{e}", file=sys.stderr) + try: + tmp.unlink() + except OSError: + pass + print(f"error: cannot replace {target}: {e}\n" + f" run: curl -fsSL {url} -o /tmp/image-use && " + f"sudo install /tmp/image-use {target}", file=sys.stderr) return 1 - if rc == 0: - latest, _ = _fetch_latest_info() - if latest and _version_tuple(latest) > _version_tuple(__version__): - print(f"完成 → v{latest}(当前进程仍是 v{__version__},下次运行即为新版)。", - file=sys.stderr) - elif latest: - print(f"已是最新(v{__version__})。", file=sys.stderr) - return rc + return 0 + + +def _run_step(argv: list[str], cwd: str | None = None) -> tuple[int, str]: + try: + r = subprocess.run(argv, cwd=cwd, capture_output=True, text=True, timeout=600) + except (OSError, subprocess.TimeoutExpired) as e: + return 1, str(e) + return r.returncode, (r.stderr or r.stdout or "").strip() + + +def _install_cli(route: str, where: Path, latest: str) -> int: + """Step 1 of `upgrade`: install ``latest`` through the route this CLI + came from. 0 ok, 2 download failed, 1 anything else.""" + if route == "claude-plugin": + return 0 # the plugin refresh in step 2 replaces this file + if route == "git": + print(f"cli: git -C {where} pull --ff-only") + rc, out = _run_step(["git", "-C", str(where), "pull", "--ff-only"]) + if rc != 0: + print(f"error: git pull failed (not forcing): {out}", file=sys.stderr) + return 1 + return 0 + if route == "skills": + argv = _update_runner() + if argv is None: + print("error: neither `skills` nor `npx` is on PATH; install Node, then run:\n" + f" npx -y skills update {_installed_skill_name()}", file=sys.stderr) + return 1 + print("cli: " + " ".join(argv)) + try: + rc = subprocess.run(argv).returncode + except OSError as e: + print(f"error: cannot run `skills update`: {e}", file=sys.stderr) + return 1 + return rc + print(f"cli: replacing {where} with the v{latest} release") + return _replace_script(where, latest) + + +def _refresh_skill(skill: dict, cli_route: str, cli_where: Path) -> int: + """Step 2 of `upgrade`: refresh one installed skill copy. 1 when a + refresh it ran failed, else 0 (printing a command is not a failure).""" + channel, path = skill["channel"], skill["path"] + label = f"skill [{channel}] {path}:" + if channel == "git" and cli_route == "git" and Path(path) == cli_where: + print(f"{label} pulled above") + elif channel == "git": + rc, out = _run_step(["git", "-C", path, "pull", "--ff-only"]) + print(f"{label} " + ("pulled" if rc == 0 + else f"not updated (git pull --ff-only failed: {out})")) + return 1 if rc else 0 + elif channel == "claude-plugin": + claude = shutil.which("claude") + if claude is None: + print(f"{label} run: {skill['update']}") + return 0 + rc, out = _run_step([claude, *skill["update"].split()[1:]]) + print(f"{label} " + ("updated" if rc == 0 + else f"`{skill['update']}` failed: {out}")) + return 1 if rc else 0 + elif channel == "copy" and cli_route == "skills": + print(f"{label} refreshed by the skills manager above") + else: + print(f"{label} run: {skill['update']}") + return 0 + + +def _upgrade_command(argv: list[str]) -> int: + """`image-use upgrade [--check] [--json]` (alias: `update`). + + Reserving these words as a subcommand also keeps `image-use upgrade` from + silently becoming an image prompt. Exit 0 on success (upgraded, already + current, or a check that ran), 2 when the check or download failed.""" + parser = argparse.ArgumentParser( + prog="image-use upgrade", + description="Install the latest image-use release and refresh the skill.") + parser.add_argument("--check", action="store_true", + help="only report current -> latest; change nothing") + parser.add_argument("--json", action="store_true", + help="like --check, as JSON") + args = parser.parse_args(argv) + + latest = _fetch_latest_release() + if latest: + _write_update_cache(time.time(), latest) + available = _is_newer(latest) + skills = _find_skill_installs() + + if args.json: + out = {"name": "image-use", "current": __version__, "latest": latest, + "update_available": available, "skills": skills} + if latest is None: + out["error"] = "could not read the latest release from GitHub" + print(json.dumps(out, ensure_ascii=False)) + return 0 if latest else 2 + if latest is None: + print(f"error: could not read the latest release from {UPDATE_RELEASES_URL}", + file=sys.stderr) + return 2 + print(f"image-use {__version__} -> {latest}" if available + else f"image-use {__version__} is up to date") + if args.check: + for skill in skills: + print(f"skill [{skill['channel']}] {skill['path']} " + f"(refresh: {skill['update']})") + return 0 + + route, where = _cli_install_route() + rc = 0 + if available: + for ver, note in _changes_since(_fetch_whatsnew(latest))[:10]: + print(f" • {ver}: {note}") + rc = _install_cli(route, where, latest) + refresh_failed = 0 + for skill in skills: + # Only a route that actually ran in step 1 can have refreshed a skill. + refresh_failed |= _refresh_skill(skill, route if available else "", where) + if rc == 0 and available: + if route == "git": + # A pull follows the checkout's branch, which may be past the tag. + print(f"pulled {where} (it follows its branch; the latest release " + f"is {latest}); the next run uses the new code") + elif route == "claude-plugin": + # `claude plugin update` installs what the marketplace lists, + # which can lag the GitHub release. + print(f"refreshed the Claude Code plugin (it installs the version the " + f"marketplace lists, which can trail release {latest}); " + f"check with `image-use --version`") + else: + print(f"upgraded to {latest} (this process is still {__version__}; " + f"the next run uses the new version)") + return rc or refresh_failed def main() -> int: + # Once-a-day "new version" notice: stderr only, so stdout (the saved path, + # JSON) stays exactly what callers parse. + if _should_check_for_update(sys.argv[1:]): + _maybe_notify_update() if len(sys.argv) > 1 and sys.argv[1] in ("style", "styles"): return _style_command(sys.argv[2:]) if len(sys.argv) > 1 and sys.argv[1] == "upload": return _upload_command(sys.argv[2:]) if len(sys.argv) > 1 and sys.argv[1] == "doctor": return _doctor_command(sys.argv[2:]) - if len(sys.argv) > 1 and sys.argv[1] in ("update", "upgrade"): - return _self_update() + if len(sys.argv) > 1 and sys.argv[1] in ("upgrade", "update"): + return _upgrade_command(sys.argv[2:]) argv = sys.argv[1:] animate_mode = bool(argv and argv[0] == "animate") if animate_mode: @@ -6011,14 +6325,6 @@ def main() -> int: f"attaching the first {REF_ATTACH_CAP} (character-first), dropped: " f"{names}"), file=sys.stderr) - # Once-a-day auto-update for interactive runs. Stderr only (stdout stays - # exactly the saved path), and skipped under --quiet/--no-progress so machine - # callers stay deterministic. `doctor` checks unconditionally for a deliberate - # look; failures here degrade to the previous one-line update notice. - if progress and not args.quiet: - _maybe_notify_update( - lambda m: print(_yellow(m), file=sys.stderr), auto_update=True) - if progress: prompt_preview = f"{args.prompt[:60]}{'…' if len(args.prompt) > 60 else ''}" mode_bit = " animation-sprite" if animate_mode else "" diff --git a/test_image_use.py b/test_image_use.py index 2e89979..6d25542 100644 --- a/test_image_use.py +++ b/test_image_use.py @@ -21,6 +21,7 @@ import re import ssl import tempfile +import threading import unittest import unittest.mock from contextlib import contextmanager @@ -94,152 +95,196 @@ def test_min_floor(self): cig._version_tuple(cig.AB_MIN_VERSION)) -class UpdateNotify(unittest.TestCase): - """The once-a-day self-update reminder: throttle, cache, env-disable, parse, - and the what's-new changelog surfaced in the notice.""" +_NO_CHECK_ENV = ("CI", "USE_NO_UPDATE_CHECK", "IMAGE_USE_NO_UPDATE_CHECK", + "CHATGPT_IMAGEGEN_NO_UPDATE_CHECK") - @contextmanager - def _patched_fetch(self, version, notes=None, counter=None): - orig = cig._fetch_latest_info - def fake(timeout=4.0): +@contextmanager +def _update_env(**extra): + """A temp XDG_CACHE_HOME with every opt-out variable cleared (CI sets `CI`), + plus ``extra``. The network is never reached: callers patch the fetch.""" + with tempfile.TemporaryDirectory() as d: + env = {k: v for k, v in os.environ.items() if k not in _NO_CHECK_ENV} + env["XDG_CACHE_HOME"] = d + env.update(extra) + with unittest.mock.patch.dict(os.environ, env, clear=True): + yield Path(d) + + +class _FakeResponse(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +class UpdateCheck(unittest.TestCase): + """The once-a-day notice: version comparison, 24 h throttle, opt-outs, + one line on stderr only — and no network (the fetch is patched).""" + + @contextmanager + def _patched_fetch(self, version, counter=None): + def fake(timeout=cig.UPDATE_FETCH_TIMEOUT): if counter is not None: - counter["n"] += 1 - return version, (notes or {}) - cig._fetch_latest_info = fake - try: + counter.append(timeout) + return version + with unittest.mock.patch.object(cig, "_fetch_latest_release", fake): yield - finally: - cig._fetch_latest_info = orig - def test_notifies_when_newer(self): - with _tmp_xdg(), self._patched_fetch("9.9.9", {"9.9.9": "shiny new thing"}): - msgs = [] - cig._maybe_notify_update(msgs.append) - self.assertTrue(msgs and "9.9.9" in msgs[0]) - - def test_notice_lists_what_changed(self): - with _tmp_xdg(), self._patched_fetch("9.9.9", {"9.9.9": "shiny new thing"}): - msgs = [] - cig._maybe_notify_update(msgs.append) - self.assertIn("shiny new thing", msgs[0]) + def test_version_comparison(self): + self.assertTrue(cig._is_newer("0.30.0", "0.29.2")) + self.assertTrue(cig._is_newer("0.29.10", "0.29.9")) + self.assertTrue(cig._is_newer("1.0.0", "0.99.99")) + self.assertFalse(cig._is_newer("0.29.2", "0.29.2")) + self.assertFalse(cig._is_newer("0.29.1", "0.29.2")) + self.assertFalse(cig._is_newer(None, "0.29.2")) + + def test_release_tag_parsing(self): + self.assertEqual(cig._release_version("v0.30.0"), "0.30.0") + self.assertEqual(cig._release_version("0.30.0"), "0.30.0") + self.assertIsNone(cig._release_version("nightly")) + self.assertIsNone(cig._release_version(None)) + + def test_fetch_reads_release_api_and_honours_github_token(self): + seen = [] + + def fake_urlopen(req, timeout): + seen.append((req.full_url, req.get_header("Authorization"), timeout)) + return _FakeResponse(json.dumps( + {"tag_name": "v9.9.9", "prerelease": False}).encode()) + + with _update_env(GITHUB_TOKEN="placeholder-token"), \ + unittest.mock.patch.object(cig.urllib.request, "urlopen", fake_urlopen): + self.assertEqual(cig._fetch_latest_release(2.0), "9.9.9") + self.assertEqual(seen, [(cig.UPDATE_RELEASES_URL, + "Bearer placeholder-token", 2.0)]) + + def test_fetch_failure_is_none(self): + def boom(*a, **k): + raise cig.urllib.error.URLError("offline") + with unittest.mock.patch.object(cig.urllib.request, "urlopen", boom): + self.assertIsNone(cig._fetch_latest_release(2.0)) + + def test_prerelease_is_ignored(self): + body = json.dumps({"tag_name": "v9.9.9", "prerelease": True}).encode() + with unittest.mock.patch.object(cig.urllib.request, "urlopen", + lambda *a, **k: _FakeResponse(body)): + self.assertIsNone(cig._fetch_latest_release(2.0)) + + def test_notice_is_one_line_on_stderr_only(self): + with _update_env(), self._patched_fetch("9.9.9"): + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + cig._maybe_notify_update() + self.assertEqual(out.getvalue(), "") + self.assertEqual( + err.getvalue(), + f"image-use 9.9.9 is available (you have {cig.__version__}). " + "Upgrade: image-use upgrade\n") def test_silent_when_same_or_older(self): - with _tmp_xdg(), self._patched_fetch(cig.__version__): + with _update_env(), self._patched_fetch(cig.__version__): msgs = [] cig._maybe_notify_update(msgs.append) - self.assertEqual(msgs, []) - - def test_throttled_no_network_within_interval(self): - with _tmp_xdg(): - counter = {"n": 0} - with self._patched_fetch("9.9.9", {"9.9.9": "x"}, counter): - cig._maybe_notify_update(lambda _m: None) # first: hits network - cig._maybe_notify_update(lambda _m: None) # second: cached - self.assertEqual(counter["n"], 1) + self.assertEqual(msgs, []) + + def test_uses_2s_timeout_and_writes_cache(self): + with _update_env() as cache_home: + calls = [] + with self._patched_fetch("9.9.9", calls): + cig._maybe_notify_update(lambda _m: None) + self.assertEqual(calls, [2.0]) + data = json.loads( + (cache_home / "image-use" / "update-check.json").read_text()) + self.assertEqual(data["latest"], "9.9.9") + self.assertAlmostEqual(data["checked_at"], time.time(), delta=60) + + def test_throttled_within_24h(self): + with _update_env(): + calls = [] + with self._patched_fetch("9.9.9", calls): + cig._maybe_notify_update(lambda _m: None) # hits the network + msgs = [] + cig._maybe_notify_update(msgs.append) # served from cache + self.assertEqual(len(calls), 1) + self.assertTrue(msgs and "9.9.9" in msgs[0]) - def test_uses_cached_latest_when_throttled(self): - with _tmp_xdg(): - with self._patched_fetch("9.9.9", {"9.9.9": "cached note"}): - cig._maybe_notify_update(lambda _m: None) # populate cache - # Network would now report an older version, but throttle keeps cached 9.9.9. - with self._patched_fetch("0.0.1", {"0.0.1": "stale"}): + def test_rechecks_after_24h(self): + with _update_env() as cache_home: + cache = cache_home / "image-use" / "update-check.json" + cache.parent.mkdir(parents=True) + cache.write_text(json.dumps( + {"checked_at": time.time() - 86400 - 5, "latest": "0.0.1"})) + calls = [] + with self._patched_fetch("9.9.9", calls): msgs = [] cig._maybe_notify_update(msgs.append) - self.assertTrue(msgs and "9.9.9" in msgs[0] and "cached note" in msgs[0]) + self.assertEqual(len(calls), 1) + self.assertTrue(msgs and "9.9.9" in msgs[0]) - def test_env_disable_is_noop(self): - with _tmp_xdg(): - counter = {"n": 0} - os.environ["CHATGPT_IMAGEGEN_NO_UPDATE_CHECK"] = "1" - try: - with self._patched_fetch("9.9.9", {"9.9.9": "x"}, counter): - msgs = [] + def test_failed_check_is_silent_and_still_stamps_checked_at(self): + with _update_env() as cache_home: + calls = [] + with self._patched_fetch(None, calls): + msgs = [] + cig._maybe_notify_update(msgs.append) + cig._maybe_notify_update(msgs.append) # not retried + data = json.loads( + (cache_home / "image-use" / "update-check.json").read_text()) + self.assertEqual((len(calls), msgs), (1, [])) + self.assertIsNone(data["latest"]) + + def test_failed_check_keeps_last_known_version(self): + with _update_env() as cache_home: + cache = cache_home / "image-use" / "update-check.json" + cache.parent.mkdir(parents=True) + cache.write_text(json.dumps({"checked_at": 0, "latest": "9.9.9"})) + with self._patched_fetch(None): + msgs = [] + cig._maybe_notify_update(msgs.append) + self.assertTrue(msgs and "9.9.9" in msgs[0]) + + def test_opt_out_env_vars(self): + for var in ("CI", "USE_NO_UPDATE_CHECK", "IMAGE_USE_NO_UPDATE_CHECK", + "CHATGPT_IMAGEGEN_NO_UPDATE_CHECK"): + with self.subTest(var=var), _update_env(**{var: "1"}) as cache_home: + calls, msgs = [], [] + with self._patched_fetch("9.9.9", calls): cig._maybe_notify_update(msgs.append) - finally: - os.environ.pop("CHATGPT_IMAGEGEN_NO_UPDATE_CHECK", None) - self.assertEqual((counter["n"], msgs), (0, [])) - - def test_interactive_run_auto_updates_newer_version(self): - class _Res: - returncode = 0 - - calls = [] - - def fake_run(argv, *args, **kwargs): - calls.append((argv, kwargs)) - return _Res() - - with _tmp_xdg(), \ - self._patched_fetch("9.9.9", {"9.9.9": "shiny new thing"}), \ - unittest.mock.patch.object(cig, "_update_runner", - return_value=["skills", "update", "image-use"]), \ - unittest.mock.patch.object(cig, "_installed_script_version", - return_value="9.9.9"), \ - unittest.mock.patch.object(cig.subprocess, "run", fake_run): - msgs = [] - cig._maybe_notify_update(msgs.append, auto_update=True) - - self.assertEqual(calls[0][0], ["skills", "update", "image-use"]) - self.assertIs(calls[0][1]["stdout"], cig.subprocess.DEVNULL) - self.assertIn("正在自动升级", msgs[0]) - self.assertIn("已自动升级到 v9.9.9", msgs[1]) - self.assertNotIn("更新:image-use update", "\n".join(msgs)) - - def test_auto_update_failure_falls_back_to_notice(self): - class _Res: - returncode = 1 - - with _tmp_xdg(), self._patched_fetch("9.9.9", {"9.9.9": "change"}), \ - unittest.mock.patch.object(cig, "_update_runner", - return_value=["skills", "update", "image-use"]), \ - unittest.mock.patch.object(cig.subprocess, "run", return_value=_Res()): - msgs = [] - cig._maybe_notify_update(msgs.append, auto_update=True) - - self.assertIn("正在自动升级", msgs[0]) - self.assertIn("更新:image-use update", msgs[-1]) - - def test_update_that_does_not_replace_this_cli_falls_back_to_notice(self): - class _Res: - returncode = 0 - - with _tmp_xdg(), self._patched_fetch("9.9.9", {"9.9.9": "change"}), \ - unittest.mock.patch.object(cig, "_update_runner", - return_value=["skills", "update", "image-use"]), \ - unittest.mock.patch.object(cig, "_installed_script_version", + self.assertEqual((calls, msgs), ([], [])) + self.assertFalse((cache_home / "image-use").exists()) + + def test_skipped_for_upgrade_version_and_help(self): + for argv in (["upgrade"], ["upgrade", "--json"], ["update"], ["doctor"], + ["--version"], ["-V"], ["--help"], ["-h"], + ["a cat", "--help"], ["style", "list", "-h"]): + with self.subTest(argv=argv): + self.assertFalse(cig._should_check_for_update(argv)) + for argv in ([], ["a cat"], ["style", "list"], ["a cat", "--", "--help"]): + with self.subTest(argv=argv): + self.assertTrue(cig._should_check_for_update(argv)) + + def test_main_upgrade_check_does_not_run_notice(self): + with _update_env(), \ + unittest.mock.patch.object(cig.sys, "argv", + ["image-use", "upgrade", "--check"]), \ + unittest.mock.patch.object(cig, "_maybe_notify_update") as notify, \ + unittest.mock.patch.object(cig, "_fetch_latest_release", return_value=cig.__version__), \ - unittest.mock.patch.object(cig.subprocess, "run", return_value=_Res()): - msgs = [] - cig._maybe_notify_update(msgs.append, auto_update=True) - - self.assertIn("更新:image-use update", msgs[-1]) - - def test_no_auto_update_keeps_notice_without_running(self): - with _tmp_xdg(), self._patched_fetch("9.9.9", {"9.9.9": "change"}), \ - unittest.mock.patch.dict( - os.environ, {"CHATGPT_IMAGEGEN_NO_AUTO_UPDATE": "1"}), \ - unittest.mock.patch.object(cig, "_update_runner") as runner: - msgs = [] - cig._maybe_notify_update(msgs.append, auto_update=True) - - runner.assert_not_called() - self.assertIn("更新:image-use update", msgs[0]) + unittest.mock.patch.object(cig, "_find_skill_installs", return_value=[]): + with redirect_stdout(io.StringIO()): + self.assertEqual(cig.main(), 0) + notify.assert_not_called() def test_changes_since_filters_and_orders(self): notes = {"0.1.0": "old", "9.9.0": "mid", "9.9.9": "new"} self.assertEqual(cig._changes_since(notes, base="9.8.0"), [("9.9.9", "new"), ("9.9.0", "mid")]) - def test_format_notice_caps_lines(self): - notes = {f"9.0.{i}": f"change {i}" for i in range(1, 6)} - out = cig._format_update_notice("9.0.5", notes, max_lines=3) - self.assertEqual(out.count("\n •"), 4) # 3 changes + "另有 N 项" - self.assertIn("另有 2 项", out) - def test_parse_whatsnew_from_real_header(self): - # Both __version__ and the newest WHATSNEW line must sit in the first 8KB, - # since the reminder only reads that prefix of the remote script. + # `upgrade` reads what changed from the first 8KB of the CLI at the + # release tag, so __version__ and its WHATSNEW line must sit there. head = Path(os.path.join(os.path.dirname(__file__), "image-use")).read_text(encoding="utf-8")[:8192] m = re.search(r'__version__\s*=\s*"([\d.]+)"', head) @@ -249,92 +294,307 @@ def test_parse_whatsnew_from_real_header(self): self.assertTrue(notes[cig.__version__]) -class SelfUpdate(unittest.TestCase): - """`update` / `upgrade` shells out to `skills update` instead of drawing.""" - - def test_runs_skills_update(self): - calls = [] +class Upgrade(unittest.TestCase): + """`upgrade` / `update`: --check, --json, exit codes, install routes and + skill refresh — all against a temp HOME with the network patched.""" - class _Res: - returncode = 0 - - def fake_run(argv, *a, **k): - calls.append(argv) - return _Res() + @contextmanager + def _home(self): + with tempfile.TemporaryDirectory() as home, _update_env(): + with unittest.mock.patch.object(cig.Path, "home", + return_value=Path(home)): + yield Path(home) + + def _run(self, argv, latest): + out, err = io.StringIO(), io.StringIO() + with unittest.mock.patch.object(cig, "_fetch_latest_release", + return_value=latest), \ + redirect_stdout(out), redirect_stderr(err): + rc = cig._upgrade_command(argv) + return rc, out.getvalue(), err.getvalue() + + def test_check_update_available(self): + with self._home(): + rc, out, _ = self._run(["--check"], "9.9.9") + self.assertEqual(rc, 0) + self.assertEqual(out.splitlines()[0], f"image-use {cig.__version__} -> 9.9.9") - with unittest.mock.patch.object(cig.shutil, "which", - return_value="/usr/bin/skills"), \ - unittest.mock.patch.object(cig.subprocess, "run", fake_run), \ - unittest.mock.patch.object(cig, "_fetch_latest_info", - return_value=(None, {})): - rc = cig._self_update() + def test_check_up_to_date(self): + with self._home(): + rc, out, _ = self._run(["--check"], cig.__version__) + self.assertEqual(rc, 0) + self.assertEqual(out.strip(), f"image-use {cig.__version__} is up to date") + + def test_check_failure_exits_2(self): + with self._home(): + rc, out, err = self._run(["--check"], None) + self.assertEqual((rc, out), (2, "")) + self.assertIn("could not read the latest release", err) + + def test_json_shape(self): + with self._home() as home: + plugins = home / ".claude" / "plugins" + plugins.mkdir(parents=True) + (plugins / "installed_plugins.json").write_text(json.dumps( + {"version": 2, "plugins": { + "image-use@leeguooooo-plugins": [ + {"installPath": str(home / "cache" / "image-use")}], + "other@leeguooooo-plugins": [{"installPath": "/x"}]}})) + copy = home / ".agents" / "skills" / "image-use" + copy.mkdir(parents=True) + (copy / "SKILL.md").write_text("---\nname: image-use\n---\n") + (home / ".claude" / "skills").mkdir(parents=True) + (home / ".claude" / "skills" / "image-use").symlink_to(copy) + with unittest.mock.patch.object(cig, "_git_toplevel", return_value=None): + rc, out, _ = self._run(["--json"], "9.9.9") self.assertEqual(rc, 0) - self.assertEqual(calls, - [["/usr/bin/skills", "update", "image-use"]]) + data = json.loads(out) + self.assertEqual(data, { + "name": "image-use", "current": cig.__version__, "latest": "9.9.9", + "update_available": True, + "skills": [ + {"channel": "claude-plugin", + "path": str(home / "cache" / "image-use"), + "update": "claude plugin update image-use@leeguooooo-plugins"}, + {"channel": "copy", "path": str(copy.resolve()), + "update": "npx skills update image-use"}, + ]}) + + def test_json_failure_exits_2_with_valid_json(self): + with self._home(): + rc, out, _ = self._run(["--json"], None) + data = json.loads(out) + self.assertEqual(rc, 2) + self.assertIsNone(data["latest"]) + self.assertFalse(data["update_available"]) + + def test_git_checkout_skill_is_detected_only_at_its_root(self): + with self._home() as home: + clone = home / "src" / "image-use" + clone.mkdir(parents=True) + (clone / "SKILL.md").write_text("x") + (home / ".agents" / "skills").mkdir(parents=True) + (home / ".agents" / "skills" / "image-use").symlink_to(clone) + with unittest.mock.patch.object(cig, "_git_toplevel", + return_value=clone.resolve()): + found = cig._find_skill_installs() + # A copy that merely sits inside some other repo is not a clone. + with unittest.mock.patch.object(cig, "_git_toplevel", + return_value=home.resolve()): + found_nested = cig._find_skill_installs() + self.assertEqual(found, [{"channel": "git", "path": str(clone.resolve()), + "update": f"git -C {clone.resolve()} pull --ff-only"}]) + self.assertEqual(found_nested[0]["channel"], "copy") + + def test_upgrade_runs_route_then_refreshes_skills(self): + skills = [{"channel": "git", "path": "/s/image-use", + "update": "git -C /s/image-use pull --ff-only"}, + {"channel": "claude-plugin", "path": "/p", + "update": "claude plugin update image-use@leeguooooo-plugins"}, + {"channel": "copy", "path": "/c", + "update": "npx skills update image-use"}] + steps = [] + + def fake_step(argv, cwd=None): + steps.append(argv) + return 0, "" + + with self._home(), \ + unittest.mock.patch.object(cig, "_find_skill_installs", return_value=skills), \ + unittest.mock.patch.object(cig, "_cli_install_route", + return_value=("file", Path("/bin/image-use"))), \ + unittest.mock.patch.object(cig, "_replace_script", return_value=0) as repl, \ + unittest.mock.patch.object(cig, "_fetch_whatsnew", + return_value={"9.9.9": "new thing"}), \ + unittest.mock.patch.object(cig, "_run_step", fake_step), \ + unittest.mock.patch.object(cig.shutil, "which", + lambda n: f"/usr/bin/{n}"): + rc, out, _ = self._run([], "9.9.9") + self.assertEqual(rc, 0) + repl.assert_called_once_with(Path("/bin/image-use"), "9.9.9") + self.assertEqual(steps, [ + ["git", "-C", "/s/image-use", "pull", "--ff-only"], + ["/usr/bin/claude", "plugin", "update", "image-use@leeguooooo-plugins"]]) + self.assertIn("9.9.9: new thing", out) + self.assertIn("run: npx skills update image-use", out) # printed, not run + self.assertIn("upgraded to 9.9.9", out) + + def test_plugin_refresh_prints_command_without_claude(self): + skill = {"channel": "claude-plugin", "path": "/p", + "update": "claude plugin update image-use@leeguooooo-plugins"} + with unittest.mock.patch.object(cig.shutil, "which", return_value=None), \ + unittest.mock.patch.object(cig, "_run_step") as step: + out = io.StringIO() + with redirect_stdout(out): + cig._refresh_skill(skill, "file", Path("/x")) + step.assert_not_called() + self.assertIn("run: claude plugin update image-use@leeguooooo-plugins", + out.getvalue()) + + def test_git_pull_failure_is_reported_not_forced(self): + skill = {"channel": "git", "path": "/s", "update": "git -C /s pull --ff-only"} + with unittest.mock.patch.object(cig, "_run_step", + return_value=(1, "Not possible to fast-forward")): + out = io.StringIO() + with redirect_stdout(out): + rc = cig._refresh_skill(skill, "file", Path("/x")) + self.assertEqual(rc, 1) + self.assertIn("not updated", out.getvalue()) + self.assertIn("Not possible to fast-forward", out.getvalue()) + + def test_up_to_date_still_pulls_the_clone(self): + # With nothing installed in step 1, a clone must not be reported as + # "pulled above" — it gets its own pull. + skill = {"channel": "git", "path": "/s", "update": "git -C /s pull --ff-only"} + with self._home(), \ + unittest.mock.patch.object(cig, "_find_skill_installs", return_value=[skill]), \ + unittest.mock.patch.object(cig, "_cli_install_route", + return_value=("git", Path("/s"))), \ + unittest.mock.patch.object(cig, "_run_step", return_value=(0, "")) as step: + rc, out, _ = self._run([], cig.__version__) + self.assertEqual(rc, 0) + step.assert_called_once_with(["git", "-C", "/s", "pull", "--ff-only"]) + self.assertIn("pulled", out) + + def test_up_to_date_installs_nothing(self): + with self._home(), \ + unittest.mock.patch.object(cig, "_find_skill_installs", return_value=[]), \ + unittest.mock.patch.object(cig, "_install_cli") as install: + rc, out, _ = self._run([], cig.__version__) + self.assertEqual(rc, 0) + install.assert_not_called() + self.assertIn("is up to date", out) - def test_missing_skills_falls_back_to_npx(self): - """`skills` is usually only reachable via npx — that must not be a dead end.""" - calls = [] + def test_replace_script_swaps_file_for_release(self): + body = (b'#!/usr/bin/env python3\n__version__ = "9.9.9"\n') + with tempfile.TemporaryDirectory() as d: + target = Path(d) / "image-use" + target.write_text("old") + with unittest.mock.patch.object(cig.urllib.request, "urlopen", + lambda *a, **k: _FakeResponse(body)): + rc = cig._replace_script(target, "9.9.9") + self.assertEqual(rc, 0) + self.assertEqual(target.read_bytes(), body) + self.assertTrue(os.access(target, os.X_OK)) - class _Res: - returncode = 0 + def test_replace_script_rejects_wrong_payload(self): + with tempfile.TemporaryDirectory() as d: + target = Path(d) / "image-use" + target.write_text("old") + with unittest.mock.patch.object( + cig.urllib.request, "urlopen", + lambda *a, **k: _FakeResponse(b"not found")), \ + redirect_stderr(io.StringIO()): + rc = cig._replace_script(target, "9.9.9") + self.assertEqual(rc, 2) + self.assertEqual(target.read_text(), "old") + + def test_route_detection(self): + with tempfile.TemporaryDirectory() as d: + plugin = Path(d) / ".claude" / "plugins" / "cache" / "x" / "image-use" + skill = Path(d) / "skills" / "image-use" + loose = Path(d) / "bin" + for folder in (plugin, skill, loose): + folder.mkdir(parents=True) + (skill / "SKILL.md").write_text("x") + with unittest.mock.patch.object(cig, "_git_toplevel", return_value=None): + for folder, route in ((plugin, "claude-plugin"), (skill, "skills"), + (loose, "file")): + with unittest.mock.patch.object(cig, "__file__", + str(folder / "image-use")): + self.assertEqual(cig._cli_install_route()[0], route) + with unittest.mock.patch.object(cig, "_git_toplevel", + return_value=skill.resolve()), \ + unittest.mock.patch.object(cig, "_git_is_this_repo", return_value=True), \ + unittest.mock.patch.object(cig, "__file__", str(skill / "image-use")): + self.assertEqual(cig._cli_install_route()[0], "git") + # Some other repo that happens to hold SKILL.md + the script: + # keep the skills route, never pull it. + with unittest.mock.patch.object(cig, "_git_toplevel", + return_value=skill.resolve()), \ + unittest.mock.patch.object(cig, "_git_is_this_repo", return_value=False), \ + unittest.mock.patch.object(cig, "__file__", str(skill / "image-use")): + self.assertEqual(cig._cli_install_route()[0], "skills") + + def test_lone_script_at_root_of_unrelated_repo_is_a_file_route(self): + # e.g. ~/bin is a dotfiles repo: never `git pull` it. + with tempfile.TemporaryDirectory() as d: + loose = Path(d) / "bin" + loose.mkdir() + with unittest.mock.patch.object(cig, "_git_toplevel", + return_value=loose.resolve()), \ + unittest.mock.patch.object(cig, "_git_is_this_repo", return_value=False), \ + unittest.mock.patch.object(cig, "__file__", str(loose / "image-use")): + self.assertEqual(cig._cli_install_route()[0], "file") + + def test_daily_fetch_is_wall_clock_bounded(self): + release = threading.Event() + + def hang(timeout=cig.UPDATE_FETCH_TIMEOUT): + release.wait(5) # e.g. a stalled DNS lookup urlopen can't time out + return "9.9.9" + try: + with unittest.mock.patch.object(cig, "_fetch_latest_release", hang): + t0 = time.monotonic() + self.assertIsNone(cig._fetch_latest_release_bounded(0.1)) + self.assertLess(time.monotonic() - t0, 2) + finally: + release.set() - def fake_which(name): - return "/usr/bin/npx" if name == "npx" else None + def test_cache_write_leaves_no_temp_files(self): + with tempfile.TemporaryDirectory() as d, \ + unittest.mock.patch.dict(os.environ, {"XDG_CACHE_HOME": d}): + cig._write_update_cache(1.0, "1.2.3") + cig._write_update_cache(2.0, "1.2.4") + folder = Path(d) / "image-use" + self.assertEqual(sorted(p.name for p in folder.iterdir()), + ["update-check.json"]) + self.assertEqual(json.loads((folder / "update-check.json").read_text()), + {"checked_at": 2, "latest": "1.2.4"}) - def fake_run(argv, *a, **k): - calls.append(argv) - return _Res() - with unittest.mock.patch.object(cig.shutil, "which", fake_which), \ - unittest.mock.patch.object(cig.subprocess, "run", fake_run), \ - unittest.mock.patch.object(cig, "_fetch_latest_info", - return_value=(None, {})): - rc = cig._self_update() - self.assertEqual(rc, 0) - self.assertEqual( - calls, - [["/usr/bin/npx", "-y", "skills", "update", "image-use"]]) +class SkillsRoute(unittest.TestCase): + """A skill-folder install upgrades through `skills update`, via npx when + `skills` itself is not on PATH.""" - def test_path_skills_wins_over_npx(self): - """A real `skills` on PATH is cheaper than spinning up npx.""" + def _install(self, which, run_rc=0): calls = [] class _Res: - returncode = 0 + returncode = run_rc def fake_run(argv, *a, **k): calls.append(argv) return _Res() - with unittest.mock.patch.object(cig.shutil, "which", - lambda name: f"/usr/bin/{name}"), \ + with unittest.mock.patch.object(cig.shutil, "which", which), \ unittest.mock.patch.object(cig.subprocess, "run", fake_run), \ - unittest.mock.patch.object(cig, "_fetch_latest_info", - return_value=(None, {})): - cig._self_update() - self.assertEqual(calls, - [["/usr/bin/skills", "update", "image-use"]]) + redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()) as err: + rc = cig._install_cli("skills", Path("/s"), "9.9.9") + return rc, calls, err.getvalue() + + def test_runs_skills_update(self): + rc, calls, _ = self._install(lambda n: "/usr/bin/skills") + self.assertEqual((rc, calls), (0, [["/usr/bin/skills", "update", "image-use"]])) + + def test_missing_skills_falls_back_to_npx(self): + rc, calls, _ = self._install( + lambda n: "/usr/bin/npx" if n == "npx" else None) + self.assertEqual( + (rc, calls), (0, [["/usr/bin/npx", "-y", "skills", "update", "image-use"]])) + + def test_path_skills_wins_over_npx(self): + _, calls, _ = self._install(lambda n: f"/usr/bin/{n}") + self.assertEqual(calls, [["/usr/bin/skills", "update", "image-use"]]) def test_no_runner_at_all_prints_command_and_fails(self): - with unittest.mock.patch.object(cig.shutil, "which", return_value=None): - buf = io.StringIO() - with redirect_stderr(buf): - rc = cig._self_update() - self.assertEqual(rc, 1) - self.assertIn("npx -y skills update image-use", buf.getvalue()) + rc, calls, err = self._install(lambda n: None) + self.assertEqual((rc, calls), (1, [])) + self.assertIn("npx -y skills update image-use", err) def test_propagates_nonzero_exit(self): - class _Res: - returncode = 3 - - with unittest.mock.patch.object(cig.shutil, "which", - return_value="/usr/bin/skills"), \ - unittest.mock.patch.object(cig.subprocess, "run", - return_value=_Res()), \ - unittest.mock.patch.object(cig, "_fetch_latest_info", - return_value=(None, {})): - self.assertEqual(cig._self_update(), 3) + rc, _, _ = self._install(lambda n: "/usr/bin/skills", run_rc=3) + self.assertEqual(rc, 3) class IsUrl(unittest.TestCase):