462 lines
15 KiB
Python
462 lines
15 KiB
Python
"""Tests for skill upload validation + install via the storage layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway.routers import skills as skills_router
|
|
from deerflow.persistence.skills import MemorySkillStore
|
|
from deerflow.skills.installer import SkillAlreadyExistsError
|
|
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
|
|
def _make_skill_zip(
|
|
*,
|
|
skill_name: str = "demo-skill",
|
|
description: str = "Use this skill when demoing things.",
|
|
license_value: str | None = "MIT",
|
|
allowed_tools: list[str] | None = None,
|
|
extra_files: dict[str, str] | None = None,
|
|
include_skill_md: bool = True,
|
|
bad_yaml: bool = False,
|
|
nest_under_root: bool = True,
|
|
) -> bytes:
|
|
"""Build an in-memory .skill ZIP archive for tests."""
|
|
buf = io.BytesIO()
|
|
root = f"{skill_name}/" if nest_under_root else ""
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
if include_skill_md:
|
|
if bad_yaml:
|
|
content = "---\nname: ???: bad\n---\nhi"
|
|
else:
|
|
lines = ["---", f"name: {skill_name}", f"description: {description}"]
|
|
if license_value:
|
|
lines.append(f"license: {license_value}")
|
|
if allowed_tools:
|
|
lines.append("allowed-tools:")
|
|
for tool in allowed_tools:
|
|
lines.append(f" - {tool}")
|
|
lines.extend(["---", "# " + skill_name, "Hello world."])
|
|
content = "\n".join(lines)
|
|
zf.writestr(f"{root}SKILL.md", content)
|
|
for rel, body in (extra_files or {}).items():
|
|
zf.writestr(f"{root}{rel}", body)
|
|
return buf.getvalue()
|
|
|
|
|
|
@pytest.fixture
|
|
def storage(tmp_path: Path) -> LocalSkillStorage:
|
|
skills_root = tmp_path / "skills"
|
|
(skills_root / "public").mkdir(parents=True)
|
|
(skills_root / "custom").mkdir(parents=True)
|
|
return LocalSkillStorage(host_path=str(skills_root))
|
|
|
|
|
|
def _write_archive(tmp_path: Path, data: bytes, *, suffix: str = ".skill") -> Path:
|
|
archive = tmp_path / f"upload{suffix}"
|
|
archive.write_bytes(data)
|
|
return archive
|
|
|
|
|
|
# ---------- validate_skill_archive ----------
|
|
|
|
|
|
def test_validate_returns_metadata_and_no_conflict(storage: LocalSkillStorage, tmp_path: Path):
|
|
archive = _write_archive(
|
|
tmp_path,
|
|
_make_skill_zip(allowed_tools=["bash", "read_file"]),
|
|
)
|
|
|
|
result = storage.validate_skill_archive(archive)
|
|
|
|
assert result["ok"] is True
|
|
assert result["skill_name"] == "demo-skill"
|
|
assert result["description"] == "Use this skill when demoing things."
|
|
assert result["license"] == "MIT"
|
|
assert result["allowed_tools"] == ["bash", "read_file"]
|
|
assert result["conflict"] is False
|
|
assert result["conflict_kind"] is None
|
|
assert result["warnings"] == []
|
|
|
|
|
|
def test_validate_flags_script_files_as_warning(storage: LocalSkillStorage, tmp_path: Path):
|
|
archive = _write_archive(
|
|
tmp_path,
|
|
_make_skill_zip(extra_files={"scripts/hi.py": "print('hi')\n"}),
|
|
)
|
|
|
|
result = storage.validate_skill_archive(archive)
|
|
|
|
assert result["ok"] is True
|
|
assert result["warnings"] and "script" in result["warnings"][0].lower()
|
|
|
|
|
|
def test_validate_detects_custom_conflict(storage: LocalSkillStorage, tmp_path: Path):
|
|
# Pre-install one with the same name into custom/
|
|
target = storage.get_skills_root_path() / "custom" / "demo-skill"
|
|
target.mkdir(parents=True)
|
|
(target / "SKILL.md").write_text(
|
|
"---\nname: demo-skill\ndescription: existing\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
archive = _write_archive(tmp_path, _make_skill_zip())
|
|
result = storage.validate_skill_archive(archive)
|
|
|
|
assert result["conflict"] is True
|
|
assert result["conflict_kind"] == "custom"
|
|
|
|
|
|
def test_validate_detects_public_conflict(storage: LocalSkillStorage, tmp_path: Path):
|
|
target = storage.get_skills_root_path() / "public" / "demo-skill"
|
|
target.mkdir(parents=True)
|
|
(target / "SKILL.md").write_text(
|
|
"---\nname: demo-skill\ndescription: existing\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
archive = _write_archive(tmp_path, _make_skill_zip())
|
|
result = storage.validate_skill_archive(archive)
|
|
|
|
assert result["conflict"] is True
|
|
assert result["conflict_kind"] == "public"
|
|
|
|
|
|
def test_validate_rejects_bad_zip(storage: LocalSkillStorage, tmp_path: Path):
|
|
archive = _write_archive(tmp_path, b"not a zip at all")
|
|
with pytest.raises(ValueError, match="not a valid ZIP"):
|
|
storage.validate_skill_archive(archive)
|
|
|
|
|
|
def test_validate_rejects_missing_skill_md(storage: LocalSkillStorage, tmp_path: Path):
|
|
archive = _write_archive(tmp_path, _make_skill_zip(include_skill_md=False))
|
|
with pytest.raises(ValueError, match="SKILL.md"):
|
|
storage.validate_skill_archive(archive)
|
|
|
|
|
|
def test_validate_rejects_bad_frontmatter(storage: LocalSkillStorage, tmp_path: Path):
|
|
archive = _write_archive(tmp_path, _make_skill_zip(bad_yaml=True))
|
|
with pytest.raises(ValueError):
|
|
storage.validate_skill_archive(archive)
|
|
|
|
|
|
# ---------- ainstall_skill_from_archive (overwrite) ----------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_writes_to_custom_dir(
|
|
storage: LocalSkillStorage, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
# Bypass the LLM security scan so the test stays hermetic.
|
|
async def _noop_scan(*args, **kwargs): # noqa: ANN001, ARG001
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"deerflow.skills.installer._scan_skill_archive_contents_or_raise",
|
|
_noop_scan,
|
|
)
|
|
|
|
archive = _write_archive(tmp_path, _make_skill_zip())
|
|
result = await storage.ainstall_skill_from_archive(archive, require_skill_suffix=False)
|
|
|
|
assert result["success"] is True
|
|
assert result["skill_name"] == "demo-skill"
|
|
target = storage.get_skills_root_path() / "custom" / "demo-skill" / "SKILL.md"
|
|
assert target.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_conflict_without_overwrite(
|
|
storage: LocalSkillStorage, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
async def _noop_scan(*args, **kwargs): # noqa: ANN001, ARG001
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"deerflow.skills.installer._scan_skill_archive_contents_or_raise",
|
|
_noop_scan,
|
|
)
|
|
|
|
archive = _write_archive(tmp_path, _make_skill_zip())
|
|
await storage.ainstall_skill_from_archive(archive, require_skill_suffix=False)
|
|
|
|
with pytest.raises(SkillAlreadyExistsError):
|
|
await storage.ainstall_skill_from_archive(archive, require_skill_suffix=False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_overwrite_replaces_existing(
|
|
storage: LocalSkillStorage, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
async def _noop_scan(*args, **kwargs): # noqa: ANN001, ARG001
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"deerflow.skills.installer._scan_skill_archive_contents_or_raise",
|
|
_noop_scan,
|
|
)
|
|
|
|
(tmp_path / "v1").mkdir(exist_ok=True)
|
|
(tmp_path / "v2").mkdir(exist_ok=True)
|
|
archive_v1 = _write_archive(tmp_path / "v1", _make_skill_zip(description="version one"))
|
|
archive_v2 = _write_archive(tmp_path / "v2", _make_skill_zip(description="version two"))
|
|
|
|
await storage.ainstall_skill_from_archive(archive_v1, require_skill_suffix=False)
|
|
await storage.ainstall_skill_from_archive(
|
|
archive_v2, overwrite=True, require_skill_suffix=False
|
|
)
|
|
|
|
skill_md = (
|
|
storage.get_skills_root_path() / "custom" / "demo-skill" / "SKILL.md"
|
|
).read_text(encoding="utf-8")
|
|
assert "version two" in skill_md
|
|
assert "version one" not in skill_md
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_refuses_to_overwrite_public_skill(
|
|
storage: LocalSkillStorage, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
async def _noop_scan(*args, **kwargs): # noqa: ANN001, ARG001
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"deerflow.skills.installer._scan_skill_archive_contents_or_raise",
|
|
_noop_scan,
|
|
)
|
|
|
|
public_dir = storage.get_skills_root_path() / "public" / "demo-skill"
|
|
public_dir.mkdir(parents=True)
|
|
(public_dir / "SKILL.md").write_text(
|
|
"---\nname: demo-skill\ndescription: built-in\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
archive = _write_archive(tmp_path, _make_skill_zip())
|
|
with pytest.raises(SkillAlreadyExistsError, match="public"):
|
|
await storage.ainstall_skill_from_archive(
|
|
archive, overwrite=True, require_skill_suffix=False
|
|
)
|
|
|
|
|
|
# ---------- ownership-aware direct upload conflicts ----------
|
|
|
|
|
|
def _request(user_id: str, *, is_admin: bool = False) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
user=SimpleNamespace(
|
|
id=user_id,
|
|
system_role="admin" if is_admin else "user",
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
async def _attach_test_owner_names(responses):
|
|
for response in responses:
|
|
if response.owner_user_id == "user-b":
|
|
response.owner_username = "bob"
|
|
return responses
|
|
|
|
|
|
class _UploadStorage:
|
|
def __init__(self) -> None:
|
|
self.install_calls = 0
|
|
|
|
def validate_skill_archive(self, _path: Path) -> dict:
|
|
return {
|
|
"ok": True,
|
|
"skill_name": "demo-skill",
|
|
"description": "demo",
|
|
"license": None,
|
|
"allowed_tools": [],
|
|
"conflict": True,
|
|
"conflict_kind": "custom",
|
|
"warnings": [],
|
|
}
|
|
|
|
async def ainstall_skill_from_archive(self, *_args, **_kwargs) -> dict:
|
|
self.install_calls += 1
|
|
return {
|
|
"success": True,
|
|
"skill_name": "demo-skill",
|
|
"message": "installed",
|
|
"overwritten": True,
|
|
"category": "custom",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_preflight_allows_owner_to_overwrite(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "demo-skill", "owner_user_id": "user-a"})
|
|
monkeypatch.setattr(skills_router, "_attach_owner_usernames", _attach_test_owner_names)
|
|
|
|
result = await skills_router._preview_install_target(
|
|
request=_request("user-a"),
|
|
skill_store=store,
|
|
skill_name="demo-skill",
|
|
target_category=SkillCategory.CUSTOM,
|
|
conflict_kind="custom",
|
|
)
|
|
|
|
can_install_as_is, can_overwrite, owned_by_me, owned_by_other, *_rest = result
|
|
assert can_install_as_is is False
|
|
assert can_overwrite is True
|
|
assert owned_by_me is True
|
|
assert owned_by_other is False
|
|
assert "upload it again and overwrite" in result[-1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_preflight_names_the_other_owner(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "demo-skill", "owner_user_id": "user-b"})
|
|
monkeypatch.setattr(skills_router, "_attach_owner_usernames", _attach_test_owner_names)
|
|
|
|
result = await skills_router._preview_install_target(
|
|
request=_request("user-a"),
|
|
skill_store=store,
|
|
skill_name="demo-skill",
|
|
target_category=SkillCategory.CUSTOM,
|
|
conflict_kind="custom",
|
|
)
|
|
|
|
can_install_as_is, can_overwrite, owned_by_me, owned_by_other = result[:4]
|
|
assert can_install_as_is is False
|
|
assert can_overwrite is False
|
|
assert owned_by_me is False
|
|
assert owned_by_other is True
|
|
assert result[6] == "bob"
|
|
assert "user 'bob'" in result[-1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_upload_surfaces_the_other_owner(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "demo-skill", "owner_user_id": "user-b"})
|
|
storage = _UploadStorage()
|
|
upload_path = tmp_path / "validate.skill"
|
|
upload_path.write_bytes(b"placeholder")
|
|
|
|
async def _persist(_file) -> Path:
|
|
return upload_path
|
|
|
|
monkeypatch.setattr(skills_router, "_persist_uploaded_skill", _persist)
|
|
monkeypatch.setattr(
|
|
skills_router,
|
|
"get_or_new_skill_storage",
|
|
lambda **_kwargs: storage,
|
|
)
|
|
monkeypatch.setattr(
|
|
skills_router,
|
|
"_attach_owner_usernames",
|
|
_attach_test_owner_names,
|
|
)
|
|
|
|
response = await skills_router.validate_skill_upload(
|
|
request=_request("user-a"),
|
|
file=object(),
|
|
target_category=SkillCategory.CUSTOM,
|
|
config=object(),
|
|
skill_store=store,
|
|
)
|
|
|
|
assert response.conflict is True
|
|
assert response.conflict_kind == "custom"
|
|
assert response.can_overwrite is False
|
|
assert response.owned_by_other is True
|
|
assert response.owner_username == "bob"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_upload_owner_can_overwrite_own_skill(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "demo-skill", "owner_user_id": "user-a"})
|
|
storage = _UploadStorage()
|
|
upload_path = tmp_path / "owner-upload.skill"
|
|
upload_path.write_bytes(b"placeholder")
|
|
|
|
async def _persist(_file) -> Path:
|
|
return upload_path
|
|
|
|
async def _refresh() -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(skills_router, "_persist_uploaded_skill", _persist)
|
|
monkeypatch.setattr(
|
|
skills_router,
|
|
"get_or_new_skill_storage",
|
|
lambda **_kwargs: storage,
|
|
)
|
|
monkeypatch.setattr(
|
|
skills_router,
|
|
"refresh_skills_system_prompt_cache_async",
|
|
_refresh,
|
|
)
|
|
monkeypatch.setattr("deerflow.skills.usage.mark_uploaded", lambda _name: None)
|
|
|
|
response = await skills_router.install_skill_upload(
|
|
request=_request("user-a"),
|
|
file=object(),
|
|
overwrite=True,
|
|
target_category=SkillCategory.CUSTOM,
|
|
config=object(),
|
|
skill_store=store,
|
|
)
|
|
|
|
assert response.success is True
|
|
assert response.overwritten is True
|
|
assert storage.install_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_upload_rechecks_owner_before_overwrite(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "demo-skill", "owner_user_id": "user-b"})
|
|
storage = _UploadStorage()
|
|
upload_path = tmp_path / "upload.skill"
|
|
upload_path.write_bytes(b"placeholder")
|
|
|
|
async def _persist(_file) -> Path:
|
|
return upload_path
|
|
|
|
monkeypatch.setattr(skills_router, "_persist_uploaded_skill", _persist)
|
|
monkeypatch.setattr(skills_router, "get_or_new_skill_storage", lambda **_kwargs: storage)
|
|
monkeypatch.setattr(skills_router, "_attach_owner_usernames", _attach_test_owner_names)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await skills_router.install_skill_upload(
|
|
request=_request("user-a", is_admin=True),
|
|
file=object(),
|
|
overwrite=True,
|
|
target_category=SkillCategory.CUSTOM,
|
|
config=object(),
|
|
skill_store=store,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert "user 'bob'" in str(exc_info.value.detail)
|
|
assert storage.install_calls == 0
|