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) == (
'\n自定义 ES 路由规则\n'
)
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 '' 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: "")
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 '' in lead_prompt
assert "自定义 ES 路由规则" in lead_prompt
assert '' in allowed_custom_prompt
assert '' 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 = '\n自定义 ES 路由规则\n'
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 '' 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 '' not in writing_prompt
assert '' not in notebook_prompt
assert '' 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 '' 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 '' in ordinary_prompt
assert "只有通过 `` 中的 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 '' 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 (
"测试报告"
f"{scripts}{'报告正文' * 180}
{body}"
)
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('发展流程图
技术突破
'),
data_mode="provided",
expected_features={"flowchart"},
)
compliant = _quality_report(
_report_html(
''
'',
'',
),
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("销售图表与城市地图
"),
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 = ''
report = _quality_report(
_report_html(
''
'',
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 = ''
payload = json.dumps(
{
"tooltip": {"formatter": "{b}: {c}"},
"series": [
{
"type": "bar",
"data": [{"value": 12, "itemStyle": {"color": "#0ea5e9"}}],
}
],
}
)
report = _quality_report(
_report_html(
''
f'',
scripts,
),
data_mode="provided",
expected_features={"chart"},
)
assert report["ok"] is True
def test_page_quality_accepts_each_catalog_geojson_map() -> None:
scripts = ''
for map_name in ("china-provinces", "us-states", "japan-prefectures", "taiwan-counties"):
report = _quality_report(
_report_html(
''
'",
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("报告页面
"), 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(
''
''
),
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(
''
''
f'',
'',
),
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"