feat: add self-update and improve skill selector
This commit is contained in:
@@ -20,6 +20,15 @@
|
||||
# ref: main
|
||||
# path: skills
|
||||
|
||||
skills:
|
||||
repo: ~/.skills
|
||||
ref: main
|
||||
path: skills
|
||||
description: 本地自研 Agent Skills 集合,由 skiff 统一创建、维护和安装。
|
||||
tags:
|
||||
- skill-management
|
||||
- owned
|
||||
|
||||
waza:
|
||||
repo: https://github.com/tw93/Waza.git
|
||||
ref: main
|
||||
|
||||
+5
-1
@@ -65,6 +65,7 @@ skiff bootstrap
|
||||
| 命令 | 说明 |
|
||||
|------|------|
|
||||
| `skiff bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent |
|
||||
| `skiff update` | 在 `~/.skills` 执行 `git pull`,更新 skiff 自身 |
|
||||
| `skiff kit init <name> [--project DIR] [--copy]` | 在项目的 `docs/<name>/` 初始化规范包;默认软链接到 SSOT |
|
||||
|
||||
### 全局安装(自研 skill)
|
||||
@@ -113,7 +114,10 @@ skiff select --project ~/code/app # 指定项目
|
||||
```
|
||||
|
||||
使用方向键移动、空格勾选、`/` 搜索、Enter 安装,按 `q` 或 Esc
|
||||
取消。已经安装到目标范围的 skill 默认勾选;取消勾选不会卸载已有 skill。
|
||||
取消。普通 `skiff select` 只安装到项目,并在每一项旁只读显示各 Agent 的
|
||||
全局安装状态;`skiff select -g` 只安装到全局。已经安装到目标范围的 skill
|
||||
默认勾选;取消勾选不会卸载已有 skill,卸载请使用 `skiff remove`。如果全局
|
||||
存在同名但指向其它来源的 skill,项目选择器会显示“全局同名冲突”。
|
||||
项目模式会把成功选择的项目写入 `.skills.yaml`。非交互环境请使用
|
||||
`skiff add <name>...`。使用 `-a` 限定 Agent 时,该范围会记录在对应的
|
||||
skill 条目中,后续 `skiff sync` 不会扩散到其他 Agent。
|
||||
|
||||
+82
-24
@@ -38,6 +38,7 @@ from skiff.registry import (
|
||||
external_checkout_path,
|
||||
external_skill_path,
|
||||
load_registry,
|
||||
registry_repo,
|
||||
save_registry,
|
||||
)
|
||||
from skiff.selector import SkillChoice, select_skills
|
||||
@@ -139,7 +140,7 @@ def _ensure_registry_fetched(name: str, entry: dict[str, object]) -> None:
|
||||
if (path / "SKILL.md").is_file() or discover_external_skills(name, entry):
|
||||
return
|
||||
|
||||
repo = str(entry["repo"])
|
||||
repo = registry_repo(entry)
|
||||
ref = entry.get("ref", "main")
|
||||
dest = external_checkout_path(name, entry)
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
@@ -286,6 +287,29 @@ def _is_fully_installed(
|
||||
)
|
||||
|
||||
|
||||
def _global_installation_note(
|
||||
name: str,
|
||||
expected: Path,
|
||||
targets: list[str],
|
||||
) -> str:
|
||||
installed: list[str] = []
|
||||
conflicts: list[str] = []
|
||||
for target in targets:
|
||||
link = agent_skill_dir(target) / name
|
||||
status = check_link(link, expected)
|
||||
if status.ok:
|
||||
installed.append(target)
|
||||
elif link.exists() or link.is_symlink():
|
||||
conflicts.append(target)
|
||||
|
||||
parts: list[str] = []
|
||||
if installed:
|
||||
parts.append(f"全局: {','.join(installed)}")
|
||||
if conflicts:
|
||||
parts.append(f"全局同名冲突: {','.join(conflicts)}")
|
||||
return ";".join(parts)
|
||||
|
||||
|
||||
def _remove_skill(
|
||||
name: str,
|
||||
targets: list[str],
|
||||
@@ -346,6 +370,13 @@ def cmd_bootstrap(args: argparse.Namespace) -> None:
|
||||
_print("已安装项目 skill 到所有 agent")
|
||||
|
||||
|
||||
def cmd_update(args: argparse.Namespace) -> None:
|
||||
del args
|
||||
ensure_skills_home()
|
||||
_print(f"更新 skiff: {SKILLS_HOME}")
|
||||
subprocess.run(["git", "-C", str(SKILLS_HOME), "pull"], check=True)
|
||||
|
||||
|
||||
def _installed_links(
|
||||
name: str,
|
||||
targets: list[str],
|
||||
@@ -525,17 +556,38 @@ def cmd_select(args: argparse.Namespace) -> None:
|
||||
for name in [*owned_names, *registry]:
|
||||
validate_skill_name(name)
|
||||
|
||||
choices = [
|
||||
SkillChoice(
|
||||
def make_choice(
|
||||
*,
|
||||
name: str,
|
||||
installed_name: str,
|
||||
expected: Path,
|
||||
kind: str,
|
||||
description: str,
|
||||
) -> SkillChoice:
|
||||
return SkillChoice(
|
||||
name=name,
|
||||
kind="owned",
|
||||
description=skill_description(name) or "",
|
||||
kind=kind,
|
||||
description=description,
|
||||
installed=_is_fully_installed(
|
||||
name,
|
||||
SKILLS_DIR / name,
|
||||
installed_name,
|
||||
expected,
|
||||
project_root,
|
||||
targets,
|
||||
),
|
||||
readonly_status=(
|
||||
_global_installation_note(installed_name, expected, targets)
|
||||
if project_root is not None
|
||||
else ""
|
||||
),
|
||||
)
|
||||
|
||||
choices = [
|
||||
make_choice(
|
||||
name=name,
|
||||
installed_name=name,
|
||||
expected=SKILLS_DIR / name,
|
||||
kind="owned",
|
||||
description=skill_description(name) or "",
|
||||
)
|
||||
for name in owned_names
|
||||
]
|
||||
@@ -550,40 +602,43 @@ def cmd_select(args: argparse.Namespace) -> None:
|
||||
_err(f"警告: registry 条目与 owned skill 同名,已忽略 external: {package}")
|
||||
continue
|
||||
choices.append(
|
||||
SkillChoice(
|
||||
make_choice(
|
||||
name=package,
|
||||
installed_name=package,
|
||||
expected=root,
|
||||
kind="external",
|
||||
description=str(entry.get("description", "")),
|
||||
installed=_is_fully_installed(
|
||||
package,
|
||||
root,
|
||||
project_root,
|
||||
targets,
|
||||
),
|
||||
)
|
||||
)
|
||||
choice_requests[package] = (package, "registry")
|
||||
continue
|
||||
for skill_name in skill_names:
|
||||
expected = root / skill_name
|
||||
if (
|
||||
skill_name in owned_names
|
||||
and expected.resolve() == (SKILLS_DIR / skill_name).resolve()
|
||||
):
|
||||
continue
|
||||
choice_name = f"{package}/{skill_name}"
|
||||
description = read_skill_meta(root / skill_name).get("description", "")
|
||||
description = read_skill_meta(expected).get("description", "")
|
||||
choices.append(
|
||||
SkillChoice(
|
||||
make_choice(
|
||||
name=choice_name,
|
||||
installed_name=skill_name,
|
||||
expected=expected,
|
||||
kind=f"external:{package}",
|
||||
description=description,
|
||||
installed=_is_fully_installed(
|
||||
skill_name,
|
||||
root / skill_name,
|
||||
project_root,
|
||||
targets,
|
||||
),
|
||||
)
|
||||
)
|
||||
choice_requests[choice_name] = (skill_name, f"registry:{package}")
|
||||
|
||||
try:
|
||||
selected = select_skills(choices)
|
||||
scope_label = (
|
||||
"全局"
|
||||
if project_root is None
|
||||
else f"项目 {project_root}(全局状态只读)"
|
||||
)
|
||||
selected = select_skills(choices, scope_label=scope_label)
|
||||
except (RuntimeError, OSError) as exc:
|
||||
raise SystemExit(str(exc)) from exc
|
||||
if selected is None:
|
||||
@@ -738,7 +793,7 @@ def cmd_fetch(args: argparse.Namespace) -> None:
|
||||
raise SystemExit(f"registry 中不存在: {args.name}")
|
||||
|
||||
entry = registry[args.name]
|
||||
repo = entry["repo"]
|
||||
repo = registry_repo(entry)
|
||||
ref = entry.get("ref", "main")
|
||||
dest = external_checkout_path(args.name, entry)
|
||||
|
||||
@@ -1149,6 +1204,9 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
)
|
||||
p_bootstrap.set_defaults(func=cmd_bootstrap)
|
||||
|
||||
p_update = sub.add_parser("update", help="通过 git pull 更新 skiff 自身")
|
||||
p_update.set_defaults(func=cmd_update)
|
||||
|
||||
p_list = sub.add_parser("list", help="列出所有 source 中的 skill")
|
||||
p_list.add_argument("--source", help="只列出指定来源(owned、registry 或 custom source)")
|
||||
p_list.set_defaults(func=cmd_list)
|
||||
|
||||
+12
-1
@@ -26,6 +26,12 @@ def save_registry(data: dict[str, dict[str, Any]], path: Path | None = None) ->
|
||||
path.write_text(yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8")
|
||||
|
||||
|
||||
def registry_repo(entry: dict[str, Any]) -> str:
|
||||
"""Expand a local home-relative repo while leaving remote URLs unchanged."""
|
||||
repo = str(entry.get("repo", ""))
|
||||
return str(Path(repo).expanduser()) if repo.startswith("~") else repo
|
||||
|
||||
|
||||
def external_repo_path(entry: dict[str, Any]) -> Path:
|
||||
"""Return the shared checkout path for a repo/ref pair."""
|
||||
from skiff.paths import EXTERNALS_DIR
|
||||
@@ -37,9 +43,14 @@ def external_repo_path(entry: dict[str, Any]) -> Path:
|
||||
|
||||
|
||||
def external_checkout_path(name: str, entry: dict[str, Any]) -> Path:
|
||||
"""Prefer an existing pre-shared-cache checkout for compatibility."""
|
||||
"""Use a local repo directly, otherwise return its external checkout."""
|
||||
from skiff.paths import EXTERNALS_DIR
|
||||
|
||||
configured_repo = str(entry.get("repo", ""))
|
||||
local_repo = Path(registry_repo(entry))
|
||||
if configured_repo.startswith("~") and local_repo.is_dir():
|
||||
return local_repo.resolve()
|
||||
|
||||
legacy = EXTERNALS_DIR / name
|
||||
return legacy if legacy.is_dir() else external_repo_path(entry)
|
||||
|
||||
|
||||
+111
-17
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, Iterable
|
||||
|
||||
@@ -12,6 +13,36 @@ class SkillChoice:
|
||||
kind: str
|
||||
description: str = ""
|
||||
installed: bool = False
|
||||
readonly_status: str = ""
|
||||
|
||||
|
||||
def fit_to_width(text: str, width: int) -> str:
|
||||
"""Trim text to terminal display width without splitting wide characters."""
|
||||
if width <= 0:
|
||||
return ""
|
||||
result: list[str] = []
|
||||
used = 0
|
||||
for char in text:
|
||||
if unicodedata.combining(char):
|
||||
char_width = 0
|
||||
else:
|
||||
char_width = 2 if unicodedata.east_asian_width(char) in {"W", "F"} else 1
|
||||
if used + char_width > width:
|
||||
break
|
||||
result.append(char)
|
||||
used += char_width
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def display_width(text: str) -> int:
|
||||
return sum(
|
||||
0
|
||||
if unicodedata.combining(char)
|
||||
else 2
|
||||
if unicodedata.east_asian_width(char) in {"W", "F"}
|
||||
else 1
|
||||
for char in text
|
||||
)
|
||||
|
||||
|
||||
def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoice]:
|
||||
@@ -24,12 +55,14 @@ def filter_choices(choices: Iterable[SkillChoice], query: str) -> list[SkillChoi
|
||||
if needle in choice.name.casefold()
|
||||
or needle in choice.kind.casefold()
|
||||
or needle in choice.description.casefold()
|
||||
or needle in choice.readonly_status.casefold()
|
||||
]
|
||||
|
||||
|
||||
def select_skills(
|
||||
choices: list[SkillChoice],
|
||||
*,
|
||||
scope_label: str = "当前作用域",
|
||||
wrapper: Callable[..., set[str] | None] | None = None,
|
||||
) -> set[str] | None:
|
||||
"""Open the selector. Return names, or None when cancelled."""
|
||||
@@ -48,29 +81,84 @@ def select_skills(
|
||||
stdscr.keypad(True)
|
||||
current = 0
|
||||
query = ""
|
||||
source_attr = curses.A_NORMAL
|
||||
status_attr = curses.A_NORMAL
|
||||
description_attr = curses.A_DIM
|
||||
try:
|
||||
if curses.has_colors():
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
curses.init_pair(1, curses.COLOR_CYAN, -1)
|
||||
curses.init_pair(2, curses.COLOR_GREEN, -1)
|
||||
source_attr = curses.color_pair(1)
|
||||
status_attr = curses.color_pair(2)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
while True:
|
||||
visible = filter_choices(choices, query)
|
||||
current = min(current, max(0, len(visible) - 1))
|
||||
stdscr.erase()
|
||||
height, width = stdscr.getmaxyx()
|
||||
header = "↑/↓ 移动 Space 勾选 / 搜索 Enter 安装 q 取消"
|
||||
stdscr.addnstr(0, 0, header, max(0, width - 1))
|
||||
text_width = max(0, width - 1)
|
||||
stdscr.addnstr(
|
||||
0,
|
||||
0,
|
||||
fit_to_width(f"安装目标: {scope_label}", text_width),
|
||||
text_width,
|
||||
)
|
||||
header = "↑↓ 移动 Space 选择 / 搜索 Enter 安装 q 退出"
|
||||
stdscr.addnstr(1, 0, fit_to_width(header, text_width), text_width)
|
||||
if query:
|
||||
stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1))
|
||||
stdscr.addnstr(
|
||||
2,
|
||||
0,
|
||||
fit_to_width(f"搜索: {query}", text_width),
|
||||
text_width,
|
||||
)
|
||||
|
||||
rows = max(1, height - 4)
|
||||
start = max(0, current - rows + 1)
|
||||
for row, choice in enumerate(visible[start : start + rows], start=2):
|
||||
index = start + row - 2
|
||||
item_capacity = max(1, (height - 4) // 3)
|
||||
start = max(0, current - item_capacity + 1)
|
||||
for offset, choice in enumerate(visible[start : start + item_capacity]):
|
||||
index = start + offset
|
||||
row = 3 + offset * 3
|
||||
mark = "x" if choice.name in selected else " "
|
||||
suffix = f" · {choice.description}" if choice.description else ""
|
||||
line = f"[{mark}] {choice.name} {choice.kind}{suffix}"
|
||||
attr = curses.A_REVERSE if index == current else curses.A_NORMAL
|
||||
stdscr.addnstr(row, 0, line, max(0, width - 1), attr)
|
||||
description = f" {choice.description}" if choice.description else ""
|
||||
focus_attr = curses.A_REVERSE if index == current else curses.A_NORMAL
|
||||
parts = [
|
||||
(f"[{mark}] ", focus_attr),
|
||||
(choice.name, focus_attr | curses.A_BOLD),
|
||||
(f" {choice.kind}", focus_attr | source_attr),
|
||||
]
|
||||
if choice.readonly_status:
|
||||
parts.append(
|
||||
(f" {choice.readonly_status}", focus_attr | status_attr)
|
||||
)
|
||||
column = 0
|
||||
for text, attr in parts:
|
||||
available = text_width - column
|
||||
if available <= 0:
|
||||
break
|
||||
fitted = fit_to_width(text, available)
|
||||
stdscr.addnstr(row, column, fitted, available, attr)
|
||||
column += display_width(fitted)
|
||||
stdscr.addnstr(
|
||||
row + 1,
|
||||
0,
|
||||
fit_to_width(description, text_width),
|
||||
text_width,
|
||||
focus_attr | description_attr,
|
||||
)
|
||||
|
||||
footer = f"已选择 {len(selected)} 项"
|
||||
stdscr.addnstr(height - 1, 0, footer, max(0, width - 1))
|
||||
footer = (
|
||||
f"已选择 {len(selected)} 项 · 取消选择不卸载,卸载用 skiff remove"
|
||||
)
|
||||
stdscr.addnstr(
|
||||
height - 1,
|
||||
0,
|
||||
fit_to_width(footer, text_width),
|
||||
text_width,
|
||||
)
|
||||
stdscr.refresh()
|
||||
key = stdscr.get_wch()
|
||||
|
||||
@@ -87,7 +175,7 @@ def select_skills(
|
||||
selected.symmetric_difference_update({name})
|
||||
elif key == "/":
|
||||
curses.curs_set(1)
|
||||
query = _read_query(stdscr, curses, width)
|
||||
query = _read_query(stdscr, curses, width, row=2)
|
||||
curses.curs_set(0)
|
||||
current = 0
|
||||
|
||||
@@ -95,12 +183,18 @@ def select_skills(
|
||||
return runner(draw)
|
||||
|
||||
|
||||
def _read_query(stdscr: object, curses: object, width: int) -> str:
|
||||
def _read_query(stdscr: object, curses: object, width: int, *, row: int = 1) -> str:
|
||||
query = ""
|
||||
while True:
|
||||
stdscr.move(1, 0)
|
||||
stdscr.move(row, 0)
|
||||
stdscr.clrtoeol()
|
||||
stdscr.addnstr(1, 0, f"搜索: {query}", max(0, width - 1))
|
||||
text_width = max(0, width - 1)
|
||||
stdscr.addnstr(
|
||||
row,
|
||||
0,
|
||||
fit_to_width(f"搜索: {query}", text_width),
|
||||
text_width,
|
||||
)
|
||||
stdscr.refresh()
|
||||
key = stdscr.get_wch()
|
||||
if key in ("\n", "\r") or key == curses.KEY_ENTER:
|
||||
|
||||
@@ -116,6 +116,9 @@ skiff add waza/think -a codex -g -y
|
||||
```
|
||||
|
||||
`skiff select` 会把 collection 展开为 `waza/think`、`waza/ui` 等候选项。
|
||||
普通 `skiff select` 只向项目安装,并只读显示每个 Agent 的全局安装状态;
|
||||
`skiff select -g` 只向全局安装。取消已勾选项不会卸载,卸载继续使用
|
||||
`skiff remove`。
|
||||
|
||||
卸载:
|
||||
|
||||
@@ -158,6 +161,7 @@ skiff kit init ack --copy # 用户明确要求时整份复制
|
||||
| skiff | 说明 |
|
||||
|-------|------|
|
||||
| `bootstrap` | 将本项目的 `skiff` skill 全局安装到所有 Agent |
|
||||
| `update` | 在 `~/.skills` 执行 `git pull`,更新 skiff 自身 |
|
||||
| `add <name> [-g] [-a AGENT...] [-y]` | 安装 |
|
||||
| `remove <name> [-g] [-a AGENT...] [-y]` | 卸载(`rm` / `r` 别名) |
|
||||
| `add --list` | 列出可用自研 skill |
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from skiff import cli
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -76,6 +80,19 @@ class CreateWorkflowTests(unittest.TestCase):
|
||||
self.assertIn("claude", first.stdout)
|
||||
self.assertIn("codex", first.stdout)
|
||||
|
||||
def test_update_pulls_skills_home(self) -> None:
|
||||
with (
|
||||
patch.object(cli, "SKILLS_HOME", self.skills_home),
|
||||
patch.object(cli, "ensure_skills_home"),
|
||||
patch.object(cli.subprocess, "run") as run,
|
||||
):
|
||||
cli.cmd_update(argparse.Namespace())
|
||||
|
||||
run.assert_called_once_with(
|
||||
["git", "-C", str(self.skills_home), "pull"],
|
||||
check=True,
|
||||
)
|
||||
|
||||
def test_create_writes_draft_and_brief_without_registering_owned_skill(self) -> None:
|
||||
project = self.home / "project"
|
||||
project.mkdir()
|
||||
|
||||
@@ -7,6 +7,9 @@ import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from skiff.registry import external_checkout_path, registry_repo
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -79,6 +82,17 @@ class RegistryCollectionTests(unittest.TestCase):
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_registry_repo_expands_home_relative_local_path(self) -> None:
|
||||
with patch.dict(os.environ, {"HOME": str(self.home)}):
|
||||
self.assertEqual(
|
||||
registry_repo({"repo": "~/.skills"}),
|
||||
str(self.skills_home),
|
||||
)
|
||||
self.assertEqual(
|
||||
external_checkout_path("skills", {"repo": "~/.skills"}),
|
||||
self.skills_home.resolve(),
|
||||
)
|
||||
|
||||
def test_add_collection_installs_every_discovered_skill(self) -> None:
|
||||
result = self.run_skiff("add", "test-pack", "-g", "-a", "codex")
|
||||
|
||||
|
||||
+177
-4
@@ -13,13 +13,17 @@ from unittest.mock import Mock, patch
|
||||
from skiff import cli
|
||||
from skiff import yaml_io
|
||||
from skiff.registry import external_repo_path, external_skill_path
|
||||
from skiff.selector import SkillChoice, filter_choices, select_skills
|
||||
from skiff.selector import SkillChoice, filter_choices, fit_to_width, select_skills
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
class SelectorTests(unittest.TestCase):
|
||||
def test_fit_to_width_counts_cjk_as_two_columns(self) -> None:
|
||||
self.assertEqual(fit_to_width("a中文b", 4), "a中")
|
||||
self.assertEqual(fit_to_width("初始化 ACK", 8), "初始化 A")
|
||||
|
||||
def test_manifest_entry_targets_round_trip(self) -> None:
|
||||
data = {
|
||||
"skills": [
|
||||
@@ -36,12 +40,18 @@ class SelectorTests(unittest.TestCase):
|
||||
def test_filter_matches_name_kind_and_description(self) -> None:
|
||||
choices = [
|
||||
SkillChoice("frontend-design", "external", "创建界面"),
|
||||
SkillChoice("discussion-notes", "owned", "维护讨论笔记"),
|
||||
SkillChoice(
|
||||
"discussion-notes",
|
||||
"owned",
|
||||
"维护讨论笔记",
|
||||
readonly_status="全局: codex",
|
||||
),
|
||||
]
|
||||
|
||||
self.assertEqual([c.name for c in filter_choices(choices, "front")], ["frontend-design"])
|
||||
self.assertEqual([c.name for c in filter_choices(choices, "owned")], ["discussion-notes"])
|
||||
self.assertEqual([c.name for c in filter_choices(choices, "界面")], ["frontend-design"])
|
||||
self.assertEqual([c.name for c in filter_choices(choices, "codex")], ["discussion-notes"])
|
||||
|
||||
def test_selector_preserves_preselected_items(self) -> None:
|
||||
choices = [
|
||||
@@ -59,6 +69,71 @@ class SelectorTests(unittest.TestCase):
|
||||
)
|
||||
|
||||
self.assertEqual(selected, {"already-there"})
|
||||
rendered = [call.args[2] for call in screen.addnstr.call_args_list]
|
||||
self.assertTrue(any("取消选择不卸载" in line for line in rendered))
|
||||
|
||||
def test_selector_renders_scope_and_readonly_global_status(self) -> None:
|
||||
choices = [
|
||||
SkillChoice(
|
||||
"ack",
|
||||
"owned",
|
||||
installed=False,
|
||||
readonly_status="全局: cursor,claude,codex",
|
||||
)
|
||||
]
|
||||
screen = Mock()
|
||||
screen.getmaxyx.return_value = (24, 120)
|
||||
screen.get_wch.return_value = "\n"
|
||||
|
||||
with patch.object(curses, "curs_set"):
|
||||
selected = select_skills(
|
||||
choices,
|
||||
scope_label="项目 /code/app(全局状态只读)",
|
||||
wrapper=lambda draw: draw(screen),
|
||||
)
|
||||
|
||||
rendered = [call.args[2] for call in screen.addnstr.call_args_list]
|
||||
self.assertEqual(selected, set())
|
||||
self.assertTrue(any("安装目标: 项目 /code/app" in line for line in rendered))
|
||||
title = "".join(
|
||||
call.args[2]
|
||||
for call in screen.addnstr.call_args_list
|
||||
if call.args[0] == 3
|
||||
)
|
||||
self.assertIn("[ ] ack", title)
|
||||
self.assertIn("全局: cursor,claude,codex", title)
|
||||
self.assertNotIn("初始化", title)
|
||||
|
||||
def test_selector_renders_description_on_indented_second_line(self) -> None:
|
||||
choices = [
|
||||
SkillChoice(
|
||||
"ack",
|
||||
"owned",
|
||||
description="初始化、检查并运行 ACK",
|
||||
readonly_status="全局: codex",
|
||||
)
|
||||
]
|
||||
screen = Mock()
|
||||
screen.getmaxyx.return_value = (24, 100)
|
||||
screen.get_wch.return_value = "\n"
|
||||
|
||||
with patch.object(curses, "curs_set"):
|
||||
select_skills(choices, wrapper=lambda draw: draw(screen))
|
||||
|
||||
title_calls = [
|
||||
call for call in screen.addnstr.call_args_list if call.args[0] == 3
|
||||
]
|
||||
description_call = next(
|
||||
call for call in screen.addnstr.call_args_list if call.args[0] == 4
|
||||
)
|
||||
self.assertEqual(
|
||||
"".join(call.args[2] for call in title_calls),
|
||||
"[ ] ack owned 全局: codex",
|
||||
)
|
||||
self.assertEqual(description_call.args[2], " 初始化、检查并运行 ACK")
|
||||
name_call = next(call for call in title_calls if call.args[2] == "ack")
|
||||
self.assertTrue(name_call.args[4] & curses.A_BOLD)
|
||||
self.assertTrue(description_call.args[4] & curses.A_DIM)
|
||||
|
||||
def test_shared_repo_path_is_same_for_different_skill_entries(self) -> None:
|
||||
first = {"repo": "https://example.test/skills.git", "ref": "main", "path": "a"}
|
||||
@@ -136,6 +211,17 @@ class SelectCommandTests(unittest.TestCase):
|
||||
stdin.isatty.return_value = True
|
||||
stdout.isatty.return_value = True
|
||||
installed: list[str] = []
|
||||
selected_choices: list[SkillChoice] = []
|
||||
selected_scope: list[str] = []
|
||||
|
||||
def choose(
|
||||
choices: list[SkillChoice],
|
||||
*,
|
||||
scope_label: str,
|
||||
) -> set[str]:
|
||||
selected_choices.extend(choices)
|
||||
selected_scope.append(scope_label)
|
||||
return {"owned-one", "external-one"}
|
||||
|
||||
with (
|
||||
patch.object(cli.sys, "stdin", stdin),
|
||||
@@ -157,6 +243,11 @@ class SelectCommandTests(unittest.TestCase):
|
||||
patch.object(cli, "_registry_skill_names", return_value=["external-one"]),
|
||||
patch.object(cli, "external_skill_path", return_value=external),
|
||||
patch.object(cli, "_list_fully_installed_names", return_value=["owned-one"]),
|
||||
patch.object(
|
||||
cli,
|
||||
"_global_installation_note",
|
||||
return_value="全局: codex",
|
||||
),
|
||||
patch.object(
|
||||
cli,
|
||||
"_is_fully_installed",
|
||||
@@ -165,7 +256,7 @@ class SelectCommandTests(unittest.TestCase):
|
||||
patch.object(
|
||||
cli,
|
||||
"select_skills",
|
||||
return_value={"owned-one", "external-one"},
|
||||
side_effect=choose,
|
||||
),
|
||||
patch.object(
|
||||
cli,
|
||||
@@ -183,6 +274,40 @@ class SelectCommandTests(unittest.TestCase):
|
||||
self.assertIn("source: registry", manifest)
|
||||
self.assertIn("targets:", manifest)
|
||||
self.assertIn("codex", manifest)
|
||||
self.assertEqual(
|
||||
[choice.readonly_status for choice in selected_choices],
|
||||
["全局: codex", "全局: codex"],
|
||||
)
|
||||
self.assertIn("全局状态只读", selected_scope[0])
|
||||
|
||||
def test_global_installation_note_reports_agents_and_source_conflicts(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
expected = root / "expected"
|
||||
other = root / "other"
|
||||
expected.mkdir()
|
||||
other.mkdir()
|
||||
global_dirs = {
|
||||
target: root / target / "skills"
|
||||
for target in ("cursor", "claude", "codex")
|
||||
}
|
||||
for directory in global_dirs.values():
|
||||
directory.mkdir(parents=True)
|
||||
(global_dirs["cursor"] / "ack").symlink_to(expected, target_is_directory=True)
|
||||
(global_dirs["claude"] / "ack").symlink_to(other, target_is_directory=True)
|
||||
|
||||
with patch.object(
|
||||
cli,
|
||||
"agent_skill_dir",
|
||||
side_effect=lambda target, project_root=None: global_dirs[target],
|
||||
):
|
||||
note = cli._global_installation_note(
|
||||
"ack",
|
||||
expected,
|
||||
["cursor", "claude", "codex"],
|
||||
)
|
||||
|
||||
self.assertEqual(note, "全局: cursor;全局同名冲突: claude")
|
||||
|
||||
def test_select_rejects_invalid_registry_name_before_rendering(self) -> None:
|
||||
args = argparse.Namespace(
|
||||
@@ -227,7 +352,7 @@ class SelectCommandTests(unittest.TestCase):
|
||||
selected_choices: list[SkillChoice] = []
|
||||
installed: list[tuple[str, str | None]] = []
|
||||
|
||||
def choose(choices: list[SkillChoice]) -> set[str]:
|
||||
def choose(choices: list[SkillChoice], **_: object) -> set[str]:
|
||||
selected_choices.extend(choices)
|
||||
return {"waza/think"}
|
||||
|
||||
@@ -268,6 +393,54 @@ class SelectCommandTests(unittest.TestCase):
|
||||
self.assertEqual([choice.name for choice in selected_choices], ["waza/think", "waza/ui"])
|
||||
self.assertEqual(installed, [("think", "registry:waza")])
|
||||
|
||||
def test_select_deduplicates_local_registry_collection_from_owned(self) -> None:
|
||||
args = argparse.Namespace(
|
||||
agents=[["codex"]],
|
||||
global_scope=True,
|
||||
project=None,
|
||||
yes=False,
|
||||
)
|
||||
stdin = Mock()
|
||||
stdout = Mock()
|
||||
stdin.isatty.return_value = True
|
||||
stdout.isatty.return_value = True
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
skills_root = Path(temp) / "skills"
|
||||
(skills_root / "ack").mkdir(parents=True)
|
||||
selected_choices: list[SkillChoice] = []
|
||||
|
||||
def choose(choices: list[SkillChoice], **_: object) -> set[str]:
|
||||
selected_choices.extend(choices)
|
||||
return set()
|
||||
|
||||
with (
|
||||
patch.object(cli.sys, "stdin", stdin),
|
||||
patch.object(cli.sys, "stdout", stdout),
|
||||
patch.object(cli, "SKILLS_DIR", skills_root),
|
||||
patch.object(cli, "ensure_skills_home"),
|
||||
patch.object(cli, "list_owned_skills", return_value=["ack"]),
|
||||
patch.object(cli, "skill_description", return_value="owned ack"),
|
||||
patch.object(
|
||||
cli,
|
||||
"load_registry",
|
||||
return_value={
|
||||
"skills": {
|
||||
"repo": "~/.skills",
|
||||
"ref": "main",
|
||||
"path": "skills",
|
||||
}
|
||||
},
|
||||
),
|
||||
patch.object(cli, "_registry_skill_names", return_value=["ack"]),
|
||||
patch.object(cli, "external_skill_path", return_value=skills_root),
|
||||
patch.object(cli, "_is_fully_installed", return_value=False),
|
||||
patch.object(cli, "select_skills", side_effect=choose),
|
||||
):
|
||||
cli.cmd_select(args)
|
||||
|
||||
self.assertEqual([choice.name for choice in selected_choices], ["ack"])
|
||||
|
||||
def test_install_rolls_back_earlier_target_when_later_target_fails(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
|
||||
Reference in New Issue
Block a user