445 lines
18 KiB
Python
445 lines
18 KiB
Python
"""Regression coverage for the opt-in online-skill metadata cache."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import concurrent.futures
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from deerflow.config import paths as paths_module
|
|
from deerflow.sandbox import skill_view
|
|
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
|
|
|
|
|
|
def _skill(root: Path, directory: str, name: str | None = None) -> Path:
|
|
location = root / "public" / directory
|
|
location.mkdir(parents=True, exist_ok=True)
|
|
(location / "SKILL.md").write_text(f"---\nname: {name or directory}\ndescription: cache test\n---\nSkill body\n", encoding="utf-8")
|
|
return location
|
|
|
|
|
|
def _wait_for_warm_validation_followers(expected: int, timeout: float = 5) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
with skill_view._WARM_VALIDATION_LOCK:
|
|
if any(flight.followers >= expected for flight in skill_view._WARM_VALIDATION_FLIGHTS.values()):
|
|
return
|
|
time.sleep(0.005)
|
|
raise AssertionError(f"warm validation did not collect {expected} followers")
|
|
|
|
|
|
@pytest.fixture
|
|
def cache_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
root = tmp_path / "skills"
|
|
_skill(root, "online")
|
|
config = tmp_path / "extensions.json"
|
|
config.write_text(json.dumps({"skills": {"online": {"enabled": True}}}), encoding="utf-8")
|
|
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config))
|
|
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "runtime"))
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: True)
|
|
skill_view.reset_skill_sandbox_view_cache()
|
|
yield root, config, LocalSkillStorage(host_path=str(root))
|
|
skill_view.reset_skill_sandbox_view_cache()
|
|
|
|
|
|
def test_warm_cache_does_not_walk_parse_copy_or_take_refresh_lock(cache_setup, monkeypatch):
|
|
_root, _config, storage = cache_setup
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
|
|
def unexpected(*args, **kwargs):
|
|
raise AssertionError("warm cache executed an expensive refresh operation")
|
|
|
|
monkeypatch.setattr(storage, "load_skills", unexpected)
|
|
monkeypatch.setattr(skill_view, "_scan_tree", unexpected)
|
|
monkeypatch.setattr(skill_view, "_copy_file_atomic", unexpected)
|
|
monkeypatch.setattr(skill_view, "_view_lock", unexpected)
|
|
assert skill_view.refresh_skill_sandbox_view(storage=storage) is first
|
|
assert first.enabled_roots == (("online", "public", "online"),)
|
|
|
|
|
|
def test_offline_and_invalid_config_invalidate_immediately(cache_setup):
|
|
root, config, storage = cache_setup
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8")
|
|
offline = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not offline.enabled_names
|
|
assert not (offline.root / "public" / "online").exists()
|
|
assert (root / "public" / "online" / "SKILL.md").exists()
|
|
config.write_text("{", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="not valid JSON"):
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not (first.root / skill_view.SKILL_VIEW_MARKER).exists()
|
|
assert not list(first.root.iterdir())
|
|
assert not skill_view._VIEW_CACHE
|
|
|
|
|
|
def test_external_source_change_same_size_preserved_mtime_and_new_files(cache_setup):
|
|
root, _config, storage = cache_setup
|
|
source = root / "public" / "online" / "helper.py"
|
|
source.write_text("VALUE=1\n", encoding="utf-8")
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
before = source.stat()
|
|
source.write_text("VALUE=2\n", encoding="utf-8")
|
|
os.utime(source, ns=(before.st_atime_ns, before.st_mtime_ns))
|
|
updated = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert updated.generation != first.generation
|
|
assert (updated.root / "public" / "online" / "helper.py").read_text() == "VALUE=2\n"
|
|
(source.parent / "new-reference.md").write_text("fresh", encoding="utf-8")
|
|
new_file_view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert (new_file_view.root / "public" / "online" / "new-reference.md").read_text() == "fresh"
|
|
source.unlink()
|
|
deleted = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not (deleted.root / "public" / "online" / "helper.py").exists()
|
|
|
|
|
|
def test_nested_new_skill_and_support_manifest_are_detected(cache_setup):
|
|
root, _config, storage = cache_setup
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
_skill(root, "nested/new", "new-skill")
|
|
support = root / "public" / "shared-lib"
|
|
support.mkdir()
|
|
(support / "helper.py").write_text("value", encoding="utf-8")
|
|
manifest = support / skill_view.SKILL_SUPPORT_MANIFEST
|
|
manifest.write_text(json.dumps({"required_by": ["online"]}), encoding="utf-8")
|
|
changed = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert ("new-skill", "public", "nested/new") in changed.enabled_roots
|
|
assert (changed.root / "public" / "shared-lib" / "helper.py").exists()
|
|
manifest.write_text(json.dumps({"required_by": ["not-online"]}), encoding="utf-8")
|
|
removed = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not (removed.root / "public" / "shared-lib").exists()
|
|
|
|
|
|
def test_destination_tampering_is_repaired_even_with_preserved_size_mtime(cache_setup):
|
|
root, _config, storage = cache_setup
|
|
source = root / "public" / "online" / "helper.py"
|
|
source.write_text("VALUE=1\n", encoding="utf-8")
|
|
view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
destination = view.root / "public" / "online" / "helper.py"
|
|
before = destination.stat()
|
|
destination.write_text("VALUE=9\n", encoding="utf-8")
|
|
os.utime(destination, ns=(before.st_atime_ns, before.st_mtime_ns))
|
|
repaired = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert (repaired.root / "public" / "online" / "helper.py").read_text() == "VALUE=1\n"
|
|
rogue = view.root / "public" / "unknown"
|
|
rogue.mkdir()
|
|
(rogue / "secret").write_text("unexpected", encoding="utf-8")
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not rogue.exists()
|
|
|
|
|
|
def test_second_process_cache_and_force_rebuild(cache_setup, monkeypatch):
|
|
_root, _config, storage = cache_setup
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
marker_metadata = (first.root / skill_view.SKILL_VIEW_MARKER).stat()
|
|
skill_view.reset_skill_sandbox_view_cache()
|
|
original_load = storage.load_skills
|
|
count = 0
|
|
|
|
def counted(**kwargs):
|
|
nonlocal count
|
|
count += 1
|
|
return original_load(**kwargs)
|
|
|
|
monkeypatch.setattr(storage, "load_skills", counted)
|
|
second = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert second == first
|
|
assert count == 0
|
|
assert (first.root / skill_view.SKILL_VIEW_MARKER).stat().st_mtime_ns == marker_metadata.st_mtime_ns
|
|
child = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"""
|
|
import sys
|
|
from deerflow.sandbox import skill_view
|
|
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
|
|
skill_view._skill_view_cache_enabled = lambda: True
|
|
storage = LocalSkillStorage(host_path=sys.argv[1])
|
|
def unexpected(**kwargs):
|
|
raise AssertionError('another process unnecessarily rediscovered skills')
|
|
storage.load_skills = unexpected
|
|
view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
print(view.generation)
|
|
""",
|
|
str(_root),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
check=True,
|
|
)
|
|
assert child.stdout.strip() == first.generation
|
|
skill_view.refresh_skill_sandbox_view(storage=storage, force=True)
|
|
assert count == 1
|
|
|
|
|
|
def test_concurrent_cold_calls_join_one_refresh(cache_setup, monkeypatch, caplog):
|
|
_root, _config, storage = cache_setup
|
|
original = storage.load_skills
|
|
count = 0
|
|
barrier = threading.Barrier(8)
|
|
|
|
def counted(**kwargs):
|
|
nonlocal count
|
|
count += 1
|
|
return original(**kwargs)
|
|
|
|
def call():
|
|
barrier.wait(timeout=5)
|
|
return skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
|
|
monkeypatch.setattr(storage, "load_skills", counted)
|
|
with caplog.at_level(logging.INFO, logger=skill_view.__name__), concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(lambda _: call(), range(8)))
|
|
assert count == 1
|
|
assert len({result.generation for result in results}) == 1
|
|
assert "lock_wait_ms=" in caplog.text
|
|
assert "refresh_ms=" in caplog.text
|
|
assert "source_entries=" in caplog.text
|
|
assert "Skill body" not in caplog.text
|
|
|
|
|
|
def test_concurrent_warm_calls_share_only_the_overlapping_validation(cache_setup, monkeypatch):
|
|
_root, _config, storage = cache_setup
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
original = skill_view._entry_matches
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
count = 0
|
|
count_lock = threading.Lock()
|
|
|
|
def blocked(entry, skills_root, config_key):
|
|
nonlocal count
|
|
with count_lock:
|
|
count += 1
|
|
entered.set()
|
|
assert release.wait(timeout=5)
|
|
return original(entry, skills_root, config_key)
|
|
|
|
monkeypatch.setattr(skill_view, "_entry_matches", blocked)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
leader = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage)
|
|
assert entered.wait(timeout=5)
|
|
followers = [pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) for _ in range(7)]
|
|
_wait_for_warm_validation_followers(7)
|
|
release.set()
|
|
results = [leader.result(timeout=5), *(future.result(timeout=5) for future in followers)]
|
|
|
|
assert count == 1
|
|
assert all(result is first for result in results)
|
|
assert not skill_view._WARM_VALIDATION_FLIGHTS
|
|
|
|
# There is no TTL/result cache: a caller after the completed batch validates again.
|
|
assert skill_view.refresh_skill_sandbox_view(storage=storage) is first
|
|
assert count == 2
|
|
|
|
|
|
def test_warm_validation_failure_reaches_followers_and_next_call_recovers(cache_setup, monkeypatch):
|
|
_root, _config, storage = cache_setup
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
original = skill_view._entry_matches
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
count = 0
|
|
|
|
def broken(*_args):
|
|
nonlocal count
|
|
count += 1
|
|
entered.set()
|
|
assert release.wait(timeout=5)
|
|
raise RuntimeError("simulated warm validation failure")
|
|
|
|
monkeypatch.setattr(skill_view, "_entry_matches", broken)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
|
|
leader = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage)
|
|
assert entered.wait(timeout=5)
|
|
followers = [pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) for _ in range(5)]
|
|
_wait_for_warm_validation_followers(5)
|
|
release.set()
|
|
for future in [leader, *followers]:
|
|
with pytest.raises(RuntimeError, match="simulated warm validation failure"):
|
|
future.result(timeout=5)
|
|
|
|
assert count == 1
|
|
assert not skill_view._WARM_VALIDATION_FLIGHTS
|
|
monkeypatch.setattr(skill_view, "_entry_matches", original)
|
|
assert skill_view.refresh_skill_sandbox_view(storage=storage) is first
|
|
|
|
|
|
def test_warm_validation_with_new_config_hash_does_not_join_old_flight(cache_setup, monkeypatch):
|
|
_root, config, storage = cache_setup
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
original = skill_view._entry_matches
|
|
old_key = skill_view._config_key()
|
|
old_entered = threading.Event()
|
|
release_old = threading.Event()
|
|
observed_keys: list[str] = []
|
|
observed_lock = threading.Lock()
|
|
|
|
def block_old_config(entry, skills_root, config_key):
|
|
with observed_lock:
|
|
observed_keys.append(config_key)
|
|
if config_key == old_key:
|
|
old_entered.set()
|
|
assert release_old.wait(timeout=5)
|
|
return original(entry, skills_root, config_key)
|
|
|
|
monkeypatch.setattr(skill_view, "_entry_matches", block_old_config)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
|
|
old_call = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage)
|
|
assert old_entered.wait(timeout=5)
|
|
config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8")
|
|
new_key = skill_view._config_key()
|
|
assert new_key != old_key
|
|
new_call = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage)
|
|
try:
|
|
# This must finish while the old-hash leader is still blocked.
|
|
changed = new_call.result(timeout=5)
|
|
finally:
|
|
release_old.set()
|
|
joined_after_change = old_call.result(timeout=5)
|
|
|
|
assert not changed.enabled_names
|
|
assert not joined_after_change.enabled_names
|
|
assert observed_keys.count(old_key) == 1
|
|
assert new_key in observed_keys
|
|
assert not skill_view._WARM_VALIDATION_FLIGHTS
|
|
|
|
|
|
def test_disabled_switch_retains_full_refresh_and_legacy_marker(cache_setup, monkeypatch):
|
|
_root, _config, storage = cache_setup
|
|
monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: False)
|
|
original = storage.load_skills
|
|
count = 0
|
|
|
|
def counted(**kwargs):
|
|
nonlocal count
|
|
count += 1
|
|
return original(**kwargs)
|
|
|
|
monkeypatch.setattr(storage, "load_skills", counted)
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert count == 2
|
|
assert not first.generation
|
|
assert "generation" not in json.loads(first.marker_content)
|
|
assert first.enabled_roots == (("online", "public", "online"),)
|
|
|
|
|
|
def test_custom_storage_instances_do_not_share_filesystem_cache(cache_setup):
|
|
root, _config, _storage = cache_setup
|
|
_skill(root, "other")
|
|
|
|
class FilteredStorage(LocalSkillStorage):
|
|
def __init__(self, selected):
|
|
super().__init__(host_path=str(root))
|
|
self.selected = selected
|
|
|
|
def load_skills(self, **kwargs):
|
|
return [skill for skill in super().load_skills(**kwargs) if skill.name == self.selected]
|
|
|
|
first = skill_view.refresh_skill_sandbox_view(storage=FilteredStorage("online"))
|
|
second = skill_view.refresh_skill_sandbox_view(storage=FilteredStorage("other"))
|
|
assert first.enabled_names == {"online"}
|
|
assert second.enabled_names == {"other"}
|
|
assert not first.generation and not second.generation
|
|
|
|
|
|
def test_config_change_mid_refresh_is_not_published(cache_setup, monkeypatch):
|
|
_root, config, storage = cache_setup
|
|
original = skill_view._copy_file_atomic
|
|
changed = False
|
|
|
|
def copy_and_disable(source, destination):
|
|
nonlocal changed
|
|
original(source, destination)
|
|
if not changed:
|
|
changed = True
|
|
config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8")
|
|
|
|
monkeypatch.setattr(skill_view, "_copy_file_atomic", copy_and_disable)
|
|
view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not view.enabled_names
|
|
assert not (view.root / "public" / "online").exists()
|
|
|
|
|
|
def test_switching_cache_off_and_back_on_does_not_reuse_a_stale_marker(cache_setup, monkeypatch):
|
|
root, _config, storage = cache_setup
|
|
original = storage.load_skills
|
|
count = 0
|
|
|
|
def counted(**kwargs):
|
|
nonlocal count
|
|
count += 1
|
|
return original(**kwargs)
|
|
|
|
monkeypatch.setattr(storage, "load_skills", counted)
|
|
first = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: False)
|
|
(root / "public" / "online" / "reference.md").write_text("new content", encoding="utf-8")
|
|
legacy = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert count == 3
|
|
assert not legacy.generation
|
|
assert "generation" not in json.loads(legacy.marker_content)
|
|
monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: True)
|
|
restored = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert restored.generation != first.generation
|
|
assert (restored.root / "public" / "online" / "reference.md").read_text() == "new content"
|
|
assert count == 4
|
|
assert skill_view.refresh_skill_sandbox_view(storage=storage) is restored
|
|
assert count == 4
|
|
|
|
|
|
def test_marker_damage_and_copy_failure_clear_capability(cache_setup, monkeypatch):
|
|
root, _config, storage = cache_setup
|
|
view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
marker = view.root / skill_view.SKILL_VIEW_MARKER
|
|
marker.write_text("not-a-valid-marker", encoding="utf-8")
|
|
repaired = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert marker.read_text() == repaired.marker_content
|
|
(root / "public" / "online" / "SKILL.md").write_text("---\nname: online\ndescription: changed\n---\n", encoding="utf-8")
|
|
|
|
def broken_copy(*args):
|
|
raise OSError("simulated disk failure")
|
|
|
|
monkeypatch.setattr(skill_view, "_copy_file_atomic", broken_copy)
|
|
with pytest.raises(OSError, match="simulated disk failure"):
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not marker.exists()
|
|
assert not skill_view._VIEW_CACHE
|
|
assert not list(view.root.iterdir())
|
|
|
|
|
|
def test_symlink_destination_does_not_modify_external_directory(cache_setup, tmp_path):
|
|
_root, _config, storage = cache_setup
|
|
view = skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
external = tmp_path / "outside"
|
|
external.mkdir()
|
|
(external / "keep").write_text("untouched", encoding="utf-8")
|
|
target = view.root / "public" / "online"
|
|
import shutil
|
|
|
|
shutil.rmtree(target)
|
|
try:
|
|
target.symlink_to(external, target_is_directory=True)
|
|
except OSError:
|
|
if os.name != "nt":
|
|
pytest.skip("directory symlinks are unavailable on this platform")
|
|
junction = subprocess.run(["cmd", "/c", "mklink", "/J", str(target), str(external)], capture_output=True, timeout=10)
|
|
if junction.returncode:
|
|
pytest.skip("directory links are unavailable on this platform")
|
|
skill_view.refresh_skill_sandbox_view(storage=storage)
|
|
assert not skill_view._is_view_link(target)
|
|
assert (external / "keep").read_text() == "untouched"
|
|
assert not (external / "SKILL.md").exists()
|