from __future__ import annotations import io import zipfile from pathlib import Path import pytest from src.services import skill_service as svc from src.storage.postgres.models_business import Skill def _build_zip(files: dict[str, str]) -> bytes: buf = io.BytesIO() with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf: for path, content in files.items(): zf.writestr(path, content) return buf.getvalue() def test_parse_skill_markdown_ok(): content = ( "---\n" "name: demo-skill\n" "description: demo description\n" "---\n" "# Demo\n" ) name, desc, meta = svc._parse_skill_markdown(content) assert name == "demo-skill" assert desc == "demo description" assert meta["name"] == "demo-skill" def test_parse_skill_markdown_requires_frontmatter(): with pytest.raises(ValueError, match="frontmatter"): svc._parse_skill_markdown("# missing") def test_validate_skill_slug(): assert svc.validate_skill_slug("demo-skill") == "demo-skill" with pytest.raises(ValueError, match="无效 skill slug"): svc.validate_skill_slug("../bad") @pytest.mark.asyncio async def test_get_skill_dependency_options(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc, "_get_buildin_tool_names", lambda: ["calculator", "search"]) monkeypatch.setattr(svc, "get_mcp_server_names", lambda: ["mcp-a", "mcp-b"]) class FakeRepo: def __init__(self, _db): pass async def list_all(self): return [ Skill(slug="alpha", name="alpha", description="a", dir_path="skills/alpha"), Skill(slug="beta", name="beta", description="b", dir_path="skills/beta"), ] monkeypatch.setattr(svc, "SkillRepository", FakeRepo) result = await svc.get_skill_dependency_options(None) assert result["tools"] == ["calculator", "search"] assert result["mcps"] == ["mcp-a", "mcp-b"] assert result["skills"] == ["alpha", "beta"] def test_resolve_relative_path_blocks_traversal(tmp_path: Path): skill_dir = tmp_path / "skill" skill_dir.mkdir(parents=True, exist_ok=True) with pytest.raises(ValueError, match="上级路径"): svc._resolve_relative_path(skill_dir, "../outside.txt") @pytest.mark.asyncio async def test_import_skill_zip_conflict_rewrite_name(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: existing_slugs = {"demo"} created_item: Skill | None = None def __init__(self, _db): pass async def exists_slug(self, slug: str) -> bool: return slug in self.__class__.existing_slugs async def create( self, *, slug: str, name: str, description: str, tool_dependencies: list[str] | None, mcp_dependencies: list[str] | None, skill_dependencies: list[str] | None, dir_path: str, created_by: str | None, ) -> Skill: item = Skill( slug=slug, name=name, description=description, tool_dependencies=tool_dependencies or [], mcp_dependencies=mcp_dependencies or [], skill_dependencies=skill_dependencies or [], dir_path=dir_path, created_by=created_by, updated_by=created_by, ) self.__class__.existing_slugs.add(slug) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "demo/SKILL.md": ( "---\n" "name: demo\n" "description: this is demo\n" "---\n" "# Demo\n" ), "demo/prompts/system.md": "You are demo skill", } ) item = await svc.import_skill_zip( None, filename="demo.zip", file_bytes=zip_bytes, created_by="root", ) assert item.slug == "demo-v2" assert item.name == "demo-v2" skill_md = (tmp_path / "skills" / "demo-v2" / "SKILL.md").read_text(encoding="utf-8") assert "name: demo-v2" in skill_md @pytest.mark.asyncio async def test_update_skill_md_syncs_metadata(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) skill_dir = tmp_path / "skills" / "demo" skill_dir.mkdir(parents=True, exist_ok=True) (skill_dir / "SKILL.md").write_text( "---\nname: demo\ndescription: old\n---\n# old\n", encoding="utf-8", ) item = Skill( slug="demo", name="demo", description="old", dir_path="skills/demo", created_by="root", updated_by="root", ) async def fake_get_skill_or_raise(_db, _slug: str): return item updates: dict[str, str | None] = {} class FakeRepo: def __init__(self, _db): pass async def update_metadata( self, _item: Skill, *, name: str, description: str, updated_by: str | None, ) -> Skill: updates["name"] = name updates["description"] = description updates["updated_by"] = updated_by return item monkeypatch.setattr(svc, "get_skill_or_raise", fake_get_skill_or_raise) monkeypatch.setattr(svc, "SkillRepository", FakeRepo) new_content = ( "---\n" "name: demo\n" "description: updated desc\n" "---\n" "# updated\n" ) await svc.update_skill_file( None, slug="demo", relative_path="SKILL.md", content=new_content, updated_by="admin", ) assert updates["name"] == "demo" assert updates["description"] == "updated desc" assert updates["updated_by"] == "admin" saved_content = (skill_dir / "SKILL.md").read_text(encoding="utf-8") assert "description: updated desc" in saved_content @pytest.mark.asyncio async def test_update_skill_dependencies(monkeypatch: pytest.MonkeyPatch): item = Skill( slug="alpha", name="alpha", description="alpha", dir_path="skills/alpha", tool_dependencies=[], mcp_dependencies=[], skill_dependencies=[], ) monkeypatch.setattr(svc, "_get_buildin_tool_names", lambda: ["calculator"]) monkeypatch.setattr(svc, "get_mcp_server_names", lambda: ["mcp-a"]) async def fake_get_skill_or_raise(_db, slug: str): assert slug == "alpha" return item captured: dict[str, list[str] | str | None] = {} class FakeRepo: def __init__(self, _db): pass async def list_all(self): return [ item, Skill( slug="beta", name="beta", description="beta", dir_path="skills/beta", tool_dependencies=[], mcp_dependencies=[], skill_dependencies=[], ), ] async def update_dependencies( self, _item: Skill, *, tool_dependencies: list[str], mcp_dependencies: list[str], skill_dependencies: list[str], updated_by: str | None, ): captured["tool_dependencies"] = tool_dependencies captured["mcp_dependencies"] = mcp_dependencies captured["skill_dependencies"] = skill_dependencies captured["updated_by"] = updated_by _item.tool_dependencies = tool_dependencies _item.mcp_dependencies = mcp_dependencies _item.skill_dependencies = skill_dependencies return _item monkeypatch.setattr(svc, "get_skill_or_raise", fake_get_skill_or_raise) monkeypatch.setattr(svc, "SkillRepository", FakeRepo) updated = await svc.update_skill_dependencies( None, slug="alpha", tool_dependencies=["calculator", "calculator"], mcp_dependencies=["mcp-a", "mcp-a"], skill_dependencies=["beta", "beta"], updated_by="root", ) assert captured["tool_dependencies"] == ["calculator"] assert captured["mcp_dependencies"] == ["mcp-a"] assert captured["skill_dependencies"] == ["beta"] assert captured["updated_by"] == "root" assert updated.skill_dependencies == ["beta"]