wehub-resource-sync
574 lines
28 KiB
Python
574 lines
28 KiB
Python
"""Tests for UserScopedSkillStorage: per-user isolation, fallback, and path safety."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import stat
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from deerflow.config.paths import Paths
|
|
from deerflow.skills.storage import reset_skill_storage, reset_user_skill_storage
|
|
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
|
|
def _skill_content(name: str, description: str = "Demo skill") -> str:
|
|
return f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_storages():
|
|
"""Reset all skill storage caches between tests."""
|
|
reset_skill_storage()
|
|
yield
|
|
reset_skill_storage()
|
|
|
|
|
|
@pytest.fixture
|
|
def base_dir(tmp_path: Path) -> Path:
|
|
"""Provide a temp directory as the DeerFlow base_dir."""
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def paths(base_dir: Path) -> Paths:
|
|
return Paths(base_dir=base_dir)
|
|
|
|
|
|
@pytest.fixture
|
|
def skills_root(base_dir: Path) -> Path:
|
|
"""Create the global skills root directory with public/ and custom/ subdirs."""
|
|
root = base_dir / "skills"
|
|
root.mkdir()
|
|
(root / "public").mkdir()
|
|
(root / "custom").mkdir()
|
|
return root
|
|
|
|
|
|
@pytest.fixture
|
|
def config(skills_root):
|
|
"""Minimal app_config-like namespace for storage construction."""
|
|
from types import SimpleNamespace
|
|
|
|
return SimpleNamespace(
|
|
skills=SimpleNamespace(
|
|
get_skills_path=lambda: skills_root,
|
|
container_path="/mnt/skills",
|
|
use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage",
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def user_storage(base_dir: Path, skills_root, config) -> UserScopedSkillStorage:
|
|
"""Create a UserScopedSkillStorage for user 'test-user'."""
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
storage = UserScopedSkillStorage("test-user", host_path=str(skills_root), app_config=config)
|
|
return storage
|
|
|
|
|
|
class TestPathRedirection:
|
|
"""Custom skill paths are redirected to per-user directories."""
|
|
|
|
def test_custom_skill_dir_is_user_scoped(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
expected = base_dir / "users" / "test-user" / "skills" / "custom" / "demo-skill"
|
|
assert user_storage.get_custom_skill_dir("demo-skill") == expected
|
|
|
|
def test_custom_skill_file_is_user_scoped(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
expected = base_dir / "users" / "test-user" / "skills" / "custom" / "demo-skill" / "SKILL.md"
|
|
assert user_storage.get_custom_skill_file("demo-skill") == expected
|
|
|
|
def test_history_file_is_user_scoped(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
expected = base_dir / "users" / "test-user" / "skills" / "custom" / ".history" / "demo-skill.jsonl"
|
|
assert user_storage.get_skill_history_file("demo-skill") == expected
|
|
|
|
def test_public_skill_paths_still_use_global_root(self, user_storage: UserScopedSkillStorage, skills_root: Path):
|
|
assert user_storage.get_skills_root_path() == skills_root
|
|
|
|
def test_user_id_property(self, user_storage: UserScopedSkillStorage):
|
|
assert user_storage.user_id == "test-user"
|
|
|
|
|
|
class TestWriteAndRead:
|
|
"""Writes go to user dir, reads from user dir when present."""
|
|
|
|
def test_write_creates_file_in_user_dir(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
user_storage.write_custom_skill("demo-skill", "SKILL.md", _skill_content("demo-skill"))
|
|
user_file = base_dir / "users" / "test-user" / "skills" / "custom" / "demo-skill" / "SKILL.md"
|
|
assert user_file.exists()
|
|
assert user_file.read_text(encoding="utf-8") == _skill_content("demo-skill")
|
|
|
|
def test_write_does_not_create_in_global_custom(self, user_storage: UserScopedSkillStorage, skills_root: Path, base_dir: Path):
|
|
user_storage.write_custom_skill("demo-skill", "SKILL.md", _skill_content("demo-skill"))
|
|
global_file = skills_root / "custom" / "demo-skill" / "SKILL.md"
|
|
assert not global_file.exists()
|
|
|
|
def test_read_from_user_dir(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
user_storage.write_custom_skill("demo-skill", "SKILL.md", _skill_content("demo-skill"))
|
|
content = user_storage.read_custom_skill("demo-skill")
|
|
assert "demo-skill" in content
|
|
|
|
def test_read_not_found_raises(self, user_storage: UserScopedSkillStorage):
|
|
with pytest.raises(FileNotFoundError):
|
|
user_storage.read_custom_skill("nonexistent")
|
|
|
|
def test_write_makes_path_sandbox_readable(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
user_storage.write_custom_skill("demo-skill", "references/ref.md", "# ref")
|
|
skill_dir = base_dir / "users" / "test-user" / "skills" / "custom" / "demo-skill"
|
|
ref_dir = skill_dir / "references"
|
|
assert stat.S_IMODE(skill_dir.stat().st_mode) & 0o055 == 0o055
|
|
assert stat.S_IMODE(ref_dir.stat().st_mode) & 0o055 == 0o055
|
|
|
|
|
|
class TestSkillLoading:
|
|
"""Public skills from global, custom from user dir + fallback."""
|
|
|
|
def test_public_skills_loaded_from_global(self, user_storage: UserScopedSkillStorage, skills_root: Path):
|
|
public_dir = skills_root / "public" / "deep-research"
|
|
public_dir.mkdir(parents=True)
|
|
(public_dir / "SKILL.md").write_text(_skill_content("deep-research"), encoding="utf-8")
|
|
|
|
skills = user_storage.load_skills(enabled_only=False)
|
|
public_skills = [s for s in skills if s.category == SkillCategory.PUBLIC]
|
|
assert len(public_skills) == 1
|
|
assert public_skills[0].name == "deep-research"
|
|
|
|
def test_custom_skills_loaded_from_user_dir(self, user_storage: UserScopedSkillStorage, base_dir: Path):
|
|
user_storage.write_custom_skill("my-skill", "SKILL.md", _skill_content("my-skill"))
|
|
|
|
skills = user_storage.load_skills(enabled_only=False)
|
|
custom_skills = [s for s in skills if s.category == SkillCategory.CUSTOM]
|
|
assert len(custom_skills) == 1
|
|
assert custom_skills[0].name == "my-skill"
|
|
|
|
def test_fallback_to_global_custom_when_user_dir_empty(self, user_storage: UserScopedSkillStorage, skills_root: Path, base_dir: Path):
|
|
# Put skill in global custom (NOT in user dir)
|
|
global_dir = skills_root / "custom" / "global-skill"
|
|
global_dir.mkdir(parents=True)
|
|
(global_dir / "SKILL.md").write_text(_skill_content("global-skill"), encoding="utf-8")
|
|
|
|
# User dir is empty → fallback loads from global custom as LEGACY
|
|
skills = user_storage.load_skills(enabled_only=False)
|
|
legacy_skills = [s for s in skills if s.category == SkillCategory.LEGACY]
|
|
assert len(legacy_skills) == 1
|
|
assert legacy_skills[0].name == "global-skill"
|
|
|
|
def test_no_fallback_when_user_dir_has_content(self, user_storage: UserScopedSkillStorage, skills_root: Path, base_dir: Path):
|
|
# Put skill in global custom
|
|
global_dir = skills_root / "custom" / "global-skill"
|
|
global_dir.mkdir(parents=True)
|
|
(global_dir / "SKILL.md").write_text(_skill_content("global-skill"), encoding="utf-8")
|
|
|
|
# Also put skill in user custom
|
|
user_storage.write_custom_skill("user-skill", "SKILL.md", _skill_content("user-skill"))
|
|
|
|
# User dir has content → no fallback, only user-level skill
|
|
skills = user_storage.load_skills(enabled_only=False)
|
|
custom_skills = [s for s in skills if s.category == SkillCategory.CUSTOM]
|
|
assert len(custom_skills) == 1
|
|
assert custom_skills[0].name == "user-skill"
|
|
|
|
def test_mixed_public_and_custom(self, user_storage: UserScopedSkillStorage, skills_root: Path, base_dir: Path):
|
|
# Create public skill
|
|
public_dir = skills_root / "public" / "deep-research"
|
|
public_dir.mkdir(parents=True)
|
|
(public_dir / "SKILL.md").write_text(_skill_content("deep-research"), encoding="utf-8")
|
|
|
|
# Create user custom skill
|
|
user_storage.write_custom_skill("my-skill", "SKILL.md", _skill_content("my-skill"))
|
|
|
|
skills = user_storage.load_skills(enabled_only=False)
|
|
assert len(skills) == 2
|
|
categories = {s.category for s in skills}
|
|
assert categories == {SkillCategory.PUBLIC, SkillCategory.CUSTOM}
|
|
|
|
|
|
class TestIsolation:
|
|
"""Different users must see different custom skills."""
|
|
|
|
def test_two_users_isolated(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
storage_a = UserScopedSkillStorage("alice", host_path=str(skills_root), app_config=config)
|
|
storage_b = UserScopedSkillStorage("bob", host_path=str(skills_root), app_config=config)
|
|
|
|
storage_a.write_custom_skill("skill-a", "SKILL.md", _skill_content("skill-a"))
|
|
storage_b.write_custom_skill("skill-b", "SKILL.md", _skill_content("skill-b"))
|
|
|
|
skills_a = [s for s in storage_a.load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
|
|
skills_b = [s for s in storage_b.load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
|
|
|
|
assert len(skills_a) == 1
|
|
assert skills_a[0].name == "skill-a"
|
|
assert len(skills_b) == 1
|
|
assert skills_b[0].name == "skill-b"
|
|
|
|
def test_delete_is_isolated(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
storage_a = UserScopedSkillStorage("alice", host_path=str(skills_root), app_config=config)
|
|
storage_b = UserScopedSkillStorage("bob", host_path=str(skills_root), app_config=config)
|
|
|
|
storage_a.write_custom_skill("skill-a", "SKILL.md", _skill_content("skill-a"))
|
|
storage_b.write_custom_skill("skill-b", "SKILL.md", _skill_content("skill-b"))
|
|
|
|
storage_a.delete_custom_skill("skill-a")
|
|
|
|
# Alice has no custom skills, Bob still has theirs
|
|
skills_a = [s for s in storage_a.load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
|
|
skills_b = [s for s in storage_b.load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
|
|
|
|
assert len(skills_a) == 0
|
|
assert len(skills_b) == 1
|
|
|
|
|
|
class TestHistoryIsolation:
|
|
"""History files are per-user."""
|
|
|
|
def test_history_per_user(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
storage_a = UserScopedSkillStorage("alice", host_path=str(skills_root), app_config=config)
|
|
storage_a.write_custom_skill("shared-name", "SKILL.md", _skill_content("shared-name"))
|
|
|
|
storage_a.append_history("shared-name", {"action": "create", "author": "alice"})
|
|
|
|
history_file_a = base_dir / "users" / "alice" / "skills" / "custom" / ".history" / "shared-name.jsonl"
|
|
assert history_file_a.exists()
|
|
|
|
def test_history_does_not_leak_to_global(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
storage = UserScopedSkillStorage("alice", host_path=str(skills_root), app_config=config)
|
|
storage.write_custom_skill("my-skill", "SKILL.md", _skill_content("my-skill"))
|
|
storage.append_history("my-skill", {"action": "create"})
|
|
|
|
global_history = skills_root / "custom" / ".history" / "my-skill.jsonl"
|
|
assert not global_history.exists()
|
|
|
|
|
|
class TestPathSafety:
|
|
"""UserScopedSkillStorage inherits path-traversal guards from LocalSkillStorage."""
|
|
|
|
def test_rejects_invalid_skill_name(self, user_storage: UserScopedSkillStorage):
|
|
with pytest.raises(ValueError, match="hyphen-case"):
|
|
user_storage.get_custom_skill_dir("../../escaped")
|
|
|
|
def test_rejects_path_traversal_in_write(self, user_storage: UserScopedSkillStorage):
|
|
with pytest.raises(ValueError, match="skill directory"):
|
|
user_storage.write_custom_skill("demo-skill", "../../escaped.txt", "x")
|
|
|
|
def test_rejects_empty_path_in_write(self, user_storage: UserScopedSkillStorage):
|
|
with pytest.raises(ValueError, match="empty"):
|
|
user_storage.write_custom_skill("demo-skill", "", "x")
|
|
|
|
|
|
class TestFactory:
|
|
"""get_or_new_user_skill_storage factory behavior."""
|
|
|
|
def test_returns_same_instance_for_same_user(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
|
|
s1 = get_or_new_user_skill_storage("alice", app_config=config)
|
|
s2 = get_or_new_user_skill_storage("alice", app_config=config)
|
|
assert s1 is s2
|
|
|
|
def test_returns_different_instance_for_different_user(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
|
|
s1 = get_or_new_user_skill_storage("alice", app_config=config)
|
|
s2 = get_or_new_user_skill_storage("bob", app_config=config)
|
|
assert s1 is not s2
|
|
|
|
def test_reset_clears_specific_user(self, base_dir: Path, skills_root, config):
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
|
|
s_alice = get_or_new_user_skill_storage("alice", app_config=config)
|
|
s_bob = get_or_new_user_skill_storage("bob", app_config=config)
|
|
|
|
reset_user_skill_storage("alice")
|
|
|
|
# Alice's storage is gone; a new one is created
|
|
s_alice_new = get_or_new_user_skill_storage("alice", app_config=config)
|
|
assert s_alice_new is not s_alice
|
|
|
|
# Bob's storage is still cached
|
|
s_bob_cached = get_or_new_user_skill_storage("bob", app_config=config)
|
|
assert s_bob_cached is s_bob
|
|
|
|
|
|
class TestSkillToggleIsolation:
|
|
"""Per-user enabled/disabled state isolation for same-named custom skills.
|
|
|
|
When Alice and Bob each own a custom skill named 'report-gen', disabling
|
|
Alice's copy must NOT affect Bob's. The enabled state is stored in
|
|
per-user ``_skill_states.json`` so same-named skills can be toggled
|
|
independently across users.
|
|
"""
|
|
|
|
def test_alice_disable_does_not_affect_bob(self, base_dir: Path, skills_root, config):
|
|
from types import SimpleNamespace
|
|
|
|
from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache, get_skills_prompt_section
|
|
from deerflow.sandbox.tools import _is_disabled_skill_path
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
|
|
# Rich config that includes skill_evolution (required by
|
|
# get_skills_prompt_section) while keeping the test skills root.
|
|
rich_config = SimpleNamespace(
|
|
skills=config.skills,
|
|
skill_evolution=SimpleNamespace(enabled=False),
|
|
)
|
|
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base_dir)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
with patch("deerflow.config.get_app_config", return_value=rich_config):
|
|
# Use the factory so storages enter the cache — both
|
|
# _is_disabled_skill_path and get_skills_prompt_section
|
|
# call the factory internally.
|
|
storage_alice = get_or_new_user_skill_storage("alice", app_config=rich_config)
|
|
storage_bob = get_or_new_user_skill_storage("bob", app_config=rich_config)
|
|
|
|
# 1. Two users each create a custom skill named "report-gen"
|
|
storage_alice.write_custom_skill("report-gen", "SKILL.md", _skill_content("report-gen", "Alice report generator"))
|
|
storage_bob.write_custom_skill("report-gen", "SKILL.md", _skill_content("report-gen", "Bob report generator"))
|
|
|
|
# 2. Alice disables her "report-gen"
|
|
storage_alice.set_skill_enabled_state("report-gen", False)
|
|
|
|
# 3. Bob's "report-gen" stays enabled in load_skills()
|
|
bob_skills = storage_bob.load_skills(enabled_only=False)
|
|
bob_report = [s for s in bob_skills if s.name == "report-gen" and s.category == SkillCategory.CUSTOM]
|
|
assert len(bob_report) == 1
|
|
assert bob_report[0].enabled is True
|
|
|
|
# Complementary: Alice's "report-gen" is disabled
|
|
alice_skills = storage_alice.load_skills(enabled_only=False)
|
|
alice_report = [s for s in alice_skills if s.name == "report-gen" and s.category == SkillCategory.CUSTOM]
|
|
assert len(alice_report) == 1
|
|
assert alice_report[0].enabled is False
|
|
|
|
# enabled_only=True filtering is also isolated
|
|
bob_enabled = storage_bob.load_skills(enabled_only=True)
|
|
assert any(s.name == "report-gen" for s in bob_enabled)
|
|
|
|
alice_enabled = storage_alice.load_skills(enabled_only=True)
|
|
assert not any(s.name == "report-gen" for s in alice_enabled)
|
|
|
|
# 4. Bob's skill still appears in the prompt section
|
|
clear_skills_system_prompt_cache()
|
|
prompt = get_skills_prompt_section(user_id="bob", app_config=rich_config)
|
|
assert "report-gen" in prompt
|
|
|
|
# 5. _is_disabled_skill_path returns False for Bob's skill path
|
|
assert _is_disabled_skill_path("/mnt/skills/custom/report-gen/SKILL.md", user_id="bob") is False
|
|
|
|
# Complementary: Alice's skill path IS disabled
|
|
assert _is_disabled_skill_path("/mnt/skills/custom/report-gen/SKILL.md", user_id="alice") is True
|
|
|
|
|
|
class TestSkillStateAtomicWrite:
|
|
"""P2-2: ``_write_skill_states`` must be atomic so a crash mid-write
|
|
cannot silently re-enable every skill the user had disabled.
|
|
"""
|
|
|
|
def test_writes_via_tempfile_then_replace(self, user_storage: UserScopedSkillStorage, base_dir: Path) -> None:
|
|
states = {"report-gen": {"enabled": False}}
|
|
user_storage._write_skill_states(states)
|
|
|
|
target = user_storage._skill_states_file
|
|
assert target.exists()
|
|
# No leftover .tmp files in the directory.
|
|
leftovers = [p for p in base_dir.glob("users/*/skills/.skill_states_*.json.tmp") if p.exists()]
|
|
assert not leftovers, f"temp file left behind: {leftovers}"
|
|
import json as _json
|
|
|
|
assert _json.loads(target.read_text(encoding="utf-8")) == states
|
|
|
|
def test_failed_write_does_not_truncate_existing_file(self, user_storage: UserScopedSkillStorage) -> None:
|
|
import json as _json
|
|
|
|
# Seed a valid state file.
|
|
target = user_storage._skill_states_file
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(_json.dumps({"old-skill": {"enabled": False}}), encoding="utf-8")
|
|
|
|
# Force the inner write to fail; ensure the existing file is intact.
|
|
with patch("pathlib.Path.replace", side_effect=OSError("boom")):
|
|
with pytest.raises(OSError):
|
|
user_storage._write_skill_states({"new-skill": {"enabled": True}})
|
|
|
|
# Pre-existing content must survive the failed replacement.
|
|
assert target.exists()
|
|
assert _json.loads(target.read_text(encoding="utf-8")) == {"old-skill": {"enabled": False}}
|
|
|
|
# And no orphan temp file should be left around in the user skills dir.
|
|
leftovers = list(target.parent.glob(".skill_states_*.json.tmp"))
|
|
assert not leftovers, f"temp file leaked: {leftovers}"
|
|
|
|
|
|
class TestSkillStateFailClosed:
|
|
"""P1-2: ``_is_disabled_skill_path`` must fail CLOSED (return True)
|
|
when the enabled state cannot be determined, so a corrupt
|
|
``_skill_states.json`` or mid-write race never lets the agent read a
|
|
disabled skill's files.
|
|
"""
|
|
|
|
def test_returns_true_when_state_lookup_raises(self) -> None:
|
|
from deerflow.sandbox.tools import _is_disabled_skill_path
|
|
|
|
def _boom(_skill_name: str) -> bool:
|
|
raise OSError("storage unavailable")
|
|
|
|
with patch("deerflow.skills.storage.user_scoped_skill_storage.UserScopedSkillStorage.get_skill_enabled_state", side_effect=_boom):
|
|
assert _is_disabled_skill_path("/mnt/skills/custom/report-gen/SKILL.md", user_id="default") is True
|
|
|
|
def test_returns_true_when_public_extensions_config_raises(self) -> None:
|
|
from deerflow.sandbox.tools import _is_disabled_skill_path
|
|
|
|
def _boom() -> bool:
|
|
raise OSError("extensions_config.json unreadable")
|
|
|
|
with patch("deerflow.config.extensions_config.ExtensionsConfig.from_file", side_effect=_boom):
|
|
assert _is_disabled_skill_path("/mnt/skills/public/bootstrap/SKILL.md", user_id="default") is True
|
|
|
|
|
|
class TestSkillLoadingRespectsGlobalDisable:
|
|
"""P2-1: when the global ``extensions_config.json`` disables a
|
|
CUSTOM/LEGACY skill, ``load_skills`` must still report it as
|
|
disabled even if the per-user state has no entry (defaulting to
|
|
enabled otherwise). Without the AND, an admin's global "off" for a
|
|
shared skill would be silently flipped to "on" the moment a new
|
|
user touches the per-user storage.
|
|
"""
|
|
|
|
def test_global_disable_wins_when_per_user_state_missing(self, tmp_path: Path) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from deerflow.config.paths import Paths
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
base = tmp_path
|
|
skills_root = base / "skills"
|
|
skills_root.mkdir()
|
|
(skills_root / "custom").mkdir()
|
|
(skills_root / "custom" / "shared-skill").mkdir()
|
|
(skills_root / "custom" / "shared-skill" / "SKILL.md").write_text(
|
|
_skill_content("shared-skill"),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Per-user state is empty (no per-user override for "shared-skill").
|
|
with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=base)):
|
|
with patch("deerflow.config.paths._paths", None):
|
|
cfg = SimpleNamespace(
|
|
skills=SimpleNamespace(
|
|
get_skills_path=lambda: skills_root,
|
|
container_path="/mnt/skills",
|
|
use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage",
|
|
),
|
|
)
|
|
with patch("deerflow.config.get_app_config", return_value=cfg):
|
|
# Global extensions_config reports the shared skill as disabled.
|
|
ext_cfg = SimpleNamespace(
|
|
skills={"shared-skill": SimpleNamespace(enabled=False)},
|
|
is_skill_enabled=lambda name, _cat: not (name == "shared-skill"),
|
|
)
|
|
# The function inside ``load_skills`` does a
|
|
# function-local ``from deerflow.config.extensions_config
|
|
# import get_extensions_config``, so patch the
|
|
# extension_config module symbol.
|
|
with patch("deerflow.config.extensions_config.get_extensions_config", return_value=ext_cfg):
|
|
storage = get_or_new_user_skill_storage("alice", app_config=cfg)
|
|
loaded = storage.load_skills(enabled_only=False)
|
|
shared = [s for s in loaded if s.name == "shared-skill" and s.category == SkillCategory.LEGACY]
|
|
assert len(shared) == 1
|
|
# AND-merge: per-user default True AND global False → False
|
|
assert shared[0].enabled is False
|
|
|
|
|
|
class TestEnabledSkillsByConfigCacheBounded:
|
|
"""P2-4: ``_enabled_skills_by_config_cache`` must be bounded so a
|
|
long-running process cannot leak one entry per distinct
|
|
(app_config, user_id) pair ever seen.
|
|
"""
|
|
|
|
def test_evicts_least_recently_used_above_maxsize(self, monkeypatch) -> None:
|
|
from collections import OrderedDict
|
|
|
|
from deerflow.agents.lead_agent import prompt as prompt_module
|
|
|
|
# Shrink the cap so the test stays fast.
|
|
monkeypatch.setattr(prompt_module, "_ENABLED_SKILLS_BY_CONFIG_CACHE_MAXSIZE", 4)
|
|
prompt_module._enabled_skills_by_config_cache = OrderedDict()
|
|
|
|
class FakeConfig:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
|
|
class FakeStorage:
|
|
def __init__(self) -> None:
|
|
self.load_calls = 0
|
|
|
|
def load_skills(self, *, enabled_only: bool = False):
|
|
self.load_calls += 1
|
|
return []
|
|
|
|
configs = [FakeConfig(f"cfg-{i}") for i in range(6)]
|
|
storages = [FakeStorage() for _ in range(6)]
|
|
# Index by cfg id so the lookups below are deterministic.
|
|
cfg_to_storage = {id(c): s for c, s in zip(configs, storages)}
|
|
|
|
def _user_storage(user_id, *, app_config=None):
|
|
return cfg_to_storage[id(app_config)]
|
|
|
|
def _global_storage(*, app_config=None):
|
|
return cfg_to_storage[id(app_config)]
|
|
|
|
# Patch the *already-imported* references inside the prompt
|
|
# module. ``from ... import`` binds the name at import time, so
|
|
# patching the storage module has no effect.
|
|
with patch.object(prompt_module, "get_or_new_user_skill_storage", side_effect=_user_storage), patch.object(prompt_module, "get_or_new_skill_storage", side_effect=_global_storage):
|
|
for i, cfg in enumerate(configs):
|
|
prompt_module.get_enabled_skills_for_config(app_config=cfg, user_id=f"user-{i}")
|
|
|
|
# After 6 distinct (cfg, user) inserts with a cap of 4, the cache
|
|
# must hold exactly the 4 most-recently-touched entries.
|
|
assert len(prompt_module._enabled_skills_by_config_cache) == 4
|
|
kept_keys = set(prompt_module._enabled_skills_by_config_cache.keys())
|
|
# The two oldest (cfg-0/user-0, cfg-1/user-1) should have been evicted.
|
|
assert (id(configs[0]), "user-0") not in kept_keys
|
|
assert (id(configs[1]), "user-1") not in kept_keys
|
|
# And the four newest must still be there.
|
|
for i in range(2, 6):
|
|
assert (id(configs[i]), f"user-{i}") in kept_keys
|
|
|
|
# Touching an older key bumps it to MRU. ``configs[0]`` is no
|
|
# longer in the cache, so this re-loads it; this also re-loads
|
|
# ``storages[0]`` from disk (so its load_calls goes from 0 to 1).
|
|
with patch.object(prompt_module, "get_or_new_user_skill_storage", side_effect=_user_storage), patch.object(prompt_module, "get_or_new_skill_storage", side_effect=_global_storage):
|
|
prompt_module.get_enabled_skills_for_config(app_config=configs[0], user_id="user-0")
|
|
# Now configs[0] is MRU. Insert a new (cfg-new, user-new) entry:
|
|
# the LRU is configs[2] and must be evicted.
|
|
new_cfg = FakeConfig("cfg-new")
|
|
new_storage = FakeStorage()
|
|
cfg_to_storage[id(new_cfg)] = new_storage
|
|
with patch.object(prompt_module, "get_or_new_user_skill_storage", side_effect=_user_storage), patch.object(prompt_module, "get_or_new_skill_storage", side_effect=_global_storage):
|
|
prompt_module.get_enabled_skills_for_config(app_config=new_cfg, user_id="user-new")
|
|
|
|
kept = set(prompt_module._enabled_skills_by_config_cache.keys())
|
|
assert (id(configs[0]), "user-0") in kept, "MRU touch should keep configs[0]"
|
|
assert (id(new_cfg), "user-new") in kept
|
|
assert (id(configs[2]), "user-2") not in kept, "LRU should have been evicted"
|
|
assert len(prompt_module._enabled_skills_by_config_cache) == 4
|