"""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 import parser as skill_parser 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 def test_load_skills_caches_parsing_and_invalidates_on_file_change( storage: LocalSkillStorage, monkeypatch: pytest.MonkeyPatch, ) -> None: skill_dir = storage.get_skills_root_path() / "public" / "cached-skill" skill_dir.mkdir(parents=True) skill_file = skill_dir / "SKILL.md" skill_file.write_text( "---\nname: cached-skill\ndescription: first version\n---\n# Cached skill\n", encoding="utf-8", ) calls = 0 original_parse = skill_parser.parse_skill_file def counted_parse(*args, **kwargs): nonlocal calls calls += 1 return original_parse(*args, **kwargs) monkeypatch.setattr(skill_parser, "parse_skill_file", counted_parse) storage._skills_cache_check_interval_seconds = 0 assert [skill.name for skill in storage.load_skills()] == ["cached-skill"] assert [skill.name for skill in storage.load_skills()] == ["cached-skill"] assert calls == 1 skill_file.write_text( "---\nname: cached-skill\ndescription: second version, longer\n---\n# Cached skill\n", encoding="utf-8", ) assert storage.load_skills()[0].description == "second version, longer" assert calls == 2 # ---------- 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, # Keep the archive non-empty so we hit the missing-SKILL.md path # instead of the empty-archive guard. extra_files={"notes.txt": "placeholder"}, ), ) 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 # ---------- unattributed (orphaned) custom leftovers are replaceable ---------- @pytest.mark.asyncio async def test_upload_preflight_allows_replacing_unclaimed_orphan( monkeypatch: pytest.MonkeyPatch, ) -> None: # On-disk conflict but NO DB row at all: the leftover is invisible in every # list, so the uploader may replace it and claim ownership. store = MemorySkillStore() 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, builtin_conflict = result[:5] assert can_install_as_is is False assert can_overwrite is True assert owned_by_me is False assert owned_by_other is False assert builtin_conflict is False assert "no owner record" in result[-1] @pytest.mark.asyncio async def test_upload_preflight_allows_replacing_null_owner_row( monkeypatch: pytest.MonkeyPatch, ) -> None: # DB row exists but owner_user_id is NULL: same unmanageable leftover case. store = MemorySkillStore() await store.create({"name": "demo-skill", "owner_user_id": None}) 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 True assert owned_by_me is False assert owned_by_other is False @pytest.mark.asyncio async def test_direct_upload_replaces_unclaimed_orphan( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: store = MemorySkillStore() storage = _UploadStorage() upload_path = tmp_path / "orphan-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 # The orphan is adopted: the uploader becomes the registered owner. record = await store.get_any("demo-skill") assert record is not None assert record["owner_user_id"] == "user-a"