58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
|
# ============================================================
|
|||
|
|
# M-D 黄金测试：Skill manifest 扩展（capabilities/ragScopes/version）
|
|||
|
|
# + 健康探测历史 + 旧配置自动升级
|
|||
|
|
# ============================================================
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
|
|||
|
|
from server.agent_core.skills import SkillRegistry
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_default_manifest_has_md_fields(tmp_path):
|
|||
|
|
reg = SkillRegistry(path=str(tmp_path / "skills.json"))
|
|||
|
|
s = reg.get("algo.stub")
|
|||
|
|
assert s is not None
|
|||
|
|
assert s["version"] == "1.0"
|
|||
|
|
assert "flex_schedule" in s["capabilities"]
|
|||
|
|
assert set(s["ragScopes"]) == {"process", "sop"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_legacy_config_upgraded_with_new_fields(tmp_path):
|
|||
|
|
path = tmp_path / "skills.json"
|
|||
|
|
legacy = {"skills": [{
|
|||
|
|
"skill_id": "algo.legacy", "name": "旧算法", "endpoint": "http://x:1",
|
|||
|
|
"auth": "", "max_power": "P1", "track": "flex", "enabled": True,
|
|||
|
|
"timeout_sec": 60.0, "description": "", "golden_tests": [],
|
|||
|
|
}]}
|
|||
|
|
path.write_text(json.dumps(legacy), encoding="utf-8")
|
|||
|
|
reg = SkillRegistry(path=str(path))
|
|||
|
|
s = reg.get("algo.legacy")
|
|||
|
|
assert s["version"] == "1.0"
|
|||
|
|
assert s["capabilities"] == ["flex_schedule"]
|
|||
|
|
assert s["ragScopes"] == []
|
|||
|
|
# 升级已落盘
|
|||
|
|
saved = json.loads(path.read_text(encoding="utf-8"))
|
|||
|
|
assert saved["skills"][0]["capabilities"] == ["flex_schedule"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_health_probe_records_history(tmp_path):
|
|||
|
|
reg = SkillRegistry(path=str(tmp_path / "skills.json"))
|
|||
|
|
reg.health("algo.stub")
|
|||
|
|
reg.health("algo.stub")
|
|||
|
|
hist = reg.history("algo.stub")
|
|||
|
|
assert len(hist) == 2
|
|||
|
|
assert all(h["ok"] for h in hist)
|
|||
|
|
assert all("ts" in h for h in hist)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_upsert_keeps_rag_scopes(tmp_path):
|
|||
|
|
reg = SkillRegistry(path=str(tmp_path / "skills.json"))
|
|||
|
|
reg.upsert({"skill_id": "algo.opt", "name": "外部优化器", "endpoint": "http://127.0.0.1:9000",
|
|||
|
|
"ragScopes": ["process"], "capabilities": ["flex_schedule", "due_promise"],
|
|||
|
|
"version": "2.1"})
|
|||
|
|
s = reg.get("algo.opt")
|
|||
|
|
assert s["version"] == "2.1"
|
|||
|
|
assert s["ragScopes"] == ["process"]
|
|||
|
|
assert "due_promise" in s["capabilities"]
|