242 lines
8.3 KiB
Python
242 lines
8.3 KiB
Python
"""Tests for skill upload validation + install via the storage layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from deerflow.skills.installer import SkillAlreadyExistsError
|
|
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
|
|
|
|
|
|
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
|
|
)
|