224 lines
10 KiB
Python
224 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""skill.json 注册表单测：内存加载 / 落库 / 内嵌常见根 / 分层查找 / 后台扫盘。
|
||
|
||
覆盖 2026-09-25 需求：推拉遇软链接空壳 -> 先查内存注册表 -> 再快扫内嵌常见
|
||
终端位置（oc/wb/cc-switch…）-> 都没中才后台扫盘（不阻塞），扫完落库 skill.json
|
||
（名字 -> 位置 的一堆键值）；用户选定的有效目录永远优先（real_src 返回 None）。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import ocskill # noqa: E402
|
||
|
||
FAIL: list[str] = []
|
||
|
||
|
||
def check(name: str, cond: bool, extra: str = "") -> None:
|
||
print(("PASS " if cond else "FAIL ") + name + (" | " + extra if extra else ""))
|
||
if not cond:
|
||
FAIL.append(name)
|
||
|
||
|
||
tmp = Path(tempfile.mkdtemp(prefix="ocskill_reg_")).resolve()
|
||
print("tmp =", tmp)
|
||
|
||
_orig = { # 测完还原，别污染同进程后续用例
|
||
"REGISTRY_PATH": ocskill.REGISTRY_PATH,
|
||
"_REG": ocskill._REG,
|
||
"env_alt": os.environ.get("OC_SKILL_ALT_ROOTS"),
|
||
"builtin": ocskill.builtin_skill_roots,
|
||
"deep": ocskill.deep_scan_roots,
|
||
"bg": ocskill.start_background_scan,
|
||
}
|
||
|
||
|
||
def reset_registry(path: Path) -> None:
|
||
ocskill.REGISTRY_PATH = path
|
||
ocskill._REG = None # 强制下次 load_registry 重读
|
||
|
||
|
||
def mk_skill(folder: Path, name: str) -> Path:
|
||
folder.mkdir(parents=True, exist_ok=True)
|
||
(folder / "SKILL.md").write_text(
|
||
f"---\nname: {name}\ndescription: 测试技能\n---\n# {name}\n",
|
||
encoding="utf-8")
|
||
return folder
|
||
|
||
|
||
try:
|
||
# ---------- 1. 空表起步 + remember + 落库 + 启动重读入内存 ----------
|
||
reset_registry(tmp / "skill.json")
|
||
reg = ocskill.load_registry()
|
||
check("1 空表起步 skills={}", reg.get("skills") == {}, str(reg))
|
||
real_a = mk_skill(tmp / "roots" / "A" / "foo-skill", "foo-skill")
|
||
ocskill.remember_skills({"foo-skill": [str(real_a)]})
|
||
check("1 remember 进内存", ocskill.registry_locations("foo-skill") == [str(real_a)],
|
||
str(ocskill.registry_locations("foo-skill")))
|
||
check("1 remember 落库文件存在", (tmp / "skill.json").is_file())
|
||
data = json.loads((tmp / "skill.json").read_text(encoding="utf-8"))
|
||
check("1 落库结构 = version/updated/skills",
|
||
{"version", "updated", "skills"} <= set(data.keys())
|
||
and data["skills"].get("foo-skill") == [str(real_a)], str(data)[:200])
|
||
ocskill._REG = None # 模拟下次启动
|
||
reg2 = ocskill.load_registry()
|
||
check("1 启动重读入内存（skill.json -> 内存）",
|
||
reg2.get("skills", {}).get("foo-skill") == [str(real_a)], str(reg2)[:200])
|
||
# 重复 remember 不产生重复条目
|
||
ocskill.remember_skills({"foo-skill": [str(real_a)]})
|
||
check("1 remember 幂等（不重复）",
|
||
ocskill.registry_locations("foo-skill") == [str(real_a)])
|
||
# 损坏文件不炸：从空表起步
|
||
(tmp / "skill.json").write_text("{broken json", encoding="utf-8")
|
||
ocskill._REG = None
|
||
reg3 = ocskill.load_registry()
|
||
check("1 损坏 skill.json -> 空表起步不炸", reg3.get("skills") == {}, str(reg3))
|
||
|
||
# ---------- 2. 内嵌常见终端根 ----------
|
||
roots = ocskill.builtin_skill_roots()
|
||
home = Path.home()
|
||
for tag, expect in [
|
||
("oc", home / ".config" / "opencode" / "skills"),
|
||
("wb", home / ".workbuddy" / "skills"),
|
||
("cc-switch", home / ".cc-switch" / "skills"),
|
||
("claude", home / ".claude" / "skills"),
|
||
]:
|
||
key = os.path.normcase(os.path.abspath(str(expect)))
|
||
check(f"2 内嵌 {tag} 位置", key in [os.path.normcase(os.path.abspath(r)) for r in roots],
|
||
str(expect))
|
||
agent_dirs = [os.path.normcase(os.path.abspath(t["dir"]))
|
||
for t in ocskill.load_agent_table() if t.get("dir")]
|
||
got = [os.path.normcase(os.path.abspath(r)) for r in roots]
|
||
check("2 agents.md 终端目录纳入", all(d in got for d in agent_dirs),
|
||
f"agents={agent_dirs}")
|
||
check("2 内嵌根无重复", len(got) == len(set(got)), str(len(got)))
|
||
check("2 env 追加根", True) # 用例 3 实际验证
|
||
|
||
# ---------- 3. tier1 快扫 scan_roots_for_name + env 追加 ----------
|
||
root_b = tmp / "roots" / "B"
|
||
target = mk_skill(root_b / "quick-skill", "quick-skill")
|
||
found = ocskill.scan_roots_for_name("quick-skill", roots=[str(root_b)])
|
||
check("3 快扫按名字命中真身", found == [str(target)], str(found))
|
||
check("3 快扫找不到时返回空", ocskill.scan_roots_for_name("nope-xyz", roots=[str(root_b)]) == [])
|
||
# env 追加的根也进 builtin
|
||
os.environ["OC_SKILL_ALT_ROOTS"] = str(root_b)
|
||
roots_env = ocskill.builtin_skill_roots()
|
||
check("2 env 追加根进 builtin",
|
||
os.path.normcase(os.path.abspath(str(root_b))) in
|
||
[os.path.normcase(os.path.abspath(r)) for r in roots_env], str(roots_env)[:200])
|
||
|
||
# ---------- 4. real_src 分层：① 内存注册表命中（零扫盘） ----------
|
||
reset_registry(tmp / "skill2.json")
|
||
ocskill.remember_skills({"foo-skill": [str(real_a)]})
|
||
shell1 = tmp / "shell" / "foo-skill"
|
||
shell1.mkdir(parents=True) # 空壳：目录在、没有 SKILL.md
|
||
bg_called: list[str] = []
|
||
ocskill.start_background_scan = lambda name="": bg_called.append(name) or True
|
||
real = ocskill.real_src(str(shell1))
|
||
check("4 空壳 -> 内存注册表一步命中", real == str(real_a), f"real={real}")
|
||
check("4 注册表命中不触发后台扫盘", bg_called == [], str(bg_called))
|
||
|
||
# ---------- 5. real_src 分层：② tier1 快扫命中并落库 ----------
|
||
shell2 = tmp / "shell" / "quick-skill"
|
||
shell2.mkdir(parents=True)
|
||
ocskill._REG = None # 空注册表，逼走 tier1
|
||
real2 = ocskill.real_src(str(shell2))
|
||
check("5 空壳 -> 快扫内嵌常见位置命中", real2 == str(target), f"real={real2}")
|
||
check("5 快扫命中即落库",
|
||
"quick-skill" in ocskill.load_registry().get("skills", {}),
|
||
str(ocskill.load_registry().get("skills", {}))[:200])
|
||
check("5 快扫命中不触发后台扫盘", bg_called == [], str(bg_called))
|
||
|
||
# ---------- 6. real_src 分层：③ 都没中 -> 启动后台扫盘（不阻塞返回 None） ----------
|
||
shell3 = tmp / "shell" / "ghost-skill"
|
||
shell3.mkdir(parents=True)
|
||
ocskill._REG = None
|
||
os.environ["OC_SKILL_ALT_ROOTS"] = str(tmp / "nowhere") # 指到不存在的根
|
||
real3 = ocskill.real_src(str(shell3))
|
||
check("6 没真身 -> 返回 None（走原诊断路径）", real3 is None, str(real3))
|
||
check("6 没真身 -> 启动后台扫盘", bg_called == ["ghost-skill"], str(bg_called))
|
||
|
||
# ---------- 7. 后台扫盘：单飞 + 扫完落库 skill.json ----------
|
||
ocskill.start_background_scan = _orig["bg"] # 还原真身
|
||
root_c = tmp / "roots" / "C"
|
||
mk_skill(root_c / "bg-skill-1", "bg-skill-1")
|
||
mk_skill(root_c / "bg-skill-2", "bg-skill-2")
|
||
(root_c / ".hidden-skill").mkdir() # 点前缀跳过
|
||
(root_c / "no-md-skill").mkdir() # 无 SKILL.md 跳过
|
||
reset_registry(tmp / "skill3.json")
|
||
ocskill.builtin_skill_roots = lambda distro=ocskill.DEFAULT_DISTRO: [str(root_c)]
|
||
ocskill.deep_scan_roots = lambda: []
|
||
# 放慢扫描以便验证单飞
|
||
_real_scan = ocskill._scan_root_skills
|
||
|
||
def _slow_scan(root):
|
||
time.sleep(0.4)
|
||
return _real_scan(root)
|
||
|
||
ocskill._scan_root_skills = _slow_scan
|
||
try:
|
||
started = ocskill.start_background_scan("x")
|
||
check("7 后台扫盘启动成功", started is True, str(started))
|
||
again = ocskill.start_background_scan("y")
|
||
check("7 后台扫盘单飞（运行中不叠加线程）", again is False, str(again))
|
||
deadline = time.time() + 10
|
||
while ocskill._BG_SCAN_RUNNING and time.time() < deadline:
|
||
time.sleep(0.05)
|
||
check("7 后台扫盘完成", not ocskill._BG_SCAN_RUNNING)
|
||
finally:
|
||
ocskill._scan_root_skills = _real_scan
|
||
skills = ocskill.load_registry().get("skills", {})
|
||
check("7 扫盘结果进内存注册表",
|
||
"bg-skill-1" in skills and "bg-skill-2" in skills, str(list(skills)))
|
||
check("7 点前缀 / 无 SKILL.md 不登记",
|
||
"no-md-skill" not in skills and ".hidden-skill" not in skills, str(list(skills)))
|
||
data3 = json.loads((tmp / "skill3.json").read_text(encoding="utf-8"))
|
||
check("7 扫盘结果落库 skill.json", "bg-skill-1" in data3.get("skills", {}),
|
||
str(data3)[:200])
|
||
|
||
# ---------- 8. 用户选定的有效目录优先（real_src 返回 None 不动） ----------
|
||
valid = mk_skill(tmp / "valid-skill", "valid-skill")
|
||
ocskill._REG = None
|
||
check("8 有效目录不动（选定优先）", ocskill.real_src(str(valid)) is None, str(valid))
|
||
# zip / SKILL.md 文件输入也不动
|
||
mdfile = valid / "SKILL.md"
|
||
check("8 SKILL.md 文件不动", ocskill.real_src(str(mdfile)) is None, str(mdfile))
|
||
|
||
# ---------- 9. deep_scan_roots 浅层扫描 ----------
|
||
old_cwd = os.getcwd()
|
||
work = tmp / "cwd"
|
||
(work / "someapp" / "skills").mkdir(parents=True)
|
||
try:
|
||
os.chdir(work)
|
||
deep = _orig["deep"]() # 用原函数（case7 把它补丁成了 lambda: []）
|
||
finally:
|
||
os.chdir(old_cwd)
|
||
key = os.path.normcase(os.path.abspath(str(work / "someapp" / "skills")))
|
||
check("9 扩展根扫到 <cwd>/someapp/skills",
|
||
key in [os.path.normcase(os.path.abspath(d)) for d in deep], str(deep))
|
||
check("9 扩展根无重复",
|
||
len([os.path.normcase(os.path.abspath(d)) for d in deep]) ==
|
||
len(set(os.path.normcase(os.path.abspath(d)) for d in deep)), str(len(deep)))
|
||
|
||
finally:
|
||
# ---------- 还原现场 ----------
|
||
ocskill.REGISTRY_PATH = _orig["REGISTRY_PATH"]
|
||
ocskill._REG = _orig["_REG"]
|
||
ocskill.builtin_skill_roots = _orig["builtin"]
|
||
ocskill.deep_scan_roots = _orig["deep"]
|
||
ocskill.start_background_scan = _orig["bg"]
|
||
if _orig["env_alt"] is None:
|
||
os.environ.pop("OC_SKILL_ALT_ROOTS", None)
|
||
else:
|
||
os.environ["OC_SKILL_ALT_ROOTS"] = _orig["env_alt"]
|
||
|
||
print()
|
||
if FAIL:
|
||
print(f"{len(FAIL)} FAILED: {FAIL}")
|
||
sys.exit(1)
|
||
print("ALL PASS")
|