import json
import logging
from types import SimpleNamespace
import pytest
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.html_runtime.html_postprocess import maybe_fix_generated_html
from deerflow.tools.builtins import page_report_template_tool as page_tool
from deerflow.tools.builtins.page_report_template_tool import (
_attach_repair_plan,
_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, "_word_heading_numbering_style", lambda: "hybrid_decimal")
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 "### 1.1 标题" in ordinary_prompt
assert "#### 1.1.1 标题" in ordinary_prompt
assert "##### 1.1.1.1 标题" in ordinary_prompt
assert "###### 1.1.1.1.1 标题" in ordinary_prompt
assert "2.1" in ordinary_prompt
assert "无序圆点列表" in ordinary_prompt
assert '' not in writing_prompt
assert '' not in notebook_prompt
assert '' not in scheduled_prompt
@pytest.mark.parametrize(
("style", "expected"),
[
("hybrid_decimal", "### 1.1 标题"),
("official_chinese", "### (一) 标题"),
("decimal", "## 1 标题"),
("chapter_section", "### 第一节 标题"),
("none", "### 二级标题"),
],
)
def test_formal_markdown_rules_follow_word_numbering_style(monkeypatch, style: str, expected: str) -> None:
monkeypatch.setattr(prompt_module, "_word_heading_numbering_style", lambda: style)
section = prompt_module._build_ordinary_qa_markdown_format_section()
assert expected in section
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 "不要用 `from`/`to`/`coords`" in ordinary_prompt
assert '"coordinates":[[116.4,39.9],[121.47,31.23]]' in ordinary_prompt
assert '"color":"red"' in ordinary_prompt
assert '"arrow":true' in ordinary_prompt
assert '"flow":true' in ordinary_prompt
assert "只有用户**明确要求**连线" in ordinary_prompt
assert "`lines` 必须为 `[]`" 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: "")
monkeypatch.setattr("deerflow.html_runtime.map_engine.html_map_engine", lambda: "echarts")
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 "流程图所在 `.page` 初始隐藏时,运行时会等容器可见后再初始化" 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 '/api/public/html-runtime/bootstrap.min.css' in ordinary_prompt
assert '/api/public/html-runtime/bootstrap.bundle.min.js' in ordinary_prompt
assert '/api/public/html-runtime/alpine.min.js' in ordinary_prompt
assert "轻量交互门户" in ordinary_prompt
assert "/api/public/html-images/" in ordinary_prompt
assert "background-image" 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 "禁止读取或执行 `web-app`、Vite、React 页面脚手架" in ordinary_prompt
assert 'data-df-theme="dark"' in ordinary_prompt
assert "description` 必须是纯文本" in ordinary_prompt
assert 'window.location.hash="#page-stations"' in ordinary_prompt
assert "首写即正确" in ordinary_prompt
assert "局部补丁" in ordinary_prompt
assert "禁止修一个错误就立刻校验" in ordinary_prompt
assert "repair_checklist" in ordinary_prompt
assert "Bootstrap 组件可能覆盖它" in ordinary_prompt
assert "中间地图舞台" in ordinary_prompt
assert "中间至少 65% 的地图画面保持无遮挡" in ordinary_prompt
assert '' not in writing_prompt
def test_html_map_webpage_setting_switches_the_map_contract(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: "")
monkeypatch.setattr("deerflow.html_runtime.map_engine.html_map_engine", lambda: "leaflet")
ordinary_prompt = prompt_module.apply_prompt_template(app_config=config)
assert 'data-df-map="area-map"' in ordinary_prompt
assert "/api/public/html-runtime/leaflet.js" in ordinary_prompt
assert "/api/public/html-runtime/leaflet.css" in ordinary_prompt
assert "地图网页" in ordinary_prompt
assert "禁止 `data-df-geo`" in ordinary_prompt
assert "禁止 Leaflet、`data-df-map`" not in ordinary_prompt
def test_markdown_file_deliverable_is_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)
assert "" in ordinary_prompt
assert "写 md 报告" in ordinary_prompt
assert "write_file" in ordinary_prompt
assert "" not in writing_prompt
assert "" not in notebook_prompt
def test_html_runtime_exposes_only_local_geojson_maps(monkeypatch) -> None:
monkeypatch.setattr(
html_runtime,
"html_runtime_base_url",
lambda: "http://gateway.internal/api/public/html-runtime",
)
monkeypatch.setattr(
html_runtime,
"html_map_tile_url",
lambda: "https://mt0.google.com/vt/lyrs=m&hl=zh-CN&x={x}&y={y}&z={z}",
)
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" in runtime.text
assert "https://mt0.google.com/vt/lyrs=m" in runtime.text
assert "data-df-map" in runtime.text
assert client.get("/api/public/html-runtime/leaflet.js").status_code == 200
assert client.get("/api/public/html-runtime/leaflet.css").status_code == 200
for filename, media_type in (
("bootstrap.min.css", "text/css"),
("bootstrap.bundle.min.js", "application/javascript"),
("alpine.min.js", "application/javascript"),
("echarts.min.js", "application/javascript"),
("g6.min.js", "application/javascript"),
):
asset = client.get(f"/api/public/html-runtime/{filename}")
assert asset.status_code == 200
assert media_type in asset.headers["content-type"]
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_accepts_g6_flow_on_hidden_logical_page() -> None:
scripts = (
''
''
''
''
)
report = _quality_report(
_report_html(
''
''
''
''
'",
scripts,
),
data_mode="provided",
expected_features={"flowchart"},
expected_page_count=2,
)
assert report["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_quality_rejects_html_inside_geo_map_descriptions() -> None:
scripts = ''
payload = json.dumps(
{
"map": "china-provinces",
"regions": [],
"markers": [
{
"name": "北京",
"coordinate": [116.4, 39.9],
"description": '查看全部',
}
],
"lines": [],
},
ensure_ascii=False,
)
report = _quality_report(
_report_html(
''
f'',
scripts,
),
data_mode="provided",
expected_features={"map"},
)
assert report["ok"] is False
assert any("description 必须是纯文本" in error for error in report["errors"])
def test_page_quality_requires_readable_dark_chart_text() -> None:
scripts = ''
unreadable_payload = json.dumps(
{
"xAxis": {"type": "category"},
"yAxis": {"type": "value", "name": "站点数"},
"series": [{"type": "bar", "data": [12]}],
}
)
unreadable_html = _report_html(
''
''
f'',
scripts,
).replace("", '')
unreadable = _quality_report(
unreadable_html,
data_mode="provided",
expected_features={"chart"},
)
readable_payload = json.dumps(
{
"textStyle": {"color": "#e0e8f0"},
"xAxis": {"type": "category", "axisLabel": {"color": "#b8c8d8"}},
"yAxis": {
"type": "value",
"name": "站点数",
"axisLabel": {"color": "#b8c8d8"},
"nameTextStyle": {"color": "#e0e8f0"},
},
"series": [{"type": "bar", "data": [12]}],
}
)
readable_html = _report_html(
''
''
f'',
scripts,
).replace("", '')
readable = _quality_report(
readable_html,
data_mode="provided",
expected_features={"chart"},
)
assert unreadable["ok"] is False
assert any("textStyle.color" in error for error in unreadable["errors"])
assert any("nameTextStyle" in error for error in unreadable["errors"])
assert readable["ok"] is True
def test_dark_html_auto_fixes_and_validates_bootstrap_component_text() -> None:
html = _report_html(
''
'卡片正文
'
''
''
).replace("", '')
before = _quality_report(html, data_mode="provided", expected_features=set())
fixed = maybe_fix_generated_html("/mnt/user-data/outputs/portal.html", html)
after = _quality_report(fixed, data_mode="provided", expected_features=set())
assert before["ok"] is False
assert any("Bootstrap `.card`" in error for error in before["errors"])
assert any("明细表" in error for error in before["errors"])
assert any("Bootstrap 弹窗" in error for error in before["errors"])
assert "data-df-dark-contrast-fix" in fixed
assert "--bs-body-color" in fixed
assert ".data-table td" in fixed
assert after["ok"] is True
assert maybe_fix_generated_html("/mnt/user-data/outputs/portal.html", fixed) == fixed
def test_dark_html_auto_fixes_chart_text_in_one_pass() -> None:
scripts = ''
payload = json.dumps(
{
"title": {"text": "订单量"},
"legend": {"data": ["订单"]},
"xAxis": {"type": "category"},
"yAxis": {"type": "value", "name": "站点数"},
"series": [{"type": "bar", "data": [12], "label": {"show": True}}],
}
)
html = _report_html(
''
''
f'',
scripts,
).replace("", '')
before = _quality_report(html, data_mode="provided", expected_features={"chart"})
fixed = maybe_fix_generated_html("/mnt/user-data/outputs/report.html", html)
after = _quality_report(fixed, data_mode="provided", expected_features={"chart"})
assert before["ok"] is False
assert len(before["errors"]) >= 3
assert any("textStyle.color" in error for error in before["errors"])
assert any("axisLabel" in error for error in before["errors"])
assert any("nameTextStyle" in error for error in before["errors"])
assert any("legend.textStyle.color" in error for error in before["errors"])
assert any("series[0].label" in error for error in before["errors"])
assert after["ok"] is True
assert '"color": "#e8f4ff"' in fixed
assert '"color": "#b8c8d8"' in fixed
assert maybe_fix_generated_html("/mnt/user-data/outputs/report.html", fixed) == fixed
def test_invalid_dark_chart_json_reports_contrast_in_same_error() -> None:
scripts = ''
html = _report_html(
''
''
'',
scripts,
).replace("", '')
report = _quality_report(html, data_mode="provided", expected_features={"chart"})
assert report["ok"] is False
assert any("不是严格 JSON" in error and "深色可读色一次性写齐" in error for error in report["errors"])
def test_quality_repair_plan_lists_all_errors_for_one_pass() -> None:
report = _attach_repair_plan(
{
"ok": False,
"errors": ["导航锚点缺失", "流程图容器高度不足"],
"warnings": [],
}
)
assert report["error_count"] == 2
assert report["repair_checklist"] == [
"1. 导航锚点缺失",
"2. 流程图容器高度不足",
]
assert "共 2 个错误,必须在同一次回复里全部修完" in report["repair_strategy"]
assert "禁止修一个就立刻 validate" in report["repair_strategy"]
assert "1. 导航锚点缺失" in report["repair_strategy"]
assert "2. 流程图容器高度不足" in report["repair_strategy"]
passed = _attach_repair_plan({"ok": True, "errors": []}, ["已自动补齐深色图表的浅色文字"])
assert passed["error_count"] == 0
assert passed["repair_checklist"] == []
assert "校验已自动修复" in passed["repair_strategy"]
assert "直接 present_files" in passed["repair_strategy"]
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_uses_portal_default_for_explicit_multi_page_brief() -> None:
assert _requested_page_count(None, "请画一个多个页面的 HTML 网页") == 3
assert _requested_page_count(4, "请画一个多页面网页") == 4
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_quality_accepts_bootstrap_alpine_multi_page_portal() -> None:
scripts = (
''
''
''
)
report = _quality_report(
_report_html(
''
''
''
''
''
'',
scripts,
),
data_mode="provided",
expected_features=set(),
expected_page_count=3,
)
assert report["ok"] is True
def test_page_quality_rejects_spoofed_or_unmatched_portal_navigation() -> None:
scripts = (
''
''
''
)
report = _quality_report(
_report_html(
''
''
''
''
''
'',
scripts,
),
data_mode="provided",
expected_features=set(),
expected_page_count=2,
)
assert report["ok"] is False
assert any('href="#page-..."' in error for error in report["errors"])
assert any("脚本切页 hash" 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"