定时任务: - 管理员可查看/编辑/删除/触发全部用户的定时任务(新增 /api/scheduled-tasks/admin/* 接口与管理页) - 卡死任务恢复:单次执行超 30 分钟超时重试(最多 3 次),耗尽后标记 failed;看门狗回收服务重启遗留的「执行中」任务 会话分享: - 新增「只读链接」分享模式,生成公开免登录的只读对话快照页(/share/:code) - 公开接口 /api/public/shares/* 提供对话快照与图片附件 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
190 lines
6.9 KiB
Python
190 lines
6.9 KiB
Python
"""Tests for the skill ``detail`` field (long-form annotation editable by
|
|
the skill owner or an admin).
|
|
|
|
Covers:
|
|
- ``SkillStore`` persists ``detail`` through create / ensure_legacy / update /
|
|
update_admin, and owner-only ``update`` still refuses non-owners.
|
|
- ``_skill_to_response`` surfaces ``detail`` from the ownership record.
|
|
- ``SkillResponse`` / ``SkillDetailRequest`` model shapes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers.skills import (
|
|
SkillDetailRequest,
|
|
SkillResponse,
|
|
_can_view_skill_source,
|
|
_skill_to_response,
|
|
)
|
|
from deerflow.config import paths as paths_module
|
|
from deerflow.persistence.skills import MemorySkillStore
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_paths(tmp_path, monkeypatch):
|
|
"""Redirect get_paths() at a fresh temp directory so usage lookups in
|
|
``_skill_to_response`` never touch real state."""
|
|
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
yield tmp_path
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillStore: detail persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_create_carries_detail() -> None:
|
|
store = MemorySkillStore()
|
|
created = await store.create(
|
|
{"name": "alpha-tool", "owner_user_id": "user-a", "detail": "初始详情"}
|
|
)
|
|
assert created["detail"] == "初始详情"
|
|
fetched = await store.get_any("alpha-tool")
|
|
assert fetched is not None and fetched["detail"] == "初始详情"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_create_defaults_detail_to_none() -> None:
|
|
store = MemorySkillStore()
|
|
created = await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
|
|
assert created["detail"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_owner_can_set_and_clear_detail() -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
|
|
|
|
updated = await store.update("alpha-tool", "user-a", {"detail": "技能用途说明"})
|
|
assert updated is not None and updated["detail"] == "技能用途说明"
|
|
|
|
# Clearing the annotation persists None.
|
|
cleared = await store.update("alpha-tool", "user-a", {"detail": None})
|
|
assert cleared is not None and cleared["detail"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_update_detail_rejects_non_owner() -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
|
|
|
|
# A different user cannot write detail via the owner-scoped update.
|
|
assert await store.update("alpha-tool", "user-b", {"detail": "x"}) is None
|
|
record = await store.get_any("alpha-tool")
|
|
assert record is not None and record["detail"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_admin_can_set_detail_on_any_skill() -> None:
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
|
|
|
|
# Admin override writes detail regardless of ownership.
|
|
updated = await store.update_admin("alpha-tool", {"detail": "管理员补充说明"})
|
|
assert updated is not None and updated["detail"] == "管理员补充说明"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_partial_update_preserves_detail() -> None:
|
|
"""Updating only ``published`` must not wipe an existing ``detail``."""
|
|
store = MemorySkillStore()
|
|
await store.create({"name": "alpha-tool", "owner_user_id": "user-a"})
|
|
await store.update("alpha-tool", "user-a", {"detail": "保留我"})
|
|
|
|
await store.update("alpha-tool", "user-a", {"published": True})
|
|
record = await store.get_any("alpha-tool")
|
|
assert record is not None
|
|
assert record["detail"] == "保留我"
|
|
assert record["published"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_store_ensure_legacy_carries_detail() -> None:
|
|
store = MemorySkillStore()
|
|
record = await store.ensure_legacy(
|
|
{"name": "builtin-skill", "owner_user_id": None, "published": True, "detail": "内置技能说明"}
|
|
)
|
|
assert record["detail"] == "内置技能说明"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Response mapping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_skill_response_detail_defaults_to_none() -> None:
|
|
resp = SkillResponse(name="x", description="d", category=SkillCategory.CUSTOM)
|
|
assert resp.detail is None
|
|
|
|
|
|
def test_skill_detail_request_defaults_to_empty_string() -> None:
|
|
assert SkillDetailRequest().detail == ""
|
|
assert SkillDetailRequest(detail="hello").detail == "hello"
|
|
|
|
|
|
def test_skill_to_response_surfaces_detail_from_ownership(isolated_paths) -> None:
|
|
skill = SimpleNamespace(
|
|
name="alpha-tool",
|
|
description="desc",
|
|
license=None,
|
|
category=SkillCategory.CUSTOM,
|
|
enabled=True,
|
|
)
|
|
ownership = {
|
|
"name": "alpha-tool",
|
|
"owner_user_id": "user-a",
|
|
"published": False,
|
|
"builtin": False,
|
|
"detail": "这是技能的详细介绍",
|
|
}
|
|
resp = _skill_to_response(skill, ownership)
|
|
assert resp.detail == "这是技能的详细介绍"
|
|
assert resp.owner_user_id == "user-a"
|
|
|
|
|
|
def test_skill_to_response_detail_none_without_ownership(isolated_paths) -> None:
|
|
skill = SimpleNamespace(
|
|
name="builtin-skill",
|
|
description="desc",
|
|
license=None,
|
|
category=SkillCategory.PUBLIC,
|
|
enabled=True,
|
|
)
|
|
resp = _skill_to_response(skill, None)
|
|
assert resp.detail is None
|
|
assert resp.builtin is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SKILL.md source visibility — admin / publisher only
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_admin_can_view_any_skill_source() -> None:
|
|
# Admin sees a built-in skill's source (no owner).
|
|
assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=True) is True
|
|
# ...and another user's custom skill.
|
|
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=True) is True
|
|
|
|
|
|
def test_publisher_can_view_own_custom_skill_source() -> None:
|
|
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-a", "user-a", is_admin=False) is True
|
|
|
|
|
|
def test_non_owner_cannot_view_custom_skill_source() -> None:
|
|
# A different user — even if the skill is published to them — cannot read source.
|
|
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=False) is False
|
|
|
|
|
|
def test_non_admin_cannot_view_builtin_skill_source() -> None:
|
|
# Built-in / public skills have no publisher, so non-admins never see source.
|
|
assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=False) is False
|