muse-agent-example/tests/architecture/test_skill_projections.py
2026-08-28 05:38:26 +08:00

140 lines
5.4 KiB
Python

"""Pi/DSH 扁平 Skill 投影与 Muse manifest 对账门。"""
from __future__ import annotations
import json
import pathlib
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
import urllib.parse
ROOT = next(
parent
for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents)
if (parent / "AGENTS.md").is_file() and (parent / ".git").exists()
)
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from framework.generate_skill_projections import generate, host_skill_root # noqa: E402
MANIFEST = ROOT / "muse" / "lifecycle" / "quality" / "harness" / "manifests" / "skills.json"
MARKDOWN_LINK = re.compile(r"\[[^\]\n]*\]\(([^)\s]+)")
HOST_TARGETS = ("dsh", "pi")
def _model_routed() -> dict[str, str]:
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
return {
entry["name"]: entry["skill_path"]
for entry in manifest["skills"]
if entry.get("invocation") == "model_routed"
}
class SkillProjectionTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
generate(ROOT)
def test_projection_is_flat_and_same_source(self) -> None:
expected = _model_routed()
for target in HOST_TARGETS:
host_root = host_skill_root(ROOT, target)
projected_names = {
path.parent.name for path in host_root.glob("*/SKILL.md")
}
self.assertEqual(projected_names, set(expected))
for name, skill_path in expected.items():
source = ROOT / skill_path
source_text = source.read_text(encoding="utf-8")
projected = host_root / name / "SKILL.md"
self.assertEqual(projected.parent.parent.name, "skills")
self.assertEqual(projected.read_text(encoding="utf-8"), source_text)
source_references = source.parent / "references"
projected_references = projected.parent / "references"
if source_references.is_dir():
self.assertEqual(
sorted(
path.relative_to(source_references).as_posix()
for path in source_references.rglob("*")
),
sorted(
path.relative_to(projected_references).as_posix()
for path in projected_references.rglob("*")
),
)
def test_projected_reference_links_resolve(self) -> None:
missing: list[str] = []
for target in HOST_TARGETS:
host_root = host_skill_root(ROOT, target)
for source in host_root.rglob("*.md"):
for raw_target in MARKDOWN_LINK.findall(source.read_text(encoding="utf-8")):
target_path = urllib.parse.unquote(
raw_target.strip("<>").split("#", 1)[0]
)
if not target_path or target_path.startswith(
("#", "http://", "https://", "mailto:", "/")
):
continue
if not (source.parent / target_path).resolve().exists():
missing.append(f"{source.relative_to(ROOT)}: {raw_target}")
self.assertEqual(missing, [], "Skill 投影存在死链:\n" + "\n".join(missing))
def test_orchestrated_skill_is_not_projected(self) -> None:
projected_names = {
path.parent.name
for target in HOST_TARGETS
for path in host_skill_root(ROOT, target).glob("*/SKILL.md")
}
self.assertNotIn("decide-candidate", projected_names)
self.assertNotIn("record-run-evidence", projected_names)
def test_framework_catalog_is_not_tracked(self) -> None:
tracked = subprocess.check_output(
["git", "ls-files", "framework/catalog"],
cwd=ROOT,
text=True,
).strip()
self.assertEqual(tracked, "", tracked)
def test_host_projections_are_gitignored(self) -> None:
for rel in (".pi/skills", ".dsh/skills"):
result = subprocess.run(
["git", "check-ignore", "-q", rel],
cwd=ROOT,
)
self.assertEqual(result.returncode, 0, f"{rel} 必须被 gitignore")
def test_generate_reruns_in_clean_tree(self) -> None:
expected = _model_routed()
with tempfile.TemporaryDirectory() as directory:
tmp = pathlib.Path(directory)
dest_manifest = tmp / MANIFEST.relative_to(ROOT)
dest_manifest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(MANIFEST, dest_manifest)
for skill_path in expected.values():
source = ROOT / skill_path
dest = tmp / skill_path
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, dest)
references = source.parent / "references"
if references.is_dir():
shutil.copytree(references, dest.parent / "references")
result = generate(tmp)
self.assertEqual(result["count"], len(expected))
for target in HOST_TARGETS:
host_root = host_skill_root(tmp, target)
for name in expected:
self.assertTrue((host_root / name / "SKILL.md").is_file())
if __name__ == "__main__":
unittest.main()