吉鑫升andClaude Opus 4.7
80fc2a9240
- 后端新增 POST /api/skills/reconcile-custom:扫描 custom 目录,把磁盘存在但 DB 无记录的技能登记到调用者名下(published=false),附 4 个单元测试 - 前端技能列表加"重新扫描技能目录"按钮,列表分页每页 10 条 - SkillsConfig 新增 security_scan.enabled(默认 true)用于关闭 LLM 安全扫描;scan_skill_content 在开关关闭时直接放行 - 智能体卡片/对话页长名 truncate 显示 + loading 占位 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
"""Tests for the ``POST /api/skills/reconcile-custom`` adoption flow.
|
|
|
|
Covers the scenario where a user drops a skill directory onto disk by hand
|
|
(bypassing the upload API) and then calls the reconcile route so it shows up
|
|
in the UI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import skills as skills_router
|
|
from deerflow.persistence.skills import MemorySkillStore
|
|
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
|
|
|
|
|
|
def _write_skill_on_disk(root: Path, name: str, description: str = "from disk") -> None:
|
|
"""Drop a valid SKILL.md into ``<root>/custom/<name>/`` like a manual unzip would."""
|
|
skill_dir = root / "custom" / name
|
|
skill_dir.mkdir(parents=True, exist_ok=True)
|
|
(skill_dir / "SKILL.md").write_text(
|
|
f"---\nname: {name}\ndescription: {description}\n---\n# {name}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _fake_request(user_id: str | None = "user-a") -> SimpleNamespace:
|
|
user = SimpleNamespace(id=user_id, system_role="user") if user_id else None
|
|
return SimpleNamespace(state=SimpleNamespace(user=user))
|
|
|
|
|
|
@pytest.fixture
|
|
def skills_root(tmp_path: Path) -> Path:
|
|
root = tmp_path / "skills"
|
|
(root / "public").mkdir(parents=True)
|
|
(root / "custom").mkdir(parents=True)
|
|
return root
|
|
|
|
|
|
@pytest.fixture
|
|
def patched_storage(
|
|
skills_root: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> LocalSkillStorage:
|
|
"""Force the router to use a LocalSkillStorage rooted at ``skills_root``."""
|
|
storage = LocalSkillStorage(host_path=str(skills_root))
|
|
monkeypatch.setattr(
|
|
skills_router, "get_or_new_skill_storage", lambda **_: storage
|
|
)
|
|
|
|
async def _noop_refresh() -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
skills_router, "refresh_skills_system_prompt_cache_async", _noop_refresh
|
|
)
|
|
return storage
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconcile_adopts_orphan_custom_skill(
|
|
skills_root: Path, patched_storage: LocalSkillStorage
|
|
):
|
|
_write_skill_on_disk(skills_root, "manually-dropped")
|
|
store = MemorySkillStore()
|
|
|
|
result = await skills_router.reconcile_custom_skills(
|
|
request=_fake_request("user-a"), config=None, skill_store=store
|
|
)
|
|
|
|
assert result.adopted == ["manually-dropped"]
|
|
assert result.already_registered == []
|
|
|
|
record = await store.get_any("manually-dropped")
|
|
assert record is not None
|
|
assert record["owner_user_id"] == "user-a"
|
|
assert record["published"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconcile_skips_already_registered(
|
|
skills_root: Path, patched_storage: LocalSkillStorage
|
|
):
|
|
_write_skill_on_disk(skills_root, "already-known")
|
|
store = MemorySkillStore()
|
|
# Pre-existing ownership row — should be left alone.
|
|
await store.create(
|
|
{"name": "already-known", "owner_user_id": "user-b", "published": True}
|
|
)
|
|
|
|
result = await skills_router.reconcile_custom_skills(
|
|
request=_fake_request("user-a"), config=None, skill_store=store
|
|
)
|
|
|
|
assert result.adopted == []
|
|
assert result.already_registered == ["already-known"]
|
|
|
|
# Ownership must not be overwritten.
|
|
record = await store.get_any("already-known")
|
|
assert record is not None
|
|
assert record["owner_user_id"] == "user-b"
|
|
assert record["published"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconcile_mixed_adopts_only_orphans(
|
|
skills_root: Path, patched_storage: LocalSkillStorage
|
|
):
|
|
_write_skill_on_disk(skills_root, "orphan-one")
|
|
_write_skill_on_disk(skills_root, "orphan-two")
|
|
_write_skill_on_disk(skills_root, "registered")
|
|
store = MemorySkillStore()
|
|
await store.create(
|
|
{"name": "registered", "owner_user_id": "user-a", "published": False}
|
|
)
|
|
|
|
result = await skills_router.reconcile_custom_skills(
|
|
request=_fake_request("user-a"), config=None, skill_store=store
|
|
)
|
|
|
|
assert sorted(result.adopted) == ["orphan-one", "orphan-two"]
|
|
assert result.already_registered == ["registered"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconcile_with_empty_custom_dir_is_noop(
|
|
skills_root: Path, patched_storage: LocalSkillStorage
|
|
):
|
|
store = MemorySkillStore()
|
|
|
|
result = await skills_router.reconcile_custom_skills(
|
|
request=_fake_request("user-a"), config=None, skill_store=store
|
|
)
|
|
|
|
assert result.adopted == []
|
|
assert result.already_registered == []
|