618 lines
20 KiB
Python
618 lines
20 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway.routers import system_settings as system_settings_router
|
|
from deerflow.config.system_settings import (
|
|
AIWritingGuideCard,
|
|
AIWritingPageDescription,
|
|
GeneralQaPageDescription,
|
|
GeneralQaPageDescriptions,
|
|
PageDescriptionsSettings,
|
|
PageHeaderDescription,
|
|
SystemSettings,
|
|
)
|
|
|
|
|
|
def test_strategy_project_demo_is_hidden_by_default() -> None:
|
|
assert SystemSettings().strategy_projects.taiwan_strait_demo_enabled is False
|
|
|
|
|
|
def test_assistant_knowledge_is_hidden_by_default() -> None:
|
|
assert SystemSettings().knowledge_base_product_version == "v1"
|
|
|
|
|
|
def test_custom_skill_versioning_is_disabled_by_default() -> None:
|
|
assert SystemSettings().skill_versioning.enabled is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_update_strategy_project_settings(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_strategy_project_settings(
|
|
system_settings_router.StrategyProjectSettingsUpdateRequest(
|
|
taiwan_strait_demo_enabled=True,
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_strategy_project_settings(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"load_system_settings",
|
|
lambda: current,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
|
|
response = await system_settings_router.update_strategy_project_settings(
|
|
system_settings_router.StrategyProjectSettingsUpdateRequest(
|
|
taiwan_strait_demo_enabled=True,
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.taiwan_strait_demo_enabled is True
|
|
assert saved[0].strategy_projects.taiwan_strait_demo_enabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_update_knowledge_base_product_version(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_knowledge_base_product_version(
|
|
system_settings_router.KnowledgeBaseProductVersionUpdateRequest(version="v2"),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_enable_assistant_knowledge_product(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"load_system_settings",
|
|
lambda: current,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
|
|
response = await system_settings_router.update_knowledge_base_product_version(
|
|
system_settings_router.KnowledgeBaseProductVersionUpdateRequest(version="v2"),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.version == "v2"
|
|
assert saved == [current.model_copy(update={"knowledge_base_product_version": "v2"})]
|
|
|
|
|
|
def test_html_map_engine_defaults_to_echarts() -> None:
|
|
assert SystemSettings().html_map.engine == "echarts"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_change_html_map_engine(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_html_map_settings(
|
|
system_settings_router.HtmlMapSettingsUpdateRequest(engine="leaflet"),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_select_html_map_webpage(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(system_settings_router, "load_system_settings", lambda: current)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
|
|
response = await system_settings_router.update_html_map_settings(
|
|
system_settings_router.HtmlMapSettingsUpdateRequest(engine="leaflet"),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.engine == "leaflet"
|
|
assert saved[0].html_map.engine == "leaflet"
|
|
|
|
|
|
def test_retry_keep_partial_defaults_to_enabled() -> None:
|
|
assert SystemSettings().retry_keep_partial.enabled is True
|
|
|
|
|
|
def test_page_descriptions_default_to_product_copy() -> None:
|
|
descriptions = SystemSettings().page_descriptions
|
|
assert descriptions.general_qa.full == GeneralQaPageDescription()
|
|
assert descriptions.general_qa.compact == GeneralQaPageDescription()
|
|
assert descriptions.knowledge_base == PageHeaderDescription()
|
|
assert descriptions.ai_writing == AIWritingPageDescription()
|
|
assert len(descriptions.ai_writing.guide_cards) == 3
|
|
assert descriptions.strategy_projects == PageHeaderDescription()
|
|
assert descriptions.deep_research == PageHeaderDescription()
|
|
assert descriptions.gongwen == PageHeaderDescription()
|
|
assert descriptions.light_app_management == PageHeaderDescription()
|
|
assert descriptions.light_app_list == PageHeaderDescription()
|
|
assert descriptions.agents == PageHeaderDescription()
|
|
assert descriptions.business_chains == PageHeaderDescription()
|
|
assert descriptions.skills == PageHeaderDescription()
|
|
assert descriptions.knowledge_management == PageHeaderDescription()
|
|
|
|
|
|
def test_legacy_general_qa_description_is_shared_by_both_layouts() -> None:
|
|
descriptions = PageDescriptionsSettings.model_validate(
|
|
{
|
|
"general_qa": {
|
|
"welcome_title": " 原有问答欢迎语 ",
|
|
"welcome_description": " 原有说明 ",
|
|
"input_placeholder": " 原有提示 ",
|
|
}
|
|
}
|
|
)
|
|
|
|
expected = GeneralQaPageDescription(
|
|
welcome_title="原有问答欢迎语",
|
|
welcome_description="原有说明",
|
|
input_placeholder="原有提示",
|
|
)
|
|
assert descriptions.general_qa.full == expected
|
|
assert descriptions.general_qa.compact == expected
|
|
|
|
legacy_request = system_settings_router.PageDescriptionsUpdateRequest(
|
|
general_qa=GeneralQaPageDescription(welcome_title="旧版请求"),
|
|
)
|
|
assert legacy_request.general_qa == GeneralQaPageDescriptions(
|
|
full=GeneralQaPageDescription(welcome_title="旧版请求"),
|
|
compact=GeneralQaPageDescription(welcome_title="旧版请求"),
|
|
)
|
|
|
|
|
|
def test_ai_writing_guide_cards_are_padded_to_three_items() -> None:
|
|
descriptions = PageDescriptionsSettings.model_validate(
|
|
{
|
|
"ai_writing": {
|
|
"guide_cards": [
|
|
{"title": " 自定义卡片 ", "description": " 自定义说明 "},
|
|
]
|
|
}
|
|
}
|
|
)
|
|
|
|
assert descriptions.ai_writing.guide_cards == [
|
|
AIWritingGuideCard(title="自定义卡片", description="自定义说明"),
|
|
AIWritingGuideCard(),
|
|
AIWritingGuideCard(),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_update_page_descriptions(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_page_descriptions(
|
|
system_settings_router.PageDescriptionsUpdateRequest(
|
|
general_qa=GeneralQaPageDescriptions(
|
|
full=GeneralQaPageDescription(welcome_title="不可保存"),
|
|
),
|
|
knowledge_base=PageHeaderDescription(),
|
|
ai_writing=AIWritingPageDescription(),
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_page_descriptions(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(system_settings_router, "load_system_settings", lambda: current)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
|
|
response = await system_settings_router.update_page_descriptions(
|
|
system_settings_router.PageDescriptionsUpdateRequest(
|
|
general_qa=GeneralQaPageDescriptions(
|
|
full=GeneralQaPageDescription(
|
|
welcome_title=" 欢迎使用工程开发智能问答助手 ",
|
|
welcome_description=" 请描述您希望完成的工程开发任务。 ",
|
|
input_placeholder=" 输入问题或任务描述 ",
|
|
),
|
|
compact=GeneralQaPageDescription(
|
|
welcome_title=" 简洁问答欢迎语 ",
|
|
),
|
|
),
|
|
knowledge_base=PageHeaderDescription(
|
|
title=" 资料库 ",
|
|
description=" 在这里整理和检索团队资料。 ",
|
|
),
|
|
ai_writing=AIWritingPageDescription(
|
|
title=" 写作工作台 ",
|
|
description=" 将想法逐步发展为文章。 ",
|
|
guide_cards=[
|
|
AIWritingGuideCard(
|
|
title=" 明确写作意图 ",
|
|
description=" 用一句话说明想写什么。 ",
|
|
),
|
|
AIWritingGuideCard(
|
|
title=" 对齐写作参数 ",
|
|
description=" 选择体裁与篇幅。 ",
|
|
),
|
|
AIWritingGuideCard(
|
|
title=" 生成并优化文章 ",
|
|
description=" 查看并完善草稿。 ",
|
|
),
|
|
],
|
|
),
|
|
strategy_projects=PageHeaderDescription(
|
|
title=" 项目工作台 ",
|
|
description=" 集中管理各类项目成果。 ",
|
|
),
|
|
deep_research=PageHeaderDescription(
|
|
title=" 开始研究 ",
|
|
description=" 收集资料并生成研究报告。 ",
|
|
),
|
|
gongwen=PageHeaderDescription(
|
|
title=" 公文工作台 ",
|
|
description=" 对话完成公文起草和校审。 ",
|
|
),
|
|
light_app_management=PageHeaderDescription(
|
|
title=" 轻应用管理 ",
|
|
description=" 登记和维护轻应用。 ",
|
|
),
|
|
light_app_list=PageHeaderDescription(
|
|
title=" 轻应用列表 ",
|
|
description=" 快速打开已登记的轻应用。 ",
|
|
),
|
|
agents=PageHeaderDescription(
|
|
title=" 智能体工作台 ",
|
|
description=" 管理可用智能体。 ",
|
|
),
|
|
business_chains=PageHeaderDescription(
|
|
title=" 业务链目录 ",
|
|
description=" 管理可复用的智能体编排。 ",
|
|
),
|
|
skills=PageHeaderDescription(
|
|
title=" 技能工作台 ",
|
|
description=" 浏览和维护技能。 ",
|
|
),
|
|
knowledge_management=PageHeaderDescription(
|
|
title=" 知识服务 ",
|
|
description=" 管理知识库与访问范围。 ",
|
|
),
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.general_qa == GeneralQaPageDescriptions(
|
|
full=GeneralQaPageDescription(
|
|
welcome_title="欢迎使用工程开发智能问答助手",
|
|
welcome_description="请描述您希望完成的工程开发任务。",
|
|
input_placeholder="输入问题或任务描述",
|
|
),
|
|
compact=GeneralQaPageDescription(welcome_title="简洁问答欢迎语"),
|
|
)
|
|
assert response.knowledge_base == PageHeaderDescription(
|
|
title="资料库",
|
|
description="在这里整理和检索团队资料。",
|
|
)
|
|
assert response.ai_writing == AIWritingPageDescription(
|
|
title="写作工作台",
|
|
description="将想法逐步发展为文章。",
|
|
guide_cards=[
|
|
AIWritingGuideCard(
|
|
title="明确写作意图",
|
|
description="用一句话说明想写什么。",
|
|
),
|
|
AIWritingGuideCard(
|
|
title="对齐写作参数",
|
|
description="选择体裁与篇幅。",
|
|
),
|
|
AIWritingGuideCard(
|
|
title="生成并优化文章",
|
|
description="查看并完善草稿。",
|
|
),
|
|
],
|
|
)
|
|
assert response.strategy_projects == PageHeaderDescription(
|
|
title="项目工作台",
|
|
description="集中管理各类项目成果。",
|
|
)
|
|
assert response.deep_research == PageHeaderDescription(
|
|
title="开始研究",
|
|
description="收集资料并生成研究报告。",
|
|
)
|
|
assert response.gongwen == PageHeaderDescription(
|
|
title="公文工作台",
|
|
description="对话完成公文起草和校审。",
|
|
)
|
|
assert response.light_app_management == PageHeaderDescription(
|
|
title="轻应用管理",
|
|
description="登记和维护轻应用。",
|
|
)
|
|
assert response.light_app_list == PageHeaderDescription(
|
|
title="轻应用列表",
|
|
description="快速打开已登记的轻应用。",
|
|
)
|
|
assert response.agents == PageHeaderDescription(
|
|
title="智能体工作台",
|
|
description="管理可用智能体。",
|
|
)
|
|
assert response.business_chains == PageHeaderDescription(
|
|
title="业务链目录",
|
|
description="管理可复用的智能体编排。",
|
|
)
|
|
assert response.skills == PageHeaderDescription(
|
|
title="技能工作台",
|
|
description="浏览和维护技能。",
|
|
)
|
|
assert response.knowledge_management == PageHeaderDescription(
|
|
title="知识服务",
|
|
description="管理知识库与访问范围。",
|
|
)
|
|
assert saved[0].page_descriptions.general_qa == response.general_qa
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_keep_partial_get_reflects_saved_settings(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings().model_copy(
|
|
update={"retry_keep_partial": system_settings_router.RetryKeepPartialSettings(enabled=False)}
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"load_system_settings",
|
|
lambda: current,
|
|
)
|
|
|
|
response = await system_settings_router.get_retry_keep_partial_setting()
|
|
assert response.enabled is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_update_retry_keep_partial(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_retry_keep_partial_setting(
|
|
system_settings_router.RetryKeepPartialUpdateRequest(enabled=False),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_retry_keep_partial(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"load_system_settings",
|
|
lambda: current,
|
|
)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
|
|
response = await system_settings_router.update_retry_keep_partial_setting(
|
|
system_settings_router.RetryKeepPartialUpdateRequest(enabled=False),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.enabled is False
|
|
assert saved[0].retry_keep_partial.enabled is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_versioning_get_reflects_saved_setting(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
current = SystemSettings().model_copy(
|
|
update={
|
|
"skill_versioning": system_settings_router.SkillVersioningSettings(
|
|
enabled=True
|
|
)
|
|
}
|
|
)
|
|
monkeypatch.setattr(system_settings_router, "load_system_settings", lambda: current)
|
|
|
|
response = await system_settings_router.get_skill_versioning()
|
|
|
|
assert response.enabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_cannot_update_skill_versioning(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def ordinary_user(_request):
|
|
return SimpleNamespace(system_role="user")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
ordinary_user,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await system_settings_router.update_skill_versioning(
|
|
system_settings_router.SkillVersioningUpdateRequest(enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert error.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_enable_skill_versioning_and_capture_baselines(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from deerflow.skills import versioning
|
|
|
|
current = SystemSettings()
|
|
saved: list[SystemSettings] = []
|
|
reconciled: list[str | None] = []
|
|
|
|
async def admin_user(_request):
|
|
return SimpleNamespace(system_role="admin", id="admin-1")
|
|
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"get_optional_user_from_request",
|
|
admin_user,
|
|
)
|
|
monkeypatch.setattr(system_settings_router, "load_system_settings", lambda: current)
|
|
monkeypatch.setattr(
|
|
system_settings_router,
|
|
"save_system_settings",
|
|
lambda settings: saved.append(settings) or settings,
|
|
)
|
|
monkeypatch.setattr(
|
|
versioning,
|
|
"reconcile_custom_skill_versions",
|
|
lambda *, author=None: reconciled.append(author) or 0,
|
|
)
|
|
|
|
response = await system_settings_router.update_skill_versioning(
|
|
system_settings_router.SkillVersioningUpdateRequest(enabled=True),
|
|
SimpleNamespace(),
|
|
)
|
|
|
|
assert response.enabled is True
|
|
assert saved[0].skill_versioning.enabled is True
|
|
assert reconciled == ["admin-1"]
|