Files

554 lines
25 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import json
import logging
from types import SimpleNamespace
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.gateway.routers import html_runtime
from deerflow.agents.lead_agent import agent as agent_module
from deerflow.agents.lead_agent import prompt as prompt_module
from deerflow.config.paths import Paths
from deerflow.config.skills_config import EsQueryRoutingConfig, SkillsConfig
from deerflow.tools.builtins import page_report_template_tool as page_tool
from deerflow.tools.builtins.page_report_template_tool import (
_brief_layout_mode,
_brief_page_count,
_brief_visualizations,
_choose_page_profile,
_quality_report,
_requested_layout_mode,
_requested_page_count,
_required_visualizations,
)
def _app_config(*, enabled: bool, prompt: str = "自定义 ES 路由规则") -> SimpleNamespace:
return SimpleNamespace(
skills=SkillsConfig(
es_query_routing=EsQueryRoutingConfig(enabled=enabled, prompt=prompt),
),
html_runtime=SimpleNamespace(
base_url="http://localhost:8001/api/public/html-runtime"
),
)
def test_es_query_routing_config_is_disabled_by_default() -> None:
config = SkillsConfig()
assert config.es_query_routing.enabled is False
assert "es_query" in config.es_query_routing.prompt
def test_es_query_routing_section_honors_switch_prompt_and_allowlist() -> None:
enabled = _app_config(enabled=True, prompt=" 自定义 ES 路由规则 ")
assert prompt_module._build_es_query_routing_section(app_config=enabled) == (
'<skill_routing skill="es_query">\n自定义 ES 路由规则\n</skill_routing>'
)
assert (
prompt_module._build_es_query_routing_section(
app_config=enabled,
available_skills={"other_skill"},
)
== ""
)
assert "自定义 ES 路由规则" in prompt_module._build_es_query_routing_section(
app_config=enabled,
available_skills={"es_query"},
)
assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=False)) == ""
assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=True, prompt=" ")) == ""
def test_es_query_routing_suppresses_explicitly_offline_skill(tmp_path, monkeypatch) -> None:
extensions_path = tmp_path / "extensions_config.json"
extensions_path.write_text(
json.dumps({"mcpServers": {}, "skills": {"es_query": {"enabled": False}}}),
encoding="utf-8",
)
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=True)) == ""
extensions_path.write_text(
json.dumps({"mcpServers": {}, "skills": {"es_query": {"enabled": True}}}),
encoding="utf-8",
)
assert '<skill_routing skill="es_query">' in prompt_module._build_es_query_routing_section(
app_config=_app_config(enabled=True),
)
def test_es_query_routing_is_injected_into_lead_and_allowed_custom_agent_prompts(monkeypatch) -> None:
config = _app_config(enabled=True)
monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "<skill_system />")
monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "")
lead_prompt = prompt_module.apply_prompt_template(app_config=config)
allowed_custom_prompt = prompt_module.apply_agent_only_prompt_template(
agent_id="agent-with-es",
agent_name="Agent with ES",
available_skills={"es_query"},
app_config=config,
)
denied_custom_prompt = prompt_module.apply_agent_only_prompt_template(
agent_id="agent-without-es",
agent_name="Agent without ES",
available_skills={"other_skill"},
app_config=config,
)
assert '<skill_routing skill="es_query">' in lead_prompt
assert "自定义 ES 路由规则" in lead_prompt
assert '<skill_routing skill="es_query">' in allowed_custom_prompt
assert '<skill_routing skill="es_query">' not in denied_custom_prompt
def test_es_query_routing_registration_log_confirms_exact_final_prompt(caplog) -> None:
config = _app_config(enabled=True, prompt="自定义 ES 路由规则")
system_prompt = '<skill_routing skill="es_query">\n自定义 ES 路由规则\n</skill_routing>'
with caplog.at_level(logging.INFO, logger=agent_module.__name__):
agent_module._log_es_query_routing_registration(
system_prompt=system_prompt,
app_config=config,
agent_id=None,
available_skills=None,
)
assert "ES query routing prompt registration" in caplog.text
assert "agent_id=default" in caplog.text
assert "enabled=True" in caplog.text
assert "eligible=True" in caplog.text
assert "registered=True" in caplog.text
assert "prompt_chars=11" in caplog.text
assert "prompt_sha256=" in caplog.text
def test_es_query_routing_registration_log_explains_custom_agent_exclusion(caplog) -> None:
config = _app_config(enabled=True, prompt="自定义 ES 路由规则")
with caplog.at_level(logging.INFO, logger=agent_module.__name__):
agent_module._log_es_query_routing_registration(
system_prompt="custom agent prompt",
app_config=config,
agent_id="agent-without-es",
available_skills={"other_skill"},
)
assert "agent_id=agent-without-es" in caplog.text
assert "enabled=True" in caplog.text
assert "eligible=False" in caplog.text
assert "registered=False" in caplog.text
def test_formal_markdown_rules_are_scoped_to_ordinary_qa(monkeypatch) -> None:
config = _app_config(enabled=False)
monkeypatch.setattr(prompt_module, "_ordinary_qa_markdown_format_enabled", lambda: True)
monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "")
ordinary_prompt = prompt_module.apply_prompt_template(app_config=config)
writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True)
notebook_prompt = prompt_module.apply_prompt_template(app_config=config, is_notebook_mode=True)
scheduled_prompt = prompt_module.apply_prompt_template(app_config=config, is_scheduled_run=True)
assert '<ordinary_qa_markdown_format>' in ordinary_prompt
assert "## 一、标题" in ordinary_prompt
assert "### (一) 标题" in ordinary_prompt
assert "#### 1. 标题" in ordinary_prompt
assert "2.1.1" in ordinary_prompt
assert "无序圆点列表" in ordinary_prompt
assert '<ordinary_qa_markdown_format>' not in writing_prompt
assert '<ordinary_qa_markdown_format>' not in notebook_prompt
assert '<ordinary_qa_markdown_format>' not in scheduled_prompt
def test_formal_markdown_rules_are_disabled_when_switch_is_off(monkeypatch) -> None:
config = _app_config(enabled=False)
monkeypatch.setattr(prompt_module, "_ordinary_qa_markdown_format_enabled", lambda: False)
monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "")
ordinary_prompt = prompt_module.apply_prompt_template(app_config=config)
assert '<ordinary_qa_markdown_format>' not in ordinary_prompt
def test_message_visualization_rules_are_offline_safe_and_scoped_to_ordinary_qa(monkeypatch) -> None:
config = _app_config(enabled=False)
monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "")
ordinary_prompt = prompt_module.apply_prompt_template(app_config=config)
writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True)
notebook_prompt = prompt_module.apply_prompt_template(app_config=config, is_notebook_mode=True)
scheduled_prompt = prompt_module.apply_prompt_template(app_config=config, is_scheduled_run=True)
assert "仅当用户明确要求在普通问答中展示地图" in ordinary_prompt
assert "map-json" in ordinary_prompt
assert "不得为了展示地图创建或写入 HTML 文件" in ordinary_prompt
assert "用户只要求画流程图时,必须直接在当前消息中输出 Mermaid" in ordinary_prompt
assert "用户只要求画图表时,必须沿用原有的消息内图表回答方式" in ordinary_prompt
assert "analysis-mindmap" in ordinary_prompt
assert "mindmap-json" in ordinary_prompt
assert "普通分析、总结、提纲、结构化 JSON、流程图或图表都不得自行升级为思维导图" in ordinary_prompt
assert "禁止读取或调用 `page-report-generation`" in ordinary_prompt
assert "默认使用消息内展示" in ordinary_prompt
assert "任务就必须作为文件交付" in ordinary_prompt
assert "调用 `present_files` 打开制品预览" in ordinary_prompt
assert "禁止在消息正文中直接输出完整 HTML 源码" in ordinary_prompt
assert "HTML 页面默认必须是单页面" in ordinary_prompt
assert "报告章节很多、已有 Markdown 很长,都不能作为生成多页面的理由" in ordinary_prompt
assert "flow-json" not in ordinary_prompt
assert "chart-json" not in ordinary_prompt
assert "禁止在地图数据中放入 URL" in ordinary_prompt
assert "map-json" not in writing_prompt
assert "map-json" not in notebook_prompt
assert "map-json" not in scheduled_prompt
assert "mindmap-json" not in writing_prompt
assert "mindmap-json" not in notebook_prompt
assert "mindmap-json" not in scheduled_prompt
def test_html_artifact_visualizations_use_local_declarative_contracts(monkeypatch) -> None:
config = _app_config(enabled=False)
monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "")
monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "")
ordinary_prompt = prompt_module.apply_prompt_template(app_config=config)
writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True)
assert '<html_artifact_visualization>' in ordinary_prompt
assert "只有通过 `<ordinary_qa_visualization>` 中的 HTML 页面意图门禁后" in ordinary_prompt
assert "不得只在聊天消息中粘贴 HTML 代码" in ordinary_prompt
assert "不得仅因为用户要求画地图" in ordinary_prompt
assert "普通地图请求必须在消息列表中使用 `map-json`" in ordinary_prompt
assert 'data-df-flow="process-flow"' in ordinary_prompt
assert "流程图不要使用 Mermaid" in ordinary_prompt
assert "AntV G6" in ordinary_prompt
assert 'data-df-chart="sales-chart"' in ordinary_prompt
assert 'data-df-geo="area-map"' in ordinary_prompt
assert '/api/public/html-runtime/g6.min.js' in ordinary_prompt
assert '/api/public/html-runtime/echarts.min.js' in ordinary_prompt
assert '/api/public/html-runtime/deerflow-runtime.js' in ordinary_prompt
assert "不要在 HTML 中省略这些引用" in ordinary_prompt
assert '"map":"china-provinces"' in ordinary_prompt
assert "禁止 Leaflet、`data-df-map`" in ordinary_prompt
assert "地图瓦片 URL" in ordinary_prompt
assert "deerflow:geo-map-click" in ordinary_prompt
assert "us-states" in ordinary_prompt
assert "japan-prefectures" in ordinary_prompt
assert "taiwan-counties" in ordinary_prompt
assert "详情浮层" in ordinary_prompt
assert "pointer-events:none" in ordinary_prompt
assert "普通问答消息列表的 `map-json` 保持原有前端瓦片渲染" in ordinary_prompt
assert "单页面”表示一个可以自然纵向滚动的 HTML 文档" in ordinary_prompt
assert "禁止固定 1920×1080 舞台" in ordinary_prompt
assert "默认禁止读取 `page-template-library`" in ordinary_prompt
assert "第一项文件操作必须是用 `write_file` 创建" in ordinary_prompt
assert "默认直接用 `write_file` 在 `/mnt/user-data/outputs/`" in ordinary_prompt
assert "data-df-chart` 的内容必须能被 `JSON.parse` 直接解析" in ordinary_prompt
assert "中间地图舞台" in ordinary_prompt
assert "中间至少 65% 的地图画面保持无遮挡" in ordinary_prompt
assert '<html_artifact_visualization>' not in writing_prompt
def test_html_runtime_exposes_only_local_geojson_maps(monkeypatch) -> None:
monkeypatch.setattr(
html_runtime,
"get_app_config",
lambda: SimpleNamespace(
html_runtime=SimpleNamespace(
base_url="http://gateway.internal/api/public/html-runtime"
)
),
)
app = FastAPI()
app.include_router(html_runtime.router)
client = TestClient(app)
runtime = client.get("/api/public/html-runtime/deerflow-runtime.js")
assert runtime.status_code == 200
assert "data-df-geo" in runtime.text
assert "df-geo-popup" in runtime.text
assert "deerflow:geo-map-click" in runtime.text
assert "mapTileUrl" not in runtime.text
assert client.get("/api/public/html-runtime/leaflet.js").status_code == 404
for map_id, filename, minimum_features in (
("china-provinces", "china-provinces.geojson", 30),
("us-states", "us-states.geojson", 51),
("japan-prefectures", "japan-prefectures.geojson", 40),
("taiwan-counties", "taiwan-counties.geojson", 20),
):
assert f"/maps/{filename}" in runtime.text
geojson = client.get(f"/api/public/html-runtime/maps/{filename}")
assert geojson.status_code == 200
assert "application/geo+json" in geojson.headers["content-type"]
assert geojson.json()["type"] == "FeatureCollection"
assert len(geojson.json()["features"]) >= minimum_features
def _report_html(body: str, scripts: str = "") -> str:
return (
"<!doctype html><html><head><title>测试报告</title>"
f"{scripts}</head><body><p>{'报告正文' * 180}</p>{body}</body></html>"
)
def test_page_brief_preserves_markdown_visualization_requirements() -> None:
assert _required_visualizations(
[],
"Markdown 报告包含 Mermaid 流程图、销售额折线图和城市坐标点地图。",
) == {"flowchart", "chart", "map"}
def test_page_quality_rejects_manual_flow_and_accepts_g6_contract() -> None:
manual = _quality_report(
_report_html('<h2>发展流程图</h2><div class="flow-step">技术突破</div>'),
data_mode="provided",
expected_features={"flowchart"},
)
compliant = _quality_report(
_report_html(
'<div id="flow" class="df-flow" style="height:420px"></div>'
'<script type="application/json" data-df-flow="flow">'
'{"nodes":[],"edges":[]}</script>',
'<script src="g6.min.js"></script><script src="deerflow-runtime.js"></script>',
),
data_mode="provided",
expected_features={"flowchart"},
)
assert manual["ok"] is False
assert any("流程图未使用平台绘图契约" in error for error in manual["errors"])
assert compliant["ok"] is True
def test_page_quality_checks_chart_and_map_runtime_contracts() -> None:
report = _quality_report(
_report_html("<h2>销售图表与城市地图</h2>"),
data_mode="provided",
expected_features={"chart", "map"},
)
assert report["ok"] is False
assert any("图表未使用平台绘图契约" in error for error in report["errors"])
assert any("地图未使用平台绘图契约" in error for error in report["errors"])
def test_page_quality_rejects_non_json_echarts_callbacks() -> None:
scripts = '<script src="echarts.min.js"></script><script src="deerflow-runtime.js"></script>'
report = _quality_report(
_report_html(
'<div id="orders" class="df-chart" style="height:360px"></div>'
'<script type="application/json" data-df-chart="orders">'
'{"series":[{"type":"bar","itemStyle":{"color":function(p){return "#0ea5e9";}},"data":[12]}]}'
'</script>',
scripts,
),
data_mode="provided",
expected_features={"chart"},
)
assert report["ok"] is False
assert any("data-df-chart 不是严格 JSON" in error for error in report["errors"])
def test_page_quality_accepts_strict_json_echarts_options() -> None:
scripts = '<script src="echarts.min.js"></script><script src="deerflow-runtime.js"></script>'
payload = json.dumps(
{
"tooltip": {"formatter": "{b}: {c}"},
"series": [
{
"type": "bar",
"data": [{"value": 12, "itemStyle": {"color": "#0ea5e9"}}],
}
],
}
)
report = _quality_report(
_report_html(
'<div id="orders" class="df-chart" style="height:360px"></div>'
f'<script type="application/json" data-df-chart="orders">{payload}</script>',
scripts,
),
data_mode="provided",
expected_features={"chart"},
)
assert report["ok"] is True
def test_page_quality_accepts_each_catalog_geojson_map() -> None:
scripts = '<script src="echarts.min.js"></script><script src="deerflow-runtime.js"></script>'
for map_name in ("china-provinces", "us-states", "japan-prefectures", "taiwan-counties"):
report = _quality_report(
_report_html(
'<div id="area-map" class="df-geo-map" style="height:560px"></div>'
'<script type="application/json" data-df-geo="area-map">'
+ json.dumps({"map": map_name, "regions": [], "markers": [], "lines": []})
+ "</script>",
scripts,
),
data_mode="provided",
expected_features={"map"},
)
assert report["ok"] is True
def test_page_validation_restores_prepare_brief_requirements(tmp_path) -> None:
actual = tmp_path / "人工智能发展报告.html"
actual.write_text(_report_html("<h2>报告页面</h2>"), encoding="utf-8")
brief = {
"requested_page_count": 1,
"required_features": [],
"required_visualizations": ["flowchart"],
"user_material_brief": "原 Markdown 中包含 Mermaid 流程图。",
}
(tmp_path / "人工智能发展报告--page-brief.json").write_text(
json.dumps(brief, ensure_ascii=False),
encoding="utf-8",
)
assert _brief_visualizations(actual, tmp_path) == {"flowchart"}
assert _brief_page_count(actual, tmp_path) == 1
def test_page_template_defaults_to_single_page_and_overrides_multi_template() -> None:
requested_page_count = _requested_page_count(None)
profile = _choose_page_profile(
theme="dark",
content_density="text-heavy",
required_features=["text-panels"],
requested_page_count=requested_page_count,
selection_key="default-single-page",
exclude_template_ids=[],
template_id="dark-charts-portal",
)
assert requested_page_count == 1
assert profile.page_count == 1
assert profile.id == "dark-document"
def test_page_template_allows_multi_page_only_with_explicit_count() -> None:
profile = _choose_page_profile(
theme="light",
content_density="text-heavy",
required_features=["text-panels"],
requested_page_count=3,
selection_key="explicit-multi-page",
exclude_template_ids=[],
template_id="light-airy",
)
assert profile.page_count > 1
def test_page_quality_rejects_multi_view_html_when_single_page_is_required() -> None:
report = _quality_report(
_report_html(
'<nav><button data-page="one">第一页</button><button data-page="two">第二页</button></nav>'
'<section class="view active">第一页</section><section class="view">第二页</section>'
),
data_mode="provided",
expected_features=set(),
expected_page_count=1,
)
assert report["ok"] is False
assert any("HTML 必须保持单页面" in error for error in report["errors"])
def test_page_layout_defaults_to_document_unless_large_screen_is_explicit() -> None:
assert _requested_layout_mode("把这份 Markdown 画成一个 HTML 页面") == "document"
assert _requested_layout_mode("制作 1920×1080 单屏驾驶舱大屏") == "screen"
def test_document_quality_rejects_fixed_canvas_and_undersized_long_flow() -> None:
nodes = [{"id": str(index), "label": f"节点{index}"} for index in range(9)]
payload = json.dumps({"direction": "LR", "nodes": nodes, "edges": []}, ensure_ascii=False)
report = _quality_report(
_report_html(
'<style>html, body { width:100%; height:100%; overflow:hidden }'
'.stage { width:1920px; height:1080px }</style>'
'<div class="stage"><div id="flow" class="df-flow" style="height:220px"></div></div>'
f'<script type="application/json" data-df-flow="flow">{payload}</script>',
'<script src="g6.min.js"></script><script src="deerflow-runtime.js"></script>',
),
data_mode="provided",
expected_features={"flowchart"},
expected_layout_mode="document",
)
assert report["ok"] is False
assert any("overflow:hidden" in error for error in report["errors"])
assert any("1920×1080" in error for error in report["errors"])
assert any("至少需要 360px" in error for error in report["errors"])
assert any("改用 TB" in error for error in report["errors"])
def test_prepare_places_live_editable_html_in_outputs(tmp_path, monkeypatch) -> None:
paths = Paths(tmp_path / ".deer-flow")
monkeypatch.setattr(page_tool, "get_paths", lambda: paths)
monkeypatch.setattr(page_tool, "resolve_path_user_id", lambda _thread_id: "tester")
runtime = SimpleNamespace(
context={"thread_id": "page-preview-test"},
config={"configurable": {"thread_id": "page-preview-test"}},
)
prepared = json.loads(
page_tool._prepare(
runtime,
theme="light",
content_density="text-heavy",
required_features=["flowchart"],
preferred_page_count=1,
selection_key="preview",
exclude_template_ids=[],
template_id="light-document",
output_name="人工智能发展报告.html",
page_brief="把 Markdown 报告生成普通单页 HTML,包含流程图。",
)
)
assert prepared["working_copy"] == "/mnt/user-data/outputs/人工智能发展报告.html"
assert prepared["working_copy"] == prepared["final_output"]
actual = paths.resolve_virtual_path(
"page-preview-test", prepared["working_copy"], user_id="tester"
)
brief_path = paths.resolve_virtual_path(
"page-preview-test", prepared["brief_path"], user_id="tester"
)
assert actual.is_file()
assert _brief_layout_mode(actual, brief_path.parent) == "document"