"""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 )