Files
deerflow-code/offline-backend-20260512/backend/tests/test_skill_detail.py
吉鑫升andClaude Opus 4.7 e0c1cb6581 feat: 定时任务管理 + 会话只读分享
定时任务:
- 管理员可查看/编辑/删除/触发全部用户的定时任务(新增 /api/scheduled-tasks/admin/* 接口与管理页)
- 卡死任务恢复:单次执行超 30 分钟超时重试(最多 3 次),耗尽后标记 failed;看门狗回收服务重启遗留的「执行中」任务

会话分享:
- 新增「只读链接」分享模式,生成公开免登录的只读对话快照页(/share/:code)
- 公开接口 /api/public/shares/* 提供对话快照与图片附件

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-18 15:39:40 +08:00

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