from __future__ import annotations import asyncio import io import zipfile from pathlib import Path from types import SimpleNamespace import pytest from yuxi.agents.skills import service as svc from yuxi.agents.toolkits import service as tool_service from yuxi.storage.postgres.models_business import Skill, User 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 _user(uid: str = "root", role: str = "admin") -> User: return User(username=uid, uid=uid, password_hash="x", role=role, department_id=1) def test_allowed_skill_access_levels_by_role(): assert svc.get_allowed_skill_access_levels(_user(role="user")) == ["user"] assert svc.get_allowed_skill_access_levels(_user(role="admin")) == ["global", "department", "user"] assert svc.get_allowed_skill_access_levels(_user(role="superadmin")) == ["global", "department", "user"] @pytest.mark.asyncio async def test_list_visible_skills_for_management_includes_owned_disabled_and_enabled_shared( monkeypatch: pytest.MonkeyPatch, ): items = [ Skill(slug="owned-disabled", name="owned-disabled", description="", created_by="root", enabled=False), Skill( slug="shared-enabled", name="shared-enabled", description="", created_by="other", enabled=True, share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, ), Skill( slug="shared-disabled", name="shared-disabled", description="", created_by="other", enabled=False, share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, ), Skill(slug="unrelated", name="unrelated", description="", created_by="other", enabled=True), ] class FakeRepo: def __init__(self, _db): pass async def list_all(self): return items monkeypatch.setattr(svc, "SkillRepository", FakeRepo) visible = await svc.list_visible_skills_for_management(None, _user("root", role="user")) assert [item.slug for item in visible] == ["owned-disabled", "shared-enabled"] @pytest.mark.asyncio @pytest.mark.parametrize( "skill,operator", [ ( Skill(slug="owned-disabled", name="owned-disabled", description="", created_by="root", enabled=False), _user("root", role="user"), ), ( Skill(slug="admin-disabled", name="admin-disabled", description="", created_by="other", enabled=False), _user("root", role="admin"), ), ( Skill( slug="shared-enabled", name="shared-enabled", description="", created_by="other", enabled=True, share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, ), _user("root", role="user"), ), ], ) async def test_management_readable_skill_allows_manageable_disabled_and_enabled_shared( monkeypatch: pytest.MonkeyPatch, skill: Skill, operator: User, ): class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str): assert slug == skill.slug return skill monkeypatch.setattr(svc, "SkillRepository", FakeRepo) result = await svc.get_management_readable_skill_or_raise(None, operator, skill.slug) assert result is skill @pytest.mark.asyncio async def test_management_readable_skill_rejects_disabled_shared_readonly(monkeypatch: pytest.MonkeyPatch): skill = Skill( slug="shared-disabled", name="shared-disabled", description="", created_by="other", enabled=False, share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, ) class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str): assert slug == skill.slug return skill monkeypatch.setattr(svc, "SkillRepository", FakeRepo) with pytest.raises(ValueError, match="不存在或无权访问"): await svc.get_management_readable_skill_or_raise(None, _user("root", role="user"), skill.slug) @pytest.mark.asyncio async def test_runtime_access_still_excludes_disabled_shared_skill(monkeypatch: pytest.MonkeyPatch): skill = Skill( slug="shared-disabled", name="shared-disabled", description="", created_by="other", enabled=False, share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, ) class FakeRepo: def __init__(self, _db): pass async def list_enabled(self): return [] monkeypatch.setattr(svc, "SkillRepository", FakeRepo) assert svc.user_can_access_skill(_user("root", role="user"), skill) is False assert await svc.list_accessible_skills(None, _user("root", role="user")) == [] @pytest.mark.asyncio async def test_normal_user_skill_upload_draft_defaults_to_user_share(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: def __init__(self, _db): pass async def exists_slug(self, _slug: str) -> bool: return False monkeypatch.setattr(svc, "SkillRepository", FakeRepo) draft = await svc.prepare_skill_upload( None, filename="SKILL.md", file_bytes=b"---\nname: demo\ndescription: demo skill\n---\n# Demo\n", operator=_user("normal-user", role="user"), ) assert draft["default_share_config"] == { "access_level": "user", "department_ids": [], "user_uids": ["normal-user"], } assert draft["allowed_access_levels"] == ["user"] @pytest.mark.parametrize( "share_config", [ {"access_level": "global", "department_ids": [], "user_uids": []}, {"access_level": "department", "department_ids": [1], "user_uids": []}, ], ) @pytest.mark.asyncio async def test_normal_user_confirm_skill_draft_rejects_wider_share_scope( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, share_config: dict, ): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: def __init__(self, _db): pass async def exists_slug(self, _slug: str) -> bool: return False async def create(self, **_kwargs) -> Skill: raise AssertionError("普通用户的越权共享范围应在创建前被拒绝") monkeypatch.setattr(svc, "SkillRepository", FakeRepo) operator = _user("normal-user", role="user") draft = await svc.prepare_skill_upload( None, filename="SKILL.md", file_bytes=b"---\nname: demo\ndescription: demo skill\n---\n# Demo\n", operator=operator, ) with pytest.raises(ValueError, match="无权使用该 Skill 共享范围"): await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=share_config, operator=operator, ) def test_parse_skill_markdown_ok(): content = "---\nname: demo-skill\ndescription: demo description\n---\n# Demo\n" slug, name, desc, meta = svc._parse_skill_markdown(content) assert slug == "demo-skill" assert name == "demo-skill" assert desc == "demo description" assert meta["name"] == "demo-skill" def test_parse_skill_markdown_supports_display_name_with_slug(): content = ( "---\n" "name: Word / DOCX\n" "slug: word-docx\n" "version: 1.0.2\n" "homepage: https://clawic.com/skills/word-docx\n" "description: Create, inspect, and edit Microsoft Word documents.\n" 'metadata: {"clawdbot":{"emoji":"📘","os":["linux","darwin","win32"]}}\n' "---\n" "# Word / DOCX\n" ) slug, name, desc, meta = svc._parse_skill_markdown(content) assert slug == "word-docx" assert name == "Word / DOCX" assert desc == "Create, inspect, and edit Microsoft Word documents." assert meta["version"] == "1.0.2" def test_parse_skill_markdown_requires_frontmatter(): with pytest.raises(ValueError, match="frontmatter"): svc._parse_skill_markdown("# missing") def test_image_gen_builtin_skill_spec(): specs = {spec["slug"]: spec for spec in svc.list_builtin_skill_specs()} assert "image-gen" in specs image_gen = specs["image-gen"] assert image_gen["name"] == "image-gen" assert image_gen["tool_dependencies"] == ["present_artifacts"] assert (image_gen["source_dir"] / "SKILL.md").exists() def test_knowledge_base_builtin_skill_spec(): specs = {spec["slug"]: spec for spec in svc.list_builtin_skill_specs()} assert "knowledge-base" in specs knowledge_base = specs["knowledge-base"] assert knowledge_base["name"] == "knowledge-base" assert knowledge_base["tool_dependencies"] == [ "list_kbs", "query_kb", "find_kb_document", "open_kb_document", "get_mindmap", "search_file", ] assert (knowledge_base["source_dir"] / "SKILL.md").exists() def test_mysql_reporter_builtin_skill_spec_replaces_reporter_and_deep_reporter(): specs = {spec["slug"]: spec for spec in svc.list_builtin_skill_specs()} assert "reporter" not in specs assert "deep-reporter" not in specs assert "mysql-reporter" in specs mysql_reporter = specs["mysql-reporter"] assert mysql_reporter["name"] == "mysql reporter" assert mysql_reporter["tool_dependencies"] == [] assert mysql_reporter["mcp_dependencies"] == ["mcp-server-chart"] assert (mysql_reporter["source_dir"] / "SKILL.md").exists() for script_name in ("list_tables.py", "describe_table.py", "query.py"): script_path = mysql_reporter["source_dir"] / "scripts" / script_name assert script_path.exists() assert script_path.read_text(encoding="utf-8").startswith("# /// script") def test_is_valid_skill_slug(): # Test valid slugs assert svc.is_valid_skill_slug("demo-skill") is True assert svc.is_valid_skill_slug("valid-name-123") is True # Test invalid slugs assert svc.is_valid_skill_slug("../bad") is False assert svc.is_valid_skill_slug("Invalid") is False # uppercase not allowed assert svc.is_valid_skill_slug("") is False def test_sync_thread_readable_skills_none_keeps_no_skills(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) skills_root = tmp_path / "skills" (skills_root / "alpha").mkdir(parents=True, exist_ok=True) (skills_root / "alpha" / "SKILL.md").write_text("alpha", encoding="utf-8") thread_root = svc.sync_thread_readable_skills("thread_1", None) assert list(thread_root.iterdir()) == [] def test_sync_thread_readable_skills_only_keeps_selected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) skills_root = tmp_path / "skills" (skills_root / "alpha").mkdir(parents=True, exist_ok=True) (skills_root / "alpha" / "SKILL.md").write_text("alpha", encoding="utf-8") (skills_root / "beta").mkdir(parents=True, exist_ok=True) (skills_root / "beta" / "SKILL.md").write_text("beta", encoding="utf-8") thread_root = svc.sync_thread_readable_skills("thread_1", ["alpha", "missing", "alpha"]) assert thread_root == tmp_path / "threads" / "thread_1" / "skills" assert sorted(path.name for path in thread_root.iterdir()) == ["alpha"] assert (thread_root / "alpha").is_dir() assert not (thread_root / "alpha").is_symlink() assert (thread_root / "alpha" / "SKILL.md").read_text(encoding="utf-8") == "alpha" svc.sync_thread_readable_skills("thread_1", ["beta"]) assert sorted(path.name for path in thread_root.iterdir()) == ["beta"] assert (thread_root / "beta" / "SKILL.md").read_text(encoding="utf-8") == "beta" @pytest.mark.asyncio async def test_get_skill_dependency_options(monkeypatch: pytest.MonkeyPatch): # Mock get_tool_metadata to return tool list def fake_get_tool_metadata(category=None): return [ {"slug": "calculator", "name": "Calculator"}, {"slug": "search", "name": "Search"}, ] monkeypatch.setattr(tool_service, "get_tool_metadata", fake_get_tool_metadata) async def fake_get_enabled_mcp_server_slugs(db=None): del db return ["mcp-a", "mcp-b"] monkeypatch.setattr(svc, "get_enabled_mcp_server_slugs", fake_get_enabled_mcp_server_slugs) user = SimpleNamespace(uid="user") async def fake_list_skill_slugs(_db, *, user): assert user.uid == "user" return ["alpha", "beta"] monkeypatch.setattr(svc, "list_skill_slugs", fake_list_skill_slugs) result = await svc.get_skill_dependency_options(None, user) assert result["tools"] == [{"slug": "calculator", "name": "Calculator"}, {"slug": "search", "name": "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_skill_upload_prepare_confirm_rewrites_conflicting_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, **kwargs) -> Skill: item = Skill(**kwargs, updated_by=kwargs["created_by"]) self.__class__.existing_slugs.add(item.slug) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "demo/SKILL.md": ("---\nname: demo\ndescription: this is demo\n---\n# Demo\n"), "demo/prompts/system.md": "You are demo skill", } ) operator = _user("root") draft = await svc.prepare_skill_upload( None, filename="demo.zip", file_bytes=zip_bytes, operator=operator, ) results = await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=draft["default_share_config"], operator=operator, ) assert results[0]["slug"] == "demo-v2" assert results[0]["success"] is True assert FakeRepo.created_item.slug == "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_skill_zip_import_uses_skill_md_name_not_zip_or_root_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: created_item: Skill | None = None def __init__(self, _db): pass async def exists_slug(self, _slug: str) -> bool: return False async def create(self, **kwargs) -> Skill: item = Skill(**kwargs, updated_by=kwargs["created_by"]) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "Bad--Archive-Name/SKILL.md": ("---\nname: valid-skill\ndescription: this is valid\n---\n# Valid\n"), "Bad--Archive-Name/prompts/system.md": "Use valid skill metadata.", } ) operator = _user("root") draft = await svc.prepare_skill_upload( None, filename="Bad--Archive-Name.zip", file_bytes=zip_bytes, operator=operator, ) results = await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=draft["default_share_config"], operator=operator, ) assert draft["items"][0]["original_name"] == "valid-skill" assert draft["items"][0]["slug"] == "valid-skill" assert results[0]["success"] is True assert results[0]["slug"] == "valid-skill" assert FakeRepo.created_item.slug == "valid-skill" assert (tmp_path / "skills" / "valid-skill" / "SKILL.md").exists() @pytest.mark.asyncio async def test_skill_zip_import_validates_skill_md_name_not_zip_filename( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: def __init__(self, _db): pass async def exists_slug(self, _slug: str) -> bool: return False monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "valid-archive/SKILL.md": ("---\nname: invalid--skill\ndescription: invalid name\n---\n# Invalid\n"), } ) with pytest.raises(ValueError, match="SKILL.md frontmatter.name 必须是小写字母/数字/短横线"): await svc.prepare_skill_upload( None, filename="valid-archive.zip", file_bytes=zip_bytes, operator=_user("root"), ) @pytest.mark.asyncio async def test_skill_zip_import_uses_frontmatter_slug_and_keeps_display_name( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: created_item: Skill | None = None def __init__(self, _db): pass async def exists_slug(self, _slug: str) -> bool: return False async def create(self, **kwargs) -> Skill: item = Skill(**kwargs, updated_by=kwargs["created_by"]) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "Word Skill/SKILL.md": ( "---\n" "name: Word / DOCX\n" "slug: word-docx\n" "version: 1.0.2\n" "homepage: https://clawic.com/skills/word-docx\n" "description: Create, inspect, and edit Microsoft Word documents.\n" "changelog: Tightened review workflows.\n" 'metadata: {"clawdbot":{"emoji":"📘","os":["linux","darwin","win32"]}}\n' "---\n" "# Word / DOCX\n" ) } ) operator = _user("root") draft = await svc.prepare_skill_upload( None, filename="Word Skill.zip", file_bytes=zip_bytes, operator=operator, ) results = await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=draft["default_share_config"], operator=operator, ) assert draft["items"][0]["slug"] == "word-docx" assert draft["items"][0]["name"] == "Word / DOCX" assert results[0]["success"] is True assert results[0]["slug"] == "word-docx" assert FakeRepo.created_item.slug == "word-docx" assert FakeRepo.created_item.name == "Word / DOCX" @pytest.mark.asyncio async def test_skill_zip_import_rewrites_conflicting_slug_not_display_name( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: existing_slugs = {"word-docx"} 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, **kwargs) -> Skill: item = Skill(**kwargs, updated_by=kwargs["created_by"]) self.__class__.existing_slugs.add(item.slug) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) zip_bytes = _build_zip( { "Word Skill/SKILL.md": ( "---\n" "name: Word / DOCX\n" "slug: word-docx\n" "description: Create, inspect, and edit Microsoft Word documents.\n" "---\n" "# Word / DOCX\n" ) } ) operator = _user("root") draft = await svc.prepare_skill_upload( None, filename="Word Skill.zip", file_bytes=zip_bytes, operator=operator, ) results = await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=draft["default_share_config"], operator=operator, ) assert results[0]["slug"] == "word-docx-v2" assert results[0]["success"] is True assert FakeRepo.created_item.name == "Word / DOCX" skill_md = (tmp_path / "skills" / "word-docx-v2" / "SKILL.md").read_text(encoding="utf-8") assert "name: Word / DOCX" in skill_md assert "slug: word-docx-v2" in skill_md @pytest.mark.asyncio async def test_skill_md_prepare_confirm_creates_single_file_skill(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) class FakeRepo: created_item: Skill | None = None def __init__(self, _db): pass async def exists_slug(self, slug: str) -> bool: return False async def create(self, **kwargs) -> Skill: item = Skill(**kwargs, updated_by=kwargs["created_by"]) self.__class__.created_item = item return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) skill_md = "---\nname: demo\ndescription: this is demo\n---\n# Demo\n" operator = _user("root") draft = await svc.prepare_skill_upload( None, filename="SKILL.md", file_bytes=skill_md.encode("utf-8"), operator=operator, ) results = await svc.confirm_skill_install_draft( None, draft_id=draft["draft_id"], share_config=draft["default_share_config"], operator=operator, ) assert results[0]["slug"] == "demo" assert results[0]["success"] is True assert FakeRepo.created_item.name == "demo" assert (tmp_path / "skills" / "demo" / "SKILL.md").read_text(encoding="utf-8") == skill_md @pytest.mark.asyncio async def test_import_skill_dir_requires_root_skill_md(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) source_dir = tmp_path / "source-skill" source_dir.mkdir(parents=True, exist_ok=True) with pytest.raises(ValueError, match="根级 SKILL.md"): await svc.import_skill_dir( None, source_dir=source_dir, created_by="root", ) @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 = "---\nname: demo\ndescription: 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", source_type="upload", dir_path="skills/alpha", share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, enabled=True, tool_dependencies=[], mcp_dependencies=[], skill_dependencies=[], ) dependency = Skill( slug="beta", name="beta", description="beta", source_type="upload", dir_path="skills/beta", share_config={"access_level": "user", "department_ids": [], "user_uids": ["root"]}, enabled=True, tool_dependencies=[], mcp_dependencies=[], skill_dependencies=[], ) # Mock get_tool_metadata to return tool list def fake_get_tool_metadata(category=None): return [{"slug": "calculator", "name": "Calculator"}] monkeypatch.setattr(tool_service, "get_tool_metadata", fake_get_tool_metadata) async def fake_get_enabled_mcp_server_slugs(db=None): del db return ["mcp-a"] monkeypatch.setattr(svc, "get_enabled_mcp_server_slugs", fake_get_enabled_mcp_server_slugs) 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, dependency] 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 async def fake_list_accessible_skills(_db, _operator): return [item, dependency] monkeypatch.setattr(svc, "get_skill_or_raise", fake_get_skill_or_raise) monkeypatch.setattr(svc, "list_accessible_skills", fake_list_accessible_skills) 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"], operator=_user("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"] @pytest.mark.asyncio async def test_init_builtin_skills_create_missing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) source_dir = tmp_path / "builtin-skills" / "reporter" source_dir.mkdir(parents=True, exist_ok=True) (source_dir / "SKILL.md").write_text( "---\nname: reporter\ndescription: SQL report\n---\n# SQL Reporter\n", encoding="utf-8", ) (source_dir / "prompts").mkdir(parents=True, exist_ok=True) (source_dir / "prompts" / "system.md").write_text("prompt", encoding="utf-8") monkeypatch.setattr( svc, "get_builtin_skill_specs", lambda: [ SimpleNamespace( slug="reporter", source_dir=source_dir, description="SQL report from python", tool_dependencies=("mysql_query",), mcp_dependencies=("charts",), skill_dependencies=("common-report",), ) ], ) class FakeRepo: created_payload: dict | None = None def __init__(self, _db): pass async def get_by_slug(self, slug: str): assert slug == "reporter" return None async def create(self, **kwargs) -> Skill: self.__class__.created_payload = kwargs return Skill(**kwargs, updated_by=kwargs["created_by"]) monkeypatch.setattr(svc, "SkillRepository", FakeRepo) items = await svc.init_builtin_skills(None) assert len(items) == 1 assert items[0].slug == "reporter" assert FakeRepo.created_payload["source_type"] == "builtin" assert FakeRepo.created_payload["share_config"] == svc.BUILTIN_SKILL_SHARE_CONFIG assert FakeRepo.created_payload["enabled"] is True assert FakeRepo.created_payload["created_by"] == "system" assert FakeRepo.created_payload["tool_dependencies"] == ["mysql_query"] assert FakeRepo.created_payload["mcp_dependencies"] == ["charts"] assert FakeRepo.created_payload["skill_dependencies"] == ["common-report"] assert (tmp_path / "skills" / "reporter" / "SKILL.md").exists() assert (tmp_path / "skills" / "reporter" / "prompts" / "system.md").read_text(encoding="utf-8") == "prompt" @pytest.mark.asyncio async def test_init_builtin_skills_updates_existing_record_and_preserves_disabled( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) source_dir = tmp_path / "builtin-skills" / "reporter" source_dir.mkdir(parents=True, exist_ok=True) (source_dir / "SKILL.md").write_text( "---\nname: reporter\ndescription: new markdown description\n---\n# SQL Reporter\n", encoding="utf-8", ) (source_dir / "prompt.md").write_text("new builtin content", encoding="utf-8") target_dir = tmp_path / "skills" / "reporter" target_dir.mkdir(parents=True, exist_ok=True) (target_dir / "prompt.md").write_text("old content", encoding="utf-8") monkeypatch.setattr( svc, "get_builtin_skill_specs", lambda: [ SimpleNamespace( slug="reporter", source_dir=source_dir, description="new description", version="1.0.1", tool_dependencies=("mysql_query",), mcp_dependencies=("charts",), skill_dependencies=(), ) ], ) existing_item = Skill( slug="reporter", name="reporter", description="old description", dir_path="skills/reporter", source_type="builtin", tool_dependencies=[], mcp_dependencies=[], skill_dependencies=[], share_config={"access_level": "global", "department_ids": [], "user_uids": []}, enabled=False, version="1.0.0", content_hash="old-hash", created_by="system", updated_by="system", ) captured: dict[str, object] = {} class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str): assert slug == "reporter" return existing_item async def update_metadata(self, item: Skill, *, name: str, description: str, updated_by: str | None) -> Skill: item.name = name item.description = description captured["metadata"] = {"name": name, "description": description, "updated_by": updated_by} return item async def update_dependencies( self, item: Skill, *, tool_dependencies: list[str], mcp_dependencies: list[str], skill_dependencies: list[str], updated_by: str | None, ) -> Skill: item.tool_dependencies = tool_dependencies item.mcp_dependencies = mcp_dependencies item.skill_dependencies = skill_dependencies captured["dependencies"] = { "tool_dependencies": tool_dependencies, "mcp_dependencies": mcp_dependencies, "skill_dependencies": skill_dependencies, "updated_by": updated_by, } return item async def update_builtin_install( self, item: Skill, *, version: str, content_hash: str, updated_by: str | None, ) -> Skill: item.version = version item.content_hash = content_hash item.source_type = "builtin" item.share_config = svc.BUILTIN_SKILL_SHARE_CONFIG.copy() item.updated_by = updated_by captured["install"] = {"version": version, "content_hash": content_hash, "updated_by": updated_by} return item monkeypatch.setattr(svc, "SkillRepository", FakeRepo) items = await svc.init_builtin_skills(None, created_by="release-bot") assert len(items) == 1 assert items[0].enabled is False assert items[0].version == "1.0.1" assert (target_dir / "prompt.md").read_text(encoding="utf-8") == "new builtin content" assert captured["metadata"] == { "name": "reporter", "description": "new description", "updated_by": "release-bot", } assert captured["dependencies"] == { "tool_dependencies": ["mysql_query"], "mcp_dependencies": ["charts"], "skill_dependencies": [], "updated_by": "release-bot", } assert captured["install"]["updated_by"] == "release-bot" @pytest.mark.asyncio async def test_init_builtin_skills_rejects_non_builtin_conflict(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) source_dir = tmp_path / "builtin" / "reporter" source_dir.mkdir(parents=True, exist_ok=True) (source_dir / "SKILL.md").write_text( "---\nname: reporter\ndescription: SQL report\n---\n# SQL Reporter\n", encoding="utf-8", ) monkeypatch.setattr( svc, "list_builtin_skill_specs", lambda: [ { "slug": "reporter", "name": "reporter", "description": "SQL report", "version": "1.0.0", "tool_dependencies": [], "mcp_dependencies": [], "skill_dependencies": [], "content_hash": "hash-v1", "source_dir": source_dir, } ], ) class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str): return Skill(slug=slug, name=slug, description="uploaded", dir_path=f"skills/{slug}", source_type="upload") monkeypatch.setattr(svc, "SkillRepository", FakeRepo) with pytest.raises(ValueError, match="非内置 skill 冲突"): await svc.init_builtin_skills(None) @pytest.mark.asyncio async def test_update_skill_enabled_allows_builtin(monkeypatch: pytest.MonkeyPatch): builtin_item = Skill( slug="reporter", name="reporter", description="builtin", dir_path="skills/reporter", source_type="builtin", enabled=True, ) async def fake_get_manageable_skill_or_raise(_db, user, slug: str): assert user.uid == "root" assert slug == "reporter" return builtin_item class FakeRepo: def __init__(self, _db): pass async def update_enabled(self, item: Skill, *, enabled: bool, updated_by: str | None): item.enabled = enabled item.updated_by = updated_by return item monkeypatch.setattr(svc, "get_manageable_skill_or_raise", fake_get_manageable_skill_or_raise) monkeypatch.setattr(svc, "SkillRepository", FakeRepo) updated = await svc.update_skill_enabled(None, slug="reporter", enabled=False, operator=_user("root")) assert updated.enabled is False assert updated.updated_by == "root" @pytest.mark.asyncio async def test_builtin_skill_file_edit_blocked(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) target_dir = tmp_path / "skills" / "reporter" target_dir.mkdir(parents=True, exist_ok=True) (target_dir / "SKILL.md").write_text( "---\nname: reporter\ndescription: builtin\n---\n# Reporter\n", encoding="utf-8", ) builtin_item = Skill( slug="reporter", name="reporter", description="builtin", dir_path="skills/reporter", source_type="builtin", ) async def fake_get_skill_or_raise(_db, _slug: str): return builtin_item monkeypatch.setattr(svc, "get_skill_or_raise", fake_get_skill_or_raise) with pytest.raises(ValueError, match="内置 skill 不允许直接修改文件"): await svc.update_skill_file( None, slug="reporter", relative_path="SKILL.md", content="new content", updated_by="root", ) @pytest.mark.asyncio async def test_delete_skills_batch_ok(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) # 模拟两个已安装的技能 (tmp_path / "skills" / "skill-a").mkdir(parents=True, exist_ok=True) (tmp_path / "skills" / "skill-b").mkdir(parents=True, exist_ok=True) item_a = Skill(slug="skill-a", name="skill-a", description="a", dir_path="skills/skill-a") item_b = Skill(slug="skill-b", name="skill-b", description="b", dir_path="skills/skill-b") db_items = {"skill-a": item_a, "skill-b": item_b} deleted_slugs = [] class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str, *, for_update: bool = False): return db_items.get(slug) async def delete(self, item: Skill): deleted_slugs.append(item.slug) db_items.pop(item.slug, None) monkeypatch.setattr(svc, "SkillRepository", FakeRepo) # 执行批量删除，skill-a, skill-b, skill-c (不存在) results = await svc.delete_skills_batch(None, slugs=["skill-a", "skill-b", "skill-c"]) assert results == [ {"slug": "skill-a", "success": True}, {"slug": "skill-b", "success": True}, {"slug": "skill-c", "success": False, "error": "技能 'skill-c' 不存在"}, ] assert deleted_slugs == ["skill-a", "skill-b"] assert not (tmp_path / "skills" / "skill-a").exists() assert not (tmp_path / "skills" / "skill-b").exists() @pytest.mark.asyncio async def test_delete_skills_batch_limit_exceeded(): slugs = [f"skill-{i}" for i in range(51)] with pytest.raises(ValueError, match="批量删除的技能数量不能超过 50 个"): await svc.delete_skills_batch(None, slugs=slugs) @pytest.mark.asyncio async def test_delete_skill_concurrent_lock(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(svc.sys_config, "save_dir", str(tmp_path)) (tmp_path / "skills" / "concurrent-skill").mkdir(parents=True, exist_ok=True) item = Skill( slug="concurrent-skill", name="concurrent-skill", description="desc", dir_path="skills/concurrent-skill" ) db_items = {"concurrent-skill": item} lock_active = asyncio.Lock() class FakeRepo: def __init__(self, _db): pass async def get_by_slug(self, slug: str, *, for_update: bool = False): # 用 asyncio.Lock 模拟 with_for_update() 的排他锁 if for_update: await lock_active.acquire() try: return db_items.get(slug) finally: pass else: return db_items.get(slug) async def delete(self, item: Skill): db_items.pop(item.slug, None) if lock_active.locked(): lock_active.release() monkeypatch.setattr(svc, "SkillRepository", FakeRepo) # 同时发起两个 delete_skill 调用 task1 = asyncio.create_task(svc.delete_skill(None, slug="concurrent-skill")) task2 = asyncio.create_task(svc.delete_skill(None, slug="concurrent-skill")) results = await asyncio.gather(task1, task2, return_exceptions=True) success_count = 0 error_count = 0 for r in results: if r is None: success_count += 1 elif isinstance(r, ValueError) and "不存在" in str(r): error_count += 1 assert success_count == 1 assert error_count == 1 assert not (tmp_path / "skills" / "concurrent-skill").exists()