470 lines
18 KiB
Python
470 lines
18 KiB
Python
"""Tests for the admin-configurable prompt-prefix toggle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from app.gateway.routers import system_settings as ss_router
|
|
from app.gateway.services import _prepend_prompt_prefix, _prompt_prefix_opted_in
|
|
from deerflow.agents.lead_agent import prompt as prompt_module
|
|
from deerflow.config import system_settings as ss
|
|
from deerflow.sandbox.prompt_policy import get_agent_runtime_policy, get_python_dependency_policy
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_runtime_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
|
|
"""Redirect ``runtime_home()`` to a per-test tmp dir."""
|
|
|
|
def fake_home() -> Path:
|
|
return tmp_path
|
|
|
|
monkeypatch.setattr(ss, "runtime_home", fake_home)
|
|
return tmp_path
|
|
|
|
|
|
def test_load_returns_defaults_when_missing(tmp_runtime_home: Path) -> None:
|
|
settings = ss.load_system_settings()
|
|
assert settings.prompt_prefix.enabled is False
|
|
assert settings.prompt_prefix.prompt_prefix == ""
|
|
assert settings.prompt_prefix.ordinary_qa_markdown_format_enabled is False
|
|
assert settings.prompt_prefix.word_heading_numbering_style == "hybrid_decimal"
|
|
assert settings.citation_display.reference_mode_enabled is False
|
|
assert settings.citation_display.cleanup_words == []
|
|
assert settings.memory.enabled is False
|
|
assert settings.conversation_context.recent_turn_window_enabled is False
|
|
assert settings.conversation_context.max_conversation_rounds == 20
|
|
assert settings.write_file_compatibility.enabled is False
|
|
assert settings.conversation_context.recent_turn_window_enabled is False
|
|
|
|
|
|
def test_save_then_load_roundtrips(tmp_runtime_home: Path) -> None:
|
|
settings = ss.SystemSettings(
|
|
prompt_prefix=ss.PromptPrefixSettings(
|
|
enabled=True,
|
|
switch_label="加入立场前缀",
|
|
prompt_prefix="你需要站在中方角度",
|
|
ordinary_qa_markdown_format_enabled=True,
|
|
word_heading_numbering_style="official_chinese",
|
|
)
|
|
)
|
|
ss.save_system_settings(settings)
|
|
|
|
file_path = tmp_runtime_home / "system_settings.json"
|
|
assert file_path.is_file()
|
|
saved = json.loads(file_path.read_text(encoding="utf-8"))
|
|
assert saved["prompt_prefix"]["prompt_prefix"] == "你需要站在中方角度"
|
|
|
|
reloaded = ss.load_system_settings()
|
|
assert reloaded.prompt_prefix.enabled is True
|
|
assert reloaded.prompt_prefix.switch_label == "加入立场前缀"
|
|
assert reloaded.prompt_prefix.prompt_prefix == "你需要站在中方角度"
|
|
assert reloaded.prompt_prefix.ordinary_qa_markdown_format_enabled is True
|
|
assert reloaded.prompt_prefix.word_heading_numbering_style == "official_chinese"
|
|
|
|
|
|
def test_runtime_markdown_format_switch_defaults_off_and_updates_without_restart(
|
|
tmp_runtime_home: Path,
|
|
) -> None:
|
|
assert prompt_module._ordinary_qa_markdown_format_enabled() is False
|
|
|
|
ss.save_system_settings(
|
|
ss.SystemSettings(
|
|
prompt_prefix=ss.PromptPrefixSettings(
|
|
ordinary_qa_markdown_format_enabled=True,
|
|
)
|
|
)
|
|
)
|
|
|
|
assert prompt_module._ordinary_qa_markdown_format_enabled() is True
|
|
|
|
|
|
def _patch_admin(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
admin = SimpleNamespace(
|
|
id="admin1",
|
|
system_role="admin",
|
|
email="admin@example.com",
|
|
)
|
|
|
|
async def _optional(_request):
|
|
return admin
|
|
|
|
monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dependency_delete_prompt_toggle_roundtrip(tmp_runtime_home, monkeypatch):
|
|
_patch_admin(monkeypatch)
|
|
assert get_python_dependency_policy() == ""
|
|
for enabled in (True, False):
|
|
response = await ss_router.update_prompt_prefix(
|
|
ss_router.PromptPrefixUpdateRequest(python_dependency_delete_protection_enabled=enabled),
|
|
SimpleNamespace(),
|
|
)
|
|
assert response.python_dependency_delete_protection_enabled is enabled
|
|
assert ss.load_system_settings().prompt_prefix.python_dependency_delete_protection_enabled is enabled
|
|
assert response.enabled is False # Independent of the first-question prefix switch.
|
|
policy = get_python_dependency_policy()
|
|
assert bool(policy) is enabled
|
|
if enabled:
|
|
assert "允许安装、升级、降级" in policy
|
|
assert "允许包管理器" in policy
|
|
assert "不得执行独立的" in policy
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("field", ["python_dependency_delete_protection_enabled", "image_generation_unavailable_hint_enabled"])
|
|
async def test_nonadmin_cannot_enable_runtime_prompt(tmp_runtime_home, monkeypatch, field):
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(ss_router, "get_optional_user_from_request", ordinary_user)
|
|
with pytest.raises(HTTPException) as error:
|
|
await ss_router.update_prompt_prefix(
|
|
ss_router.PromptPrefixUpdateRequest(**{field: True}),
|
|
SimpleNamespace(),
|
|
)
|
|
assert error.value.status_code == 403
|
|
assert get_python_dependency_policy() == ""
|
|
assert get_agent_runtime_policy() == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_generation_hint_independent_toggle(tmp_runtime_home, monkeypatch):
|
|
_patch_admin(monkeypatch)
|
|
assert get_agent_runtime_policy() == ""
|
|
await ss_router.update_prompt_prefix(
|
|
ss_router.PromptPrefixUpdateRequest(python_dependency_delete_protection_enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
for enabled in (True, False):
|
|
response = await ss_router.update_prompt_prefix(
|
|
ss_router.PromptPrefixUpdateRequest(image_generation_unavailable_hint_enabled=enabled),
|
|
SimpleNamespace(),
|
|
)
|
|
assert response.image_generation_unavailable_hint_enabled is enabled
|
|
assert response.python_dependency_delete_protection_enabled is True
|
|
assert response.enabled is False
|
|
assert ss.load_system_settings().prompt_prefix.image_generation_unavailable_hint_enabled is enabled
|
|
policy = get_agent_runtime_policy()
|
|
assert "<python_dependency_boundary>" in policy
|
|
assert ("<image_generation_unavailable>" in policy) is enabled
|
|
if enabled:
|
|
assert "当前服务暂不支持生成或编辑图片" in policy
|
|
assert "此限制不影响" in policy
|
|
assert "绘图提示词" in policy
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prompt_settings_update_can_enable_markdown_format_without_changing_prefix(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
current = ss.SystemSettings(
|
|
prompt_prefix=ss.PromptPrefixSettings(
|
|
enabled=True,
|
|
switch_label="加入立场前缀",
|
|
prompt_prefix="你需要站在中方角度",
|
|
)
|
|
)
|
|
ss.save_system_settings(current)
|
|
|
|
response = await ss_router.update_prompt_prefix(
|
|
ss_router.PromptPrefixUpdateRequest(
|
|
ordinary_qa_markdown_format_enabled=True,
|
|
word_heading_numbering_style="chapter_section",
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.ordinary_qa_markdown_format_enabled is True
|
|
assert response.word_heading_numbering_style == "chapter_section"
|
|
assert response.enabled is True
|
|
assert response.switch_label == "加入立场前缀"
|
|
assert response.prompt_prefix == "你需要站在中方角度"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_citation_display_update_preserves_unspecified_fields(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
current = ss.SystemSettings(
|
|
citation_display=ss.CitationDisplaySettings(
|
|
cleanup_words=["广告", "免责声明"],
|
|
)
|
|
)
|
|
ss.save_system_settings(current)
|
|
|
|
response = await ss_router.update_citation_display(
|
|
ss_router.CitationDisplayUpdateRequest(reference_mode_enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.reference_mode_enabled is True
|
|
assert response.cleanup_words == ["广告", "免责声明"]
|
|
|
|
payload = json.loads((tmp_runtime_home / "system_settings.json").read_text(encoding="utf-8"))
|
|
assert payload["citation_display"]["reference_mode_enabled"] is True
|
|
assert payload["citation_display"]["cleanup_words"] == ["广告", "免责声明"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_enable_memory_runtime_switch(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
|
|
response = await ss_router.update_memory_runtime_setting(
|
|
ss_router.MemoryRuntimeUpdateRequest(enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.enabled is True
|
|
assert ss.load_system_settings().memory.enabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_enable_recent_turn_context_window(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
|
|
response = await ss_router.update_conversation_context_setting(
|
|
ss_router.ConversationContextUpdateRequest(recent_turn_window_enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.recent_turn_window_enabled is True
|
|
assert response.recent_turns == 3
|
|
assert response.max_conversation_rounds == 20
|
|
assert ss.load_system_settings().conversation_context.recent_turn_window_enabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_conversation_round_limit_without_changing_window(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
ss.save_system_settings(
|
|
ss.SystemSettings(
|
|
conversation_context=ss.ConversationContextSettings(
|
|
recent_turn_window_enabled=True,
|
|
)
|
|
)
|
|
)
|
|
|
|
response = await ss_router.update_conversation_context_setting(
|
|
ss_router.ConversationContextUpdateRequest(max_conversation_rounds=45),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.max_conversation_rounds == 45
|
|
assert response.recent_turn_window_enabled is True
|
|
persisted = ss.load_system_settings().conversation_context
|
|
assert persisted.max_conversation_rounds == 45
|
|
assert persisted.recent_turn_window_enabled is True
|
|
|
|
|
|
@pytest.mark.parametrize("value", [0, 51])
|
|
def test_conversation_round_limit_rejects_out_of_range_values(value: int) -> None:
|
|
with pytest.raises(ValueError):
|
|
ss_router.ConversationContextUpdateRequest(max_conversation_rounds=value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_enable_recent_turn_context_window(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
user = SimpleNamespace(id="user1", system_role="user", email="user@example.com")
|
|
|
|
async def _optional(_request):
|
|
return user
|
|
|
|
monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await ss_router.update_conversation_context_setting(
|
|
ss_router.ConversationContextUpdateRequest(recent_turn_window_enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_toggle_write_file_compatibility_without_restart(
|
|
tmp_runtime_home: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_admin(monkeypatch)
|
|
|
|
enabled = await ss_router.update_write_file_compatibility_setting(
|
|
ss_router.WriteFileCompatibilityUpdateRequest(enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
assert enabled.enabled is True
|
|
assert ss.get_write_file_compatibility_settings().enabled is True
|
|
|
|
disabled = await ss_router.update_write_file_compatibility_setting(
|
|
ss_router.WriteFileCompatibilityUpdateRequest(enabled=False),
|
|
SimpleNamespace(),
|
|
)
|
|
assert disabled.enabled is False
|
|
assert ss.get_write_file_compatibility_settings().enabled is False
|
|
|
|
|
|
def test_prepend_prefix_tags_first_human_without_touching_content() -> None:
|
|
graph_input = {"messages": [HumanMessage(content="评价一下中美贸易")]}
|
|
out = _prepend_prompt_prefix(graph_input, "你需要站在中方角度")
|
|
msg = out["messages"][0]
|
|
assert isinstance(msg, HumanMessage)
|
|
# Content stays clean so the frontend never shows the prefix.
|
|
assert msg.content == "评价一下中美贸易"
|
|
# The prefix rides along in additional_kwargs for the middleware to inject.
|
|
assert msg.additional_kwargs["prompt_prefix"] == "你需要站在中方角度"
|
|
|
|
|
|
def test_prepend_prefix_preserves_block_content() -> None:
|
|
graph_input = {
|
|
"messages": [
|
|
HumanMessage(content=[{"type": "text", "text": "评价一下中美贸易"}])
|
|
]
|
|
}
|
|
out = _prepend_prompt_prefix(graph_input, "你需要站在中方角度")
|
|
msg = out["messages"][0]
|
|
assert msg.content == [{"type": "text", "text": "评价一下中美贸易"}]
|
|
assert msg.additional_kwargs["prompt_prefix"] == "你需要站在中方角度"
|
|
|
|
|
|
def test_prepend_prefix_only_touches_first_human() -> None:
|
|
graph_input = {
|
|
"messages": [
|
|
AIMessage(content="hi"),
|
|
HumanMessage(content="q1"),
|
|
HumanMessage(content="q2"),
|
|
]
|
|
}
|
|
out = _prepend_prompt_prefix(graph_input, "PREFIX")
|
|
# first human is index 1; index 2 must remain untouched
|
|
assert out["messages"][1].additional_kwargs.get("prompt_prefix") == "PREFIX"
|
|
assert out["messages"][1].content == "q1"
|
|
assert out["messages"][2].additional_kwargs.get("prompt_prefix") is None
|
|
assert out["messages"][2].content == "q2"
|
|
|
|
|
|
def test_prepend_prefix_no_messages_is_noop() -> None:
|
|
assert _prepend_prompt_prefix({"messages": []}, "X") == {"messages": []}
|
|
assert _prepend_prompt_prefix({}, "X") == {}
|
|
|
|
|
|
def test_prompt_prefix_middleware_injects_prefix_for_model_only() -> None:
|
|
"""The middleware prepends the prefix to the request copy, not to state."""
|
|
from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
|
|
|
|
mw = PromptPrefixMiddleware()
|
|
tagged = HumanMessage(
|
|
content="评价一下中美贸易",
|
|
additional_kwargs={"prompt_prefix": "你需要站在中方角度"},
|
|
)
|
|
patched = PromptPrefixMiddleware._apply([tagged])
|
|
assert patched is not None
|
|
assert patched[0].content == "你需要站在中方角度\n\n评价一下中美贸易"
|
|
# The original message in state is untouched.
|
|
assert tagged.content == "评价一下中美贸易"
|
|
assert mw is not None
|
|
|
|
|
|
def test_prompt_prefix_middleware_noop_without_tag() -> None:
|
|
from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
|
|
|
|
plain = HumanMessage(content="just a question")
|
|
assert PromptPrefixMiddleware._apply([plain]) is None
|
|
|
|
|
|
def test_prompt_prefix_middleware_block_content() -> None:
|
|
from deerflow.agents.middlewares.prompt_prefix_middleware import PromptPrefixMiddleware
|
|
|
|
tagged = HumanMessage(
|
|
content=[{"type": "text", "text": "评价一下中美贸易"}],
|
|
additional_kwargs={"prompt_prefix": "你需要站在中方角度"},
|
|
)
|
|
patched = PromptPrefixMiddleware._apply([tagged])
|
|
assert patched is not None
|
|
assert patched[0].content[0] == {"type": "text", "text": "你需要站在中方角度"}
|
|
assert patched[0].content[1] == {"type": "text", "text": "评价一下中美贸易"}
|
|
|
|
|
|
def test_smart_knowledge_scope_uses_only_latest_real_user_turn() -> None:
|
|
from deerflow.agents.middlewares.prompt_prefix_middleware import (
|
|
SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
|
|
PromptPrefixMiddleware,
|
|
)
|
|
|
|
old_turn = HumanMessage(
|
|
content="这个库讲了什么",
|
|
additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "旧库：政策库"},
|
|
)
|
|
new_turn = HumanMessage(
|
|
content="这个库有哪些新闻",
|
|
additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "新库：新闻库"},
|
|
)
|
|
|
|
patched = PromptPrefixMiddleware._apply(
|
|
[old_turn, AIMessage(content="旧回答"), new_turn]
|
|
)
|
|
|
|
assert patched is not None
|
|
assert patched[0].content == "这个库讲了什么"
|
|
assert patched[2].content == "新库：新闻库\n\n这个库有哪些新闻"
|
|
assert "旧库：政策库" not in str(patched[2].content)
|
|
|
|
|
|
def test_classic_latest_turn_does_not_revive_historical_smart_scope() -> None:
|
|
from deerflow.agents.middlewares.prompt_prefix_middleware import (
|
|
SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
|
|
PromptPrefixMiddleware,
|
|
)
|
|
|
|
old_smart_turn = HumanMessage(
|
|
content="这个库讲了什么",
|
|
additional_kwargs={SMART_KNOWLEDGE_SCOPE_PROMPT_KEY: "旧智能库范围"},
|
|
)
|
|
latest_classic_turn = HumanMessage(content="请继续回答")
|
|
|
|
patched = PromptPrefixMiddleware._apply(
|
|
[old_smart_turn, AIMessage(content="旧回答"), latest_classic_turn]
|
|
)
|
|
|
|
assert patched is None
|
|
assert old_smart_turn.content == "这个库讲了什么"
|
|
assert latest_classic_turn.content == "请继续回答"
|
|
|
|
|
|
def test_opted_in_via_context() -> None:
|
|
body = SimpleNamespace(context={"prompt_prefix_enabled": True}, metadata=None)
|
|
assert _prompt_prefix_opted_in(body) is True
|
|
|
|
|
|
def test_opted_in_via_metadata() -> None:
|
|
body = SimpleNamespace(context=None, metadata={"prompt_prefix_enabled": True})
|
|
assert _prompt_prefix_opted_in(body) is True
|
|
|
|
|
|
def test_not_opted_in_by_default() -> None:
|
|
body = SimpleNamespace(context=None, metadata=None)
|
|
assert _prompt_prefix_opted_in(body) is False
|
|
body2 = SimpleNamespace(context={"mode": "ultra"}, metadata={})
|
|
assert _prompt_prefix_opted_in(body2) is False
|